dm-aiomodbus 0.3.7

This is my custom aiomodbus client

DM-aiomodbus

Asynchronous Modbus clients for Python with TCP and Serial connection support.

Links

• PyPI
• GitHub

---

Section	Description
Installation	How to install the package
Usage	How to use the package
Types	Types and classes used in the package
Inner Client Methods	List of methods available for inner client

---

Installation

Check if you have Python 3.8 or higher installed:

python3 --version

Install the package using pip:

pip install dm-aiomodbus

---

Usage

Windows Setup

import asyncio
import sys

if sys.platform == "win32":
    asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())

Code Example

import asyncio
from dm_aiomodbus import (
   DMAioModbusSerialClient, DMAioModbusSerialClientConfig,
   DMAioModbusTcpClient, DMAioModbusTcpClientConfig,
   DMAioModbusInnerClient
)


async def main():
    # Initialize Serial client
    serial_modbus_client = DMAioModbusSerialClient(
        config=DMAioModbusSerialClientConfig(
            port="/dev/ttyUSB0",
            baudrate=9600,  # default value
            bytesize=8,  # default value
            stopbits=2,  # default value
            parity="N",  # default value
            disconnect_timeout_s=20,  # default value
            operation_timeout_ms=100,  # default value
            error_logging=False  # default value
        )
    )

    # Initialize TCP client
    tcp_modbus_client = DMAioModbusTcpClient(
        config=DMAioModbusTcpClientConfig(
            host="192.168.1.5",
            port=502,  # default value
            disconnect_timeout_s=20,  # default value
            operation_timeout_ms=100,  # default value
            error_logging=False  # default value
        )
    )

    # Read/write register(s)
    async def callback(client: DMAioModbusInnerClient):
        await client.write_register(256, 1)
        result = await client.read_holding_registers(256, count=3)
        print(result)

    # Execute callback
    await serial_modbus_client.execute(callback)
    # or
    await tcp_modbus_client.execute(callback)


if __name__ == "__main__":
    asyncio.run(main())

Note: All read/write methods should be called inside your callback function

---

Types

Serial Client Config

class DMAioModbusSerialClientConfig:
    port: str  # Serial port name. Example: "/dev/ttyS0" - Linux UART, "/dev/ttyUSB0" - linux USB, "COM1" - Windows USB
    baudrate: int = 9600  # Baudrate in bits per second
    bytesize: Literal[7, 8] = 8  # Number of data bits
    stopbits: Literal[1, 2] = 2  # Number of stop bits
    parity: Literal["N", "E", "O"] = "N"  # Parity mode. N - None, E - Even, O - Odd
    disconnect_timeout_s: int = 20  # Disconnect timeout in seconds
    operation_timeout_ms: int = 100  # Not recommended to set it lower than 100ms
    error_logging: bool = False  # Enable error logging

TCP Client Config

class DMAioModbusTcpClientConfig:
    host: str  # IP address of the device
    port: int = 502  # Port number
    disconnect_timeout_s: int = 20  # Disconnect timeout in seconds
    operation_timeout_ms: int = 100  # For remote TCP connections, it is recommended to set 200 or higher
    error_logging: bool = False  # Enable error logging

Read Response

class DMAioModbusReadResponse:
    data: list[int]  # List of values read from device
    error: str  # Error message if operation failed

Note: This class has to_dict() method that returns a dictionary

Warning: If the operation failed, the data field will be an empty list

Write Response

class DMAioModbusWriteResponse:
    status: bool  # Boolean indicating success or failure
    error: str  # Error message if operation failed

Note: This class has to_dict() method that returns a dictionary

Warning: The status is considered True, if the operation did not result in an error.

---

Inner Client Methods

Method	Arguments	Response Type
read_coils	- address int - count int = 1 - slave int= 1	ReadResponse
read_discrete_inputs	- address int - count int = 1 - slave int= 1	ReadResponse
read_holding_registers	- address int - count int = 1 - slave int= 1	ReadResponse
read_input_registers	- address int - count int = 1 - slave int= 1	ReadResponse
write_coil	- address int - value int - slave int = 1	WriteResponse
write_register	- address int - value int - slave int = 1	WriteResponse
write_coils	- address int - values list[int] | int - slave int= 1	WriteResponse
write_registers	- address int - values list[int] | int - slave int= 1	WriteResponse

Parameters Description

• address: Register address (single integer)
• count: Number of items to read (default: 1)
• value/values: Value(s) to write (single integer or list of integers)
• slave: Slave unit address (default: 1)

Set custom logger parameters

from dm_aiomodbus import DMAioModbusSerialClient
from dm_logger import FormatterConfig


# set up custom logger for all clients
DMAioModbusSerialClient.set_logger_params(
   {
      "name": "my_name",
      "formatter_config": FormatterConfig(
         show_datetime=False,
      )
   }
)

See more about DMLogger here