# DM-aiomodbus
## Urls
* [PyPI](https://pypi.org/project/dm-aiomodbus)
* [GitHub](https://github.com/DIMKA4621/dm-aiomodbus)
## Usage
### Connection
* Serial
```python
from dm_aiomodbus import DMAioModbusSerialClient
modbus_client = DMAioModbusSerialClient(
port="/dev/ttyUSB0",
baudrate=9600,
bytesize=8,
stopbits=2,
parity="N",
name_tag="my_serial_plc"
)
```
* TCP
```python
from dm_aiomodbus import DMAioModbusTcpClient
modbus_client = DMAioModbusTcpClient(
host="192.168.0.0",
port=501,
name_tag="my_tcp_plc"
)
```
### Requests
```python
from dm_aiomodbus import DMAioModbusTcpClient, DMAioModbusTempClientInterface
import asyncio
async def main():
# create client
modbus_client = DMAioModbusTcpClient(
host="192.168.0.0",
port=501,
name_tag="my_tcp_plc"
)
# create read callback
async def read_callback(client: DMAioModbusTempClientInterface):
reg_258_259, err1 = await client.read_holding_registers(258, count=2) # get values and error if any
reg_256, err2 = await client.read_holding_registers(256)
reg_260_2, err3 = await client.read_holding_registers(address=260, slave=2) # read second slave-device
print(reg_258_259, reg_256, reg_260_2)
# create read callback
async def write_callback(client: DMAioModbusTempClientInterface):
status, err = await client.write_register(256, 1) # get write status and error if any
await client.write_register(260, value=0, slave=2) # write second slave-device
# request to plc
modbus_client.execute(read_callback) # execute without waiting result
# or
await modbus_client.execute_and_return(write_callback, timeout=3) # execute and wait result with timeout 3s (default 5)
if __name__ == "__main__":
asyncio.run(main())
```
### Optional init parameters
| Parameter | Type | Default Value | Description |
|----------------------------|-------|---------------|---------------------------------------------------------------------|
| `disconnect_timeout_s` | `int` | `20` | timeout waiting for an active connection after the last request (s) |
| `after_execute_timeout_ms` | `int` | `3` | timeout between requests (ms) |
### Set custom logger
_If you want set up custom logger_
```python
from dm_aiomodbus import DMAioModbusTcpClient # or another client
# create custom logger
class MyLogger:
def debug(self, message):
pass
def info(self, message):
pass
def warning(self, message):
print(message)
def error(self, message):
print(message)
# set up custom logger for all clients
DMAioModbusTcpClient.set_logger(MyLogger())
```
### Run in Windows
_If you run async code in **Windows**, set correct selector_
```python
import asyncio
import sys
if sys.platform == "win32":
asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
```
Raw data
{
"_id": null,
"home_page": "https://pypi.org/project/dm-aiomodbus",
"name": "dm-aiomodbus",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.8",
"maintainer_email": null,
"keywords": "dm aiomodbus",
"author": "dimka4621",
"author_email": "mismartconfig@gmail.com",
"download_url": "https://files.pythonhosted.org/packages/9d/85/6a5d53cd7fdae04096223e8cfd35dbc8c8f73501d473002dca8857f305de/dm_aiomodbus-0.1.8.tar.gz",
"platform": null,
"description": "# DM-aiomodbus\n\n## Urls\n\n* [PyPI](https://pypi.org/project/dm-aiomodbus)\n* [GitHub](https://github.com/DIMKA4621/dm-aiomodbus)\n\n## Usage\n\n### Connection\n\n* Serial\n ```python\n from dm_aiomodbus import DMAioModbusSerialClient\n\n modbus_client = DMAioModbusSerialClient(\n port=\"/dev/ttyUSB0\",\n baudrate=9600,\n bytesize=8,\n stopbits=2,\n parity=\"N\",\n name_tag=\"my_serial_plc\"\n )\n ```\n\n* TCP\n ```python\n from dm_aiomodbus import DMAioModbusTcpClient\n\n modbus_client = DMAioModbusTcpClient(\n host=\"192.168.0.0\",\n port=501,\n name_tag=\"my_tcp_plc\"\n )\n ```\n\n### Requests\n```python\nfrom dm_aiomodbus import DMAioModbusTcpClient, DMAioModbusTempClientInterface\nimport asyncio\n\n\nasync def main():\n # create client\n modbus_client = DMAioModbusTcpClient(\n host=\"192.168.0.0\",\n port=501,\n name_tag=\"my_tcp_plc\"\n )\n\n # create read callback\n async def read_callback(client: DMAioModbusTempClientInterface):\n reg_258_259, err1 = await client.read_holding_registers(258, count=2) # get values and error if any\n reg_256, err2 = await client.read_holding_registers(256)\n reg_260_2, err3 = await client.read_holding_registers(address=260, slave=2) # read second slave-device\n print(reg_258_259, reg_256, reg_260_2)\n\n # create read callback\n async def write_callback(client: DMAioModbusTempClientInterface):\n status, err = await client.write_register(256, 1) # get write status and error if any\n await client.write_register(260, value=0, slave=2) # write second slave-device\n\n # request to plc\n modbus_client.execute(read_callback) # execute without waiting result\n # or\n await modbus_client.execute_and_return(write_callback, timeout=3) # execute and wait result with timeout 3s (default 5)\n\n\nif __name__ == \"__main__\":\n asyncio.run(main())\n```\n\n### Optional init parameters\n\n| Parameter | Type | Default Value | Description |\n|----------------------------|-------|---------------|---------------------------------------------------------------------|\n| `disconnect_timeout_s` | `int` | `20` | timeout waiting for an active connection after the last request (s) |\n| `after_execute_timeout_ms` | `int` | `3` | timeout between requests (ms) |\n\n### Set custom logger\n\n_If you want set up custom logger_\n\n```python\nfrom dm_aiomodbus import DMAioModbusTcpClient # or another client\n\n\n# create custom logger\nclass MyLogger:\n def debug(self, message):\n pass\n\n def info(self, message):\n pass\n\n def warning(self, message):\n print(message)\n\n def error(self, message):\n print(message)\n\n\n# set up custom logger for all clients\nDMAioModbusTcpClient.set_logger(MyLogger())\n```\n\n### Run in Windows\n\n_If you run async code in **Windows**, set correct selector_\n\n```python\nimport asyncio\nimport sys\n\nif sys.platform == \"win32\":\n asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())\n```\n",
"bugtrack_url": null,
"license": null,
"summary": "This is my custom aiomodbus client",
"version": "0.1.8",
"project_urls": {
"GitHub": "https://github.com/DIMKA4621/dm-aiomodbus",
"Homepage": "https://pypi.org/project/dm-aiomodbus"
},
"split_keywords": [
"dm",
"aiomodbus"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "4c63f2056ef50910bc3ac4cfa07e3041f5bf04ed6960e7c0e15483d4d4338835",
"md5": "32e4c0f4e3c114feb460f7a223719df4",
"sha256": "27678a1c1690c6ec3cf299c84b7d83822cd38916aeec22ed38d8aece96669ebe"
},
"downloads": -1,
"filename": "dm_aiomodbus-0.1.8-py3-none-any.whl",
"has_sig": false,
"md5_digest": "32e4c0f4e3c114feb460f7a223719df4",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.8",
"size": 6586,
"upload_time": "2024-04-19T11:40:42",
"upload_time_iso_8601": "2024-04-19T11:40:42.266150Z",
"url": "https://files.pythonhosted.org/packages/4c/63/f2056ef50910bc3ac4cfa07e3041f5bf04ed6960e7c0e15483d4d4338835/dm_aiomodbus-0.1.8-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "9d856a5d53cd7fdae04096223e8cfd35dbc8c8f73501d473002dca8857f305de",
"md5": "918776db12ecd558d03003d0ea401265",
"sha256": "bb68a6ce95136ea07a4f69e76d2a92ef2c17e3ed05fca58857e15efa021a91d7"
},
"downloads": -1,
"filename": "dm_aiomodbus-0.1.8.tar.gz",
"has_sig": false,
"md5_digest": "918776db12ecd558d03003d0ea401265",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.8",
"size": 5401,
"upload_time": "2024-04-19T11:40:44",
"upload_time_iso_8601": "2024-04-19T11:40:44.665822Z",
"url": "https://files.pythonhosted.org/packages/9d/85/6a5d53cd7fdae04096223e8cfd35dbc8c8f73501d473002dca8857f305de/dm_aiomodbus-0.1.8.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2024-04-19 11:40:44",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "DIMKA4621",
"github_project": "dm-aiomodbus",
"github_not_found": true,
"lcname": "dm-aiomodbus"
}
JSON
