### RTU Host/Master Setup (MicroPython) Source: https://micropython-modbus.readthedocs.io/en/latest/EXAMPLES Sets up a MicroPython device as a Modbus RTU host (master) to communicate with a client/slave device. It initializes ModbusRTUMaster with the specified pins and UART ID, and demonstrates reading coil status from a specified slave address and starting address. ```python from umodbus.serial import Serial as ModbusRTUMaster # RTU Host/Master setup # the following definition is for an ESP32 rtu_pins = (25, 26) # (TX, RX) uart_id = 1 # the following definition is for a RP2 # rtu_pins = (Pin(0), Pin(1)) # (TX, RX) # uart_id = 0 # # rtu_pins = (Pin(4), Pin(5)) # (TX, RX) # uart_id = 1 # the following definition is for a pyboard # rtu_pins = (Pin(PB6), Pin(PB7)) # (TX, RX) # uart_id = 1 host = ModbusRTUMaster( pins=rtu_pins, # given as tuple (TX, RX) # baudrate=9600, # optional, default 9600 # data_bits=8, # optional, default 8 # stop_bits=1, # optional, default 1 # parity=None, # optional, default None # ctrl_pin=12, # optional, control DE/RE uart_id=uart_id # optional, default 1, see port specific documentation ) coil_status = host.read_coils(slave_addr=10, starting_addr=123, coil_qty=1) print('Status of coil 123: {}'.format(coil_status)) ``` -------------------------------- ### Install Release Candidate of micropython-modbus using upip Source: https://micropython-modbus.readthedocs.io/en/latest/INSTALLATION Installs a release candidate version of the micropython-modbus library from TestPyPI using the 'upip' package manager. This is for older MicroPython versions. It overwrites the index URLs to target TestPyPI. ```python import upip # overwrite index_urls to only take artifacts from test.pypi.org upip.index_urls = ['https://test.pypi.org/pypi'] upip.install('micropython-modbus') ``` -------------------------------- ### Install Python Dependencies Source: https://micropython-modbus.readthedocs.io/en/latest/SETUP Installs required Python tools and dependencies using pip and a virtual environment. It checks the Python version and activates the virtual environment before installing packages from requirements files. ```bash python --version python3 --version python3 -m venv .venv source .venv/bin/activate pip install -r requirements.txt pip install -r modules/requirements.txt ``` -------------------------------- ### Copy be_helpers module using rshell Source: https://micropython-modbus.readthedocs.io/en/latest/INSTALLATION Manually copies all files and folders from the 'be_helpers' directory to the '/lib/be_helpers' directory on the MicroPython board using 'rshell'. This is for installing additional helper packages required for examples. ```bash mkdir /pyboard/lib/be_helpers cp be_helpers/* /pyboard/lib/be_helpers ``` -------------------------------- ### Install micropython-modbus using upip Source: https://micropython-modbus.readthedocs.io/en/latest/INSTALLATION Installs the micropython-modbus library from PyPI using the 'upip' package manager. This method is primarily for older MicroPython versions (prior to 1.19.1) as 'upip' installation of specific versions is not supported in newer versions. It always installs the latest available version. ```python import upip upip.install('micropython-modbus') ``` -------------------------------- ### Setup Virtual Environment and Install Dependencies (Python) Source: https://micropython-modbus.readthedocs.io/en/latest/CONTRIBUTING Commands to set up a Python virtual environment and install testing dependencies. This ensures a consistent environment for verifying changes. ```shell python3 -m venv .venv source .venv/bin/activate pip install -r requirements-test.txt ``` -------------------------------- ### Install micropython-modbus using mpremote Source: https://micropython-modbus.readthedocs.io/en/latest/INSTALLATION Installs the micropython-modbus library on a connected MicroPython device from a host machine using the 'mpremote' tool. This method is for installations without direct network access on the MicroPython board. Requires 'mpremote' to be installed on the host. ```bash mpremote connect /dev/tty.SLAB_USBtoUART mip install github:brainelectronics/micropython-modbus ``` -------------------------------- ### Install Release Candidate of micropython-modbus using mip Source: https://micropython-modbus.readthedocs.io/en/latest/INSTALLATION Installs a specific release candidate version of the micropython-modbus library from GitHub using the 'mip' package manager. Requires MicroPython 1.19.1 or later. Specify the desired version tag. ```python import mip mip.install('github:brainelectronics/micropython-modbus', version='2.3.3-rc31.dev59') ``` -------------------------------- ### TCP Client/Slave Setup (MicroPython) Source: https://micropython-modbus.readthedocs.io/en/latest/EXAMPLES Initializes a MicroPython device to act as a Modbus TCP client/slave. It requires network connection setup and then instantiates the ModbusTCP class, specifying the local IP address and TCP port for communication. This setup allows the device to provide data to requesting host devices over a TCP socket. ```python import network from umodbus.tcp import ModbusTCP # network connections shall be made here, check the MicroPython port specific # documentation for connecting to or creating a network # TCP Client/Slave setup # set IP address of this MicroPython device explicitly # local_ip = '192.168.4.1' # IP address # or get it from the system after a connection to the network has been made # it is not the task of this lib to provide a detailed explanation for this station = network.WLAN(network.STA_IF) local_ip = station.ifconfig()[0] tcp_port = 502 # port to listen for requests/providing data client = ModbusTCP() ``` -------------------------------- ### Install micropython-modbus using mip Source: https://micropython-modbus.readthedocs.io/en/latest/INSTALLATION Installs the micropython-modbus library from GitHub using the 'mip' package manager. 'mip' is available in MicroPython 1.19.1 and later. This method uses the package definition file from the repository. ```python import mip mip.install('github:brainelectronics/micropython-modbus') ``` -------------------------------- ### Setup Registers Source: https://micropython-modbus.readthedocs.io/en/latest/umodbus Sets up all registers for the Modbus client. ```APIDOC ## POST /registers/setup ### Description Sets up all registers of the Modbus client. ### Method POST ### Endpoint /registers/setup ### Parameters #### Request Body - **registers** (dict, optional) - A dictionary containing the registers to set up. Defaults to an empty dictionary. - **use_default_vals** (Optional[bool], optional) - A flag to use default dummy values for registers. Defaults to False. ### Request Example ```json { "registers": { "01": {"type": "holding", "value": 100}, "02": {"type": "coil", "value": true} }, "use_default_vals": false } ``` ### Response #### Success Response (200) - **message** (string) - Indicates success or failure of the operation. #### Response Example ```json { "message": "Registers set up successfully" } ``` ``` -------------------------------- ### TCP Client Setup (MicroPython) Source: https://micropython-modbus.readthedocs.io/en/latest/USAGE Sets up a MicroPython device as a Modbus TCP client (formerly slave). This involves copying example scripts and rebooting the device to start serving dummy registers that can be read and updated by another device. The output shows network connection details and register serving status. ```bash cp examples/tcp_client_example.py /pyboard/main.py cp examples/boot.py /pyboard/boot.py repl ``` -------------------------------- ### RTU Client/Slave Setup (MicroPython) Source: https://micropython-modbus.readthedocs.io/en/latest/EXAMPLES Configures a MicroPython device to act as a Modbus RTU client/slave. It initializes the ModbusRTU class with specified pins and UART ID, then sets up register definitions for coils, holding registers, discrete inputs, and input registers. The device continuously processes requests. ```python from umodbus.serial import ModbusRTU # RTU Client/Slave setup # the following definition is for an ESP32 rtu_pins = (25, 26) # (TX, RX) uart_id = 1 # the following definition is for a RP2 # rtu_pins = (Pin(0), Pin(1)) # (TX, RX) # uart_id = 0 # # rtu_pins = (Pin(4), Pin(5)) # (TX, RX) # uart_id = 1 # the following definition is for a pyboard # rtu_pins = (Pin(PB6), Pin(PB7)) # (TX, RX) # uart_id = 1 slave_addr = 10 # address on bus as client client = ModbusRTU( addr=slave_addr, # address on bus pins=rtu_pins, # given as tuple (TX, RX) # baudrate=9600, # optional, default 9600 # data_bits=8, # optional, default 8 # stop_bits=1, # optional, default 1 # parity=None, # optional, default None # ctrl_pin=12, # optional, control DE/RE uart_id=uart_id # optional, default 1, see port specific documentation ) register_definitions = { "COILS": { "EXAMPLE_COIL": { "register": 123, "len": 1, "val": 1 } }, "HREGS": { "EXAMPLE_HREG": { "register": 93, "len": 1, "val": 19 } }, "ISTS": { "EXAMPLE_ISTS": { "register": 67, "len": 1, "val": 0 } }, "IREGS": { "EXAMPLE_IREG": { "register": 10, "len": 1, "val": 60001 } } } # use the defined values of each register type provided by register_definitions client.setup_registers(registers=register_definitions) while True: try: result = client.process() except KeyboardInterrupt: print('KeyboardInterrupt, stopping RTU client...') break except Exception as e: print('Exception during execution: {}'.format(e)) ``` -------------------------------- ### Set up MicroPython RTU Client Source: https://micropython-modbus.readthedocs.io/en/latest/USAGE Copies example files for an RTU client to the pyboard and initiates a soft reboot. The device will then serve Modbus registers. ```shell cp examples/rtu_client_example.py /pyboard/main.py cp examples/boot.py /pyboard/boot.py repl ``` -------------------------------- ### Install MicroPython Modbus Library using mip or upip Source: https://micropython-modbus.readthedocs.io/en/latest/readme_link Installs the 'micropython-modbus' library on a MicroPython board using either the 'mip' package manager (for newer versions) or 'upip' (for older versions). Requires network connectivity and basic MicroPython setup. ```micropython import machine import network import time import mip station = network.WLAN(network.STA_IF) station.active(True) station.connect('SSID', 'PASSWORD') time.sleep(1) print('Device connected to network: {}'.format(station.isconnected())) mip.install('github:brainelectronics/micropython-modbus') print('Installation completed') machine.soft_reset() ``` ```micropython import machine import network import time import upip station = network.WLAN(network.STA_IF) station.active(True) station.connect('SSID', 'PASSWORD') time.sleep(1) print('Device connected to network: {}'.format(station.isconnected())) upip.install('micropython-modbus') print('Installation completed') machine.soft_reset() ``` -------------------------------- ### Copy umodbus module using mpremote Source: https://micropython-modbus.readthedocs.io/en/latest/INSTALLATION Manually copies all files and folders of the 'umodbus' module to the MicroPython board using the 'mpremote' tool. This is a no-network installation method from the host machine. ```bash mpremote connect /dev/tty.SLAB_USBtoUART cp -r umodbus/ : ``` -------------------------------- ### Setup and Process Modbus TCP Client Registers Source: https://micropython-modbus.readthedocs.io/en/latest/EXAMPLES This snippet demonstrates how to bind a Modbus TCP client to a local IP and port, define register structures (Coils, Holding Registers, Input Status, Input Registers), and then continuously process Modbus requests. It handles potential KeyboardInterrupt and other exceptions during processing. ```python if not client.get_bound_status(): client.bind(local_ip=local_ip, local_port=tcp_port) register_definitions = { "COILS": { "EXAMPLE_COIL": { "register": 123, "len": 1, "val": 1 } }, "HREGS": { "EXAMPLE_HREG": { "register": 93, "len": 1, "val": 19 } }, "ISTS": { "EXAMPLE_ISTS": { "register": 67, "len": 1, "val": 0 } }, "IREGS": { "EXAMPLE_IREG": { "register": 10, "len": 1, "val": 60001 } } } client.setup_registers(registers=register_definitions) while True: try: result = client.process() except KeyboardInterrupt: print('KeyboardInterrupt, stopping TCP client...') break except Exception as e: print('Exception during execution: {}'.format(e)) ``` -------------------------------- ### Modbus Register Setup with Callbacks in Python Source: https://micropython-modbus.readthedocs.io/en/latest/USAGE This Python code demonstrates setting up Modbus registers with associated callback functions for get and set operations. It defines a dictionary `register_definitions` including a coil with its register, length, default value, and callback functions. The `client.setup_registers` function is then used to apply these definitions. ```python # assuming the client specific setup (port/ID settings, network connections, # UART setup) has already been done # Check the provided examples for further details # define some registers, for simplicity only a single coil is used register_definitions = { "COILS": { "EXAMPLE_COIL": { "register": 123, "len": 1, "val": 0, "on_get_cb": my_coil_get_cb, "on_set_cb": my_coil_set_cb } } } print('Setting up registers ...') # use the defined values of each register type provided by register_definitions client.setup_registers(registers=register_definitions) # alternatively use dummy default values (True for bool regs, 999 otherwise) # client.setup_registers(registers=register_definitions, use_default_vals=True) ``` -------------------------------- ### Set up MicroPython RTU Host Source: https://micropython-modbus.readthedocs.io/en/latest/USAGE Copies example files for an RTU host to the pyboard and initiates a soft reboot. The device will then request and update registers from an RTU client. ```shell cp examples/rtu_host_example.py /pyboard/main.py cp examples/boot.py /pyboard/boot.py repl ``` -------------------------------- ### Register Setup with Callbacks Source: https://micropython-modbus.readthedocs.io/en/latest/USAGE Demonstrates how to add coils with custom callback functions for 'on_set' and 'on_get' events. ```APIDOC ## POST /api/registers/coil/add ### Description Adds a coil with optional callback functions for setting and getting its value. ### Method POST ### Endpoint `/api/registers/coil/add` ### Parameters #### Request Body - **address** (integer) - Required - The address of the coil. - **value** (boolean) - Required - The initial value of the coil. - **on_set_cb** (function) - Optional - Callback function executed when the coil is set. - **on_get_cb** (function) - Optional - Callback function executed when the coil is read. ### Request Example ```json { "address": 123, "value": true, "on_set_cb": "my_coil_set_cb", "on_get_cb": "my_coil_get_cb" } ``` ### Response #### Success Response (200) - **status** (string) - Indicates 'Register setup done'. #### Response Example ```json { "status": "Register setup done" } ``` ``` -------------------------------- ### Copy umodbus module using rshell Source: https://micropython-modbus.readthedocs.io/en/latest/INSTALLATION Manually copies all files and folders of the 'umodbus' module to the '/lib/umodbus' directory on the MicroPython board using 'rshell'. This requires an active 'rshell' session. ```bash mkdir /pyboard/lib mkdir /pyboard/lib/umodbus cp umodbus/* /pyboard/lib/umodbus ``` -------------------------------- ### Connect to rshell Source: https://micropython-modbus.readthedocs.io/en/latest/INSTALLATION Opens a remote shell connection to a MicroPython device using 'rshell'. This is a prerequisite for manually copying files or installing packages without direct network access on the board. The '-p' flag specifies the serial port, and '--editor' sets the preferred editor. ```bash rshell -p /dev/tty.SLAB_USBtoUART --editor nano ``` -------------------------------- ### Install Additional MicroPython Packages Source: https://micropython-modbus.readthedocs.io/en/latest/readme_link Installs supplementary MicroPython modules required for certain functionalities, such as 'micropython-modules' or 'micropython-brainelectronics-helpers', using 'mip' or 'upip'. ```micropython # with MicroPython version 1.19.1 or newer import mip mip.install('github:brainelectronics/micropython-modules') ``` ```micropython # before MicroPython version 1.19.1 import upip upip.install('micropython-brainelectronics-helpers') ``` -------------------------------- ### Modbus TCP Master Read Coils Example Source: https://micropython-modbus.readthedocs.io/en/latest/EXAMPLES This example shows how to configure and use a Modbus TCP Master (host) to read data from a slave device. It initializes the master with the slave's IP address and port, then reads the status of a specific coil. Ensure the network connections are valid before execution. ```python from umodbus.tcp import TCP as ModbusTCPMaster slave_tcp_port = 502 salve_ip = '192.168.178.69' host = ModbusTCPMaster( slave_ip=slave_ip, slave_port=slave_tcp_port, ) coil_status = host.read_coils(slave_addr=10, starting_addr=123, coil_qty=1) print('Status of coil 123: {}'.format(coil_status)) ``` -------------------------------- ### Register Callbacks for Modbus TCP Source: https://micropython-modbus.readthedocs.io/en/latest/EXAMPLES This section details how to implement custom callback functions in micropython-modbus. It shows defining functions for `on_set_cb` and `on_get_cb` events, associating them with specific registers during setup or later using `add_coil`. Note that `on_set_cb` is only available for coils and holding registers. ```python def my_coil_set_cb(reg_type, address, val): print('Custom callback, called on setting {} at {} to: {}'. format(reg_type, address, val)) def my_coil_get_cb(reg_type, address, val): print('Custom callback, called on getting {} at {}, currently: {}'. format(reg_type, address, val)) register_definitions = { "COILS": { "EXAMPLE_COIL": { "register": 123, "len": 1, "val": 1, "on_get_cb": my_coil_get_cb, "on_set_cb": my_coil_set_cb } } } client.setup_registers(registers=register_definitions) client.add_coil( address=123, value=bool(1), on_set_cb=my_coil_set_cb, on_get_cb=my_coil_get_cb ) ``` -------------------------------- ### TCP Host Setup (MicroPython) Source: https://micropython-modbus.readthedocs.io/en/latest/USAGE Sets up a MicroPython device as a Modbus TCP host (formerly master). This involves copying example scripts and rebooting the device. After reboot, the device will request and update registers on a TCP client. The log output demonstrates successful connection and data exchange, including reading and writing coils and registers. ```bash cp examples/tcp_host_example.py /pyboard/main.py cp examples/boot.py /pyboard/boot.py repl ``` -------------------------------- ### Flash MicroPython Firmware to ESP32 Source: https://micropython-modbus.readthedocs.io/en/latest/SETUP Flashes the MicroPython firmware to an ESP32 board using esptool.py. This involves erasing the existing flash and then writing the specified firmware file. ```bash esptool.py --chip esp32 --port /dev/tty.SLAB_USBtoUART erase_flash esptool.py --chip esp32 --port /dev/tty.SLAB_USBtoUART --baud 921600 write_flash -z 0x1000 esp32spiram-20220117-v1.18.bin ``` -------------------------------- ### Update Git Submodule Source: https://micropython-modbus.readthedocs.io/en/latest/SETUP Updates the Git submodule for brainelectronics python modules. Ensures the necessary modules are cloned and can be checked out to a specific version or the develop branch. ```bash git submodule update --init --recursive cd modules git fetch # maybe checkout a newer version with the following command git checkout x.y.z # or use the latest develop branch git checkout develop git pull ``` -------------------------------- ### Install and Run Pre-commit Hooks (Python) Source: https://micropython-modbus.readthedocs.io/en/latest/CONTRIBUTING Commands to install and execute pre-commit hooks. These hooks automate various checks, including code formatting and changelog validation, before each commit. ```shell # install pre-commit to run before each commit, optionally pre-commit install pre-commit run --all-files ``` -------------------------------- ### Run Simple MicroPython Docker Container Source: https://micropython-modbus.readthedocs.io/en/latest/TESTING Starts a simple MicroPython Docker container for basic testing and interactive REPL sessions. It mounts the host's network and sets the entrypoint to bash. ```bash docker run -it --name micropython-1.18 --network=host --entrypoint bash micropython/unix:v1.18 ``` -------------------------------- ### Minimal Modbus Coil Configuration (JSON) Source: https://micropython-modbus.readthedocs.io/en/latest/USAGE A minimal example demonstrating the JSON structure required to define a single Modbus coil. It only includes the essential 'register' and 'len' properties. ```json { "COILS": { "COIL_NAME": { "register": 42, "len": 1 } } } ``` -------------------------------- ### Build and Run MicroPython Tests with Docker Source: https://micropython-modbus.readthedocs.io/en/latest/CONTRIBUTING Docker commands to build the test image, execute client/host examples, and run TCP/RTU tests. These commands facilitate testing in isolated environments, simulating hardware interactions. ```shell # build and run the "native" unittests docker build --tag micropython-test --file Dockerfile.tests . # Execute client/host TCP examples docker compose up --build --exit-code-from micropython-host # Run client/host TCP tests docker compose -f docker-compose-tcp-test.yaml up --build --exit-code-from micropython-host # Run client/host RTU examples with faked RTU via TCP docker compose -f docker-compose-rtu.yaml up --build --exit-code-from micropython-host # Run client/host RTU tests docker compose -f docker-compose-rtu-test.yaml up --build --exit-code-from micropython-host ``` -------------------------------- ### Connect to Wireless Network in MicroPython Source: https://micropython-modbus.readthedocs.io/en/latest/INSTALLATION This code snippet demonstrates how to connect a MicroPython board to a wireless network using the 'network' module. It requires the board to have network capabilities. The output is a boolean indicating if the connection was successful. ```python import network station = network.WLAN(network.STA_IF) station.connect('SSID', 'PASSWORD') # wait some time to establish the connection station.isconnected() ``` -------------------------------- ### Modbus Register Configuration Example Source: https://micropython-modbus.readthedocs.io/en/latest/USAGE A Python dictionary defining Modbus registers, including custom names, register addresses, and the amount of registers to request. This structure is used to configure coils within the Modbus communication setup. ```python { "COILS": { # this key shall contain all coils "COIL_NAME": { # custom name of a coil "register": 124, # register address of the coil "len": 5 # amount of registers to request aka quantity } } } ``` -------------------------------- ### Read Holding Registers (MicroPython Modbus) Source: https://micropython-modbus.readthedocs.io/en/latest/UPGRADE This snippet demonstrates reading multiple holding registers from a Modbus slave device using MicroPython. It specifies the slave address, starting register address, quantity of registers to read, and indicates that the values are unsigned. The function returns a tuple of the read register values. ```python >>> host.read_holding_registers(slave_addr=10, starting_addr=93, register_qty=2, signed=False) (19, 29) ``` -------------------------------- ### Setup Async Mock Object in Python Source: https://micropython-modbus.readthedocs.io/en/latest/_modules/mock/mock Configures an asynchronous mock object by setting attributes related to await calls. It initializes await counts, arguments, and lists, and defines assertion wrappers using `partial` to correctly bind attributes during late binding scenarios. ```python def _setup_async_mock(mock): mock._is_coroutine = asyncio.coroutines._is_coroutine mock.await_count = 0 mock.await_args = None mock.await_args_list = _CallList() def wrapper(attr, *args, **kwargs): return getattr(mock.mock, attr)(*args, **kwargs) for attribute in ('assert_awaited', 'assert_awaited_once', 'assert_awaited_with', 'assert_awaited_once_with', 'assert_any_await', 'assert_has_awaits', 'assert_not_awaited'): setattr(mock, attribute, partial(wrapper, attribute)) ``` -------------------------------- ### Setup Registers Source: https://micropython-modbus.readthedocs.io/en/latest/_modules/umodbus/modbus Sets up all registers for the Modbus client. This method can be used to initialize registers with provided values or default values. It supports different register types like COILS, HREGS, ISTS, and IREGS. ```APIDOC ## SETUP_REGISTERS ### Description Sets up all registers for the Modbus client. This method can be used to initialize registers with provided values or default values. It supports different register types like COILS, HREGS, ISTS, and IREGS. ### Method Not specified (likely an internal method) ### Endpoint Not applicable (internal method) ### Parameters #### Path Parameters None #### Query Parameters None #### Request Body - **registers** (dict) - Optional - A dictionary containing register configurations. Structure: `{ 'REGISTER_TYPE': { 'register_id': { 'register': address, 'len': length, 'val': value, 'on_set_cb': callback, 'on_get_cb': callback } } }` - **use_default_vals** (Optional[bool]) - Optional - If True, uses default values for registers. Defaults to False. ### Request Example ```json { "registers": { "COILS": { "0": {"register": 0, "val": false, "on_set_cb": "my_coil_set_callback"} }, "HREGS": { "1": {"register": 1, "len": 2, "val": [10, 20]} } }, "use_default_vals": false } ``` ### Response #### Success Response (200) This method does not return a value (`None`). #### Response Example None ``` -------------------------------- ### Manage Modbus Coils Source: https://micropython-modbus.readthedocs.io/en/latest/_modules/umodbus/modbus Functions for managing Modbus coils, including adding, removing, setting, and getting coil values. Supports optional callbacks for set and get operations. Requires address and optionally a value or callback functions. ```python def remove_coil(self, address: int) -> Union[None, bool, List[bool]]: """ Remove a coil from the modbus register dictionary. :param address: The address (ID) of the register :type address: int :returns: Register value, None if register did not exist in dict :rtype: Union[None, bool, List[bool]] """ return self._remove_reg_from_dict(reg_type='COILS', address=address) def set_coil(self, address: int, value: Union[bool, List[bool]] = False) -> None: """ Set the coil value. :param address: The address (ID) of the register :type address: int :param value: The default value :type value: Union[bool, List[bool]], optional """ self._set_reg_in_dict(reg_type='COILS', address=address, value=value) def get_coil(self, address: int) -> Union[bool, List[bool]]: """ Get the coil value. :param address: The address (ID) of the register :type address: bool :returns: Coil value :rtype: Union[bool, List[bool]] """ return self._get_reg_in_dict(reg_type='COILS', address=address) @property def coils(self) -> dict_keys: """ Get the configured coils. :returns: The dictionary keys. :rtype: dict_keys """ return self._get_regs_of_dict(reg_type='COILS') ``` -------------------------------- ### Enter Patch Context Manager Source: https://micropython-modbus.readthedocs.io/en/latest/_modules/mock/mock This method sets up the patching context. It normalizes various spec and autospec arguments, validates combinations, retrieves the original attribute, and determines the appropriate mock class (MagicMock or AsyncMock) based on the original object's type and patching requirements. ```python def __enter__(self): """Perform the patch.""" new, spec, spec_set = self.new, self.spec, self.spec_set autospec, kwargs = self.autospec, self.kwargs new_callable = self.new_callable self.target = self.getter() # normalise False to None if spec is False: spec = None if spec_set is False: spec_set = None if autospec is False: autospec = None if spec is not None and autospec is not None: raise TypeError("Can't specify spec and autospec") if ((spec is not None or autospec is not None) and spec_set not in (True, None)): raise TypeError("Can't provide explicit spec_set *and* spec or autospec") original, local = self.get_original() if new is DEFAULT and autospec is None: inherit = False if spec is True: # set spec to the object we are replacing spec = original if spec_set is True: spec_set = original spec = None elif spec is not None: if spec_set is True: spec_set = spec spec = None elif spec_set is True: spec_set = original if spec is not None or spec_set is not None: if original is DEFAULT: raise TypeError("Can't use 'spec' with create=True") if isinstance(original, type): # If we're patching out a class and there is a spec inherit = True if spec is None and _is_async_obj(original): Klass = AsyncMock else: Klass = MagicMock _kwargs = {} if new_callable is not None: Klass = new_callable elif spec is not None or spec_set is not None: this_spec = spec if spec_set is not None: this_spec = spec_set if _is_list(this_spec): not_callable = '__call__' not in this_spec else: not_callable = not callable(this_spec) if _is_async_obj(this_spec): Klass = AsyncMock elif not_callable: Klass = NonCallableMagicMock if spec is not None: _kwargs['spec'] = spec if spec_set is not None: _kwargs['spec_set'] = spec_set # add a name to mocks if ( isinstance(Klass, type) and issubclass(Klass, NonCallableMock) and self.attribute): _kwargs['name'] = self.attribute _kwargs.update(kwargs) new = Klass(**_kwargs) if inherit and _is_instance_mock(new): # we can only tell if the instance should be callable if the # spec is not a list this_spec = spec if spec_set is not None: this_spec = spec_set if (not _is_list(this_spec) and not _instance_callable(this_spec)): Klass = NonCallableMagicMock _kwargs.pop('name') new.return_value = Klass(_new_parent=new, _new_name='()', **_kwargs) elif autospec is not None: # spec is ignored, new *must* be default, spec_set is treated ``` -------------------------------- ### Manage Modbus Holding Registers Source: https://micropython-modbus.readthedocs.io/en/latest/_modules/umodbus/modbus Functions for managing Modbus holding registers, including adding, removing, setting, and getting register values. Supports optional callbacks for set and get operations. Requires address and optionally a value or callback functions. ```python def add_hreg(self, address: int, value: Union[int, List[int]] = 0, on_set_cb: Callable[[str, int, List[int]], None] = None, on_get_cb: Callable[[str, int, List[int]], None] = None) -> None: """ Add a holding register to the modbus register dictionary. :param address: The address (ID) of the register :type address: int :param value: The default value :type value: Union[int, List[int]], optional :param on_set_cb: Callback on setting the holding register :type on_set_cb: Callable[[str, int, List[int]], None] :param on_get_cb: Callback on getting the holding register :type on_get_cb: Callable[[str, int, List[int]], None] """ self._set_reg_in_dict(reg_type='HREGS', address=address, value=value, on_set_cb=on_set_cb, on_get_cb=on_get_cb) def remove_hreg(self, address: int) -> Union[None, int, List[int]]: """ Remove a holding register from the modbus register dictionary. :param address: The address (ID) of the register :type address: int :returns: Register value, None if register did not exist in dict :rtype: Union[None, int, List[int]] """ return self._remove_reg_from_dict(reg_type='HREGS', address=address) def set_hreg(self, address: int, value: Union[int, List[int]] = 0) -> None: """ Set the holding register value. :param address: The address (ID) of the register :type address: int :param value: The default value :type value: int or list of int, optional """ self._set_reg_in_dict(reg_type='HREGS', address=address, value=value) def get_hreg(self, address: int) -> Union[int, List[int]]: """ Get the holding register value. :param address: The address (ID) of the register :type address: int :returns: Holding register value :rtype: Union[int, List[int]] """ return self._get_reg_in_dict(reg_type='HREGS', address=address) @property def hregs(self) -> dict_keys: """ Get the configured holding registers. :returns: The dictionary keys. :rtype: dict_keys """ return self._get_regs_of_dict(reg_type='HREGS') ``` -------------------------------- ### Start or Stop Socket Server (Python) Source: https://micropython-modbus.readthedocs.io/en/latest/_modules/fakes/machine This setter method controls the running state of the socket server. If the value is True and the server is not already running, it acquires a lock and starts a new thread for the server. If the value is False and the server is running, it releases the lock to stop the server. ```python @serving.setter def serving(self, value: bool) -> None: """ Start or stop running the socket server. :param value: The value :type value: bool """ if value and (not self._server_lock.locked()): # start socket server if not already running self._server_lock.acquire() # parameters of the _serve function params = ( self._sock, self._send_queue, self._receive_queue, self._server_lock, self.logger ) _thread.start_new_thread(self._serve, params) self.logger.info('Socket {} started'. format('server' if self._is_server else 'client')) elif (value is False) and self._server_lock.locked(): # stop server if not already stopped self._server_lock.release() self.logger.info('Socket {} lock released'. format('server' if self._is_server else 'client')) ``` -------------------------------- ### Mock Module - Main Code (Python) Source: https://micropython-modbus.readthedocs.io/en/latest/_modules/mock/mock The main source code for the mock module, defining various components for mocking and patching in Python. It includes imports for asyncio, contextlib, io, inspect, pprint, sys, builtins, and types. Dependencies include the 'mock' library for specific backports. ```python # mock.py # Test tools for mocking and patching. # Maintained by Michael Foord # Backport for other versions of Python available from # https://pypi.org/project/mock __all__ = ( 'Mock', 'MagicMock', 'patch', 'sentinel', 'DEFAULT', 'ANY', 'call', 'create_autospec', 'AsyncMock', 'FILTER_DIR', 'NonCallableMock', 'NonCallableMagicMock', 'mock_open', 'PropertyMock', 'seal', ) import asyncio import contextlib import io import inspect import pprint import sys import builtins from asyncio import iscoroutinefunction from types import CodeType, ModuleType, MethodType from unittest.util import safe_repr from functools import wraps, partial from mock import IS_PYPY from .backports import iscoroutinefunction _builtins = {name for name in dir(builtins) if not name.startswith('_')} FILTER_DIR = True # Workaround for issue #12370 ``` -------------------------------- ### machine.Pin Class Source: https://micropython-modbus.readthedocs.io/en/latest/_modules/fakes/machine Documentation for the machine.Pin class, including initialization and methods for setting and getting pin values. ```APIDOC ## Class: machine.Pin ### Description Provides an interface to control GPIO pins on the microcontroller. This class allows setting pin modes, reading digital values, and controlling output states. See https://docs.micropython.org/en/latest/library/machine.Pin.html for more details. ### Constants * **IN** (int): Represents the input mode for a pin. * **OUT** (int): Represents the output mode for a pin. ### Methods #### `__init__(self, pin: int, mode: int)` Initialize a Pin object. * **pin** (int) - Required - The pin number to configure. * **mode** (int) - Required - The mode to set the pin to (e.g., `Pin.IN` or `Pin.OUT`). #### `value(self, val: Optional[Union[int, bool]] = None) -> Optional[bool]` Set or get the value of the pin. * **val** (Optional[Union[int, bool]]) - Optional - If provided, sets the pin's value. Accepts `True`/`1` for high, `False`/`0` for low. If `None`, the current value is returned. * **Returns**: `Optional[bool]` - The current state of the pin if `val` is `None`, otherwise `None` after setting the value. #### `on(self) -> None` Set the pin to a high output level ('1'). This method only has an effect if the pin is configured in `Pin.OUT` mode. #### `off(self) -> None` Set the pin to a low output level ('0'). This method only has an effect if the pin is configured in `Pin.OUT` mode. ### Request Example ```python # Example usage (not a direct request/response structure) from machine import Pin pin_out = Pin(2, Pin.OUT) pin_out.on() # Set pin 2 to high pin_in = Pin(3, Pin.IN) current_state = pin_in.value() # Read the state of pin 3 ``` ### Response #### Success Response (200) * **value** (bool) - The state of the pin (True for high, False for low) when queried using `value()` with no arguments. #### Response Example ```python # Example of reading a pin value # Assume pin_in.value() returns True if the pin is high # This is conceptual as the actual return is direct from the method # { # "value": true # } ``` ``` -------------------------------- ### Build and Run RTU Integration Tests with Docker Compose Source: https://micropython-modbus.readthedocs.io/en/latest/TESTING This command builds the Docker images, starts the services defined in the `docker-compose-rtu-test.yaml` file, and runs the integration tests. It ensures that the `micropython-host-rtu` service exits with the correct code and cleans up orphaned containers afterward. ```bash docker compose -f docker-compose-rtu-test.yaml up --build --exit-code-from micropython-host-rtu --remove-orphans ``` -------------------------------- ### machine.Pin Class Source: https://micropython-modbus.readthedocs.io/en/latest/fakes A fake implementation of the MicroPython machine.Pin class. It allows setting and getting the value of a pin, simulating hardware pin behavior in a software environment. ```APIDOC ## machine.Pin Class ### Description Fake Micropython Pin class. Simulates hardware pin behavior. See https://docs.micropython.org/en/latest/library/machine.Pin.html ### Methods - `on() -> None`: Set pin to "1" output level. - `off() -> None`: Set pin to "0" output level. - `value(val: Optional[Union[int, bool]] = None) -> Optional[bool]`: Set or get the value of the pin. - **Parameters**: - `val` (Optional[Union[int, bool]]): The value to set the pin to. - **Returns**: State of the pin if no value is specified, None otherwise. - **Return Type**: Optional[bool] ``` -------------------------------- ### Build and Run Manual Unittests Docker Image Source: https://micropython-modbus.readthedocs.io/en/latest/TESTING Builds a Docker image specifically for manually running unit tests and then runs this image. This isolates the testing environment. ```bash docker build -t micropython-test-manually -f Dockerfile.tests_manually . docker run -it --name micropython-test-manually micropython-test-manually ``` -------------------------------- ### Mock Patching Lifecycle Management (Python) Source: https://micropython-modbus.readthedocs.io/en/latest/_modules/mock/mock Manages the activation and deactivation of mock patches. The `start` method activates a patch and returns any created mock, while `stop` deactivates an active patch. It handles exceptions during patching and ensures proper cleanup using an exit stack. ```python class _patch: # ... (previous methods like __init__) def __enter__(self): """Enter the runtime context related to this object.""" new_attr = self._create_new() self.temp_original = DEFAULT self.is_local = True self._exit_stack = contextlib.ExitStack() try: # Avoid calling _get_target for the object itself if self.getter is None: self.target = new_attr self.attribute = None else: self.target = self.getter() self.attribute = self._name if self.attribute is None: # patching a module or class return new_attr # If new is DEFAULT, we are creating a new mock if new_attr is DEFAULT: # Mock is created by _create_new, assign it new_attr = self.new # Original is stored only if it's not DEFAULT if self.original is not DEFAULT: self.temp_original = getattr(self.target, self.attribute) setattr(self.target, self.attribute, new_attr) # Handle additional patchers if provided if self.additional_patchers: extra_args = {} for patching in self.additional_patchers: arg = self._exit_stack.enter_context(patching) if patching.new is DEFAULT: extra_args.update(arg) if extra_args: return extra_args return new_attr except: # If an error occurs, attempt to exit the stack gracefully self._exit_stack.__exit__(*sys.exc_info()) if not self.__exit__(*sys.exc_info()): raise def __exit__(self, *exc_info): """Undo the patch.""" if self.is_local and self.temp_original is not DEFAULT: setattr(self.target, self.attribute, self.temp_original) else: # Use delattr if the attribute exists, otherwise do nothing if hasattr(self.target, self.attribute): delattr(self.target, self.attribute) # Special handling for certain attributes or proxy objects if not self.create and (not hasattr(self.target, self.attribute) or self.attribute in ('__doc__', '__module__', '__defaults__', '__annotations__', '__kwdefaults__')): if self.temp_original is not DEFAULT: setattr(self.target, self.attribute, self.temp_original) # Clean up temporary storage if hasattr(self, 'temp_original'): del self.temp_original if hasattr(self, 'is_local'): del self.is_local if hasattr(self, 'target'): del self.target exit_stack = self._exit_stack if hasattr(self, '_exit_stack'): del self._exit_stack # Ensure the exit stack is properly closed return exit_stack.__exit__(*exc_info) def start(self): """Activate a patch, returning any created mock.""" result = self.__enter__() self._active_patches.append(self) return result def stop(self): """Stop an active patch.""" try: self._active_patches.remove(self) except ValueError: # If the patch hasn't been started this will fail return None return self.__exit__(None, None, None) ```