Skip to content

Commit

Permalink
Test of write_coil. (#1479)
Browse files Browse the repository at this point in the history
  • Loading branch information
janiversen authored Apr 8, 2023
1 parent 1be1568 commit 5e3f8d5
Showing 1 changed file with 88 additions and 0 deletions.
88 changes: 88 additions & 0 deletions examples/client_sync_test.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,88 @@
#!/usr/bin/env python3
"""Pymodbus Synchronous Client Example.
An example of a single threaded synchronous client.
usage: client_sync.py [-h] [--comm {tcp,udp,serial,tls}]
[--framer {ascii,binary,rtu,socket,tls}]
[--log {critical,error,warning,info,debug}]
[--port PORT]
options:
-h, --help show this help message and exit
--comm {tcp,udp,serial,tls}
"serial", "tcp", "udp" or "tls"
--framer {ascii,binary,rtu,socket,tls}
"ascii", "binary", "rtu", "socket" or "tls"
--log {critical,error,warning,info,debug}
"critical", "error", "warning", "info" or "debug"
--port PORT the port to use
--baudrate BAUDRATE the baud rate to use for the serial device
The corresponding server must be started before e.g. as:
python3 server_sync.py
"""
import logging

import pymodbus.bit_read_message as pdu_bit_read
import pymodbus.bit_write_message as pdu_bit_write

# --------------------------------------------------------------------------- #
# import the various client implementations
# --------------------------------------------------------------------------- #
from examples.helper import get_commandline
from pymodbus.client import ModbusTcpClient
from pymodbus.exceptions import ModbusException


_logger = logging.getLogger()


def run_sync_client():
"""Run sync client."""
args = get_commandline()
_logger.info("### Create client object")
client = ModbusTcpClient(
args.host,
port=args.port,
framer=args.framer,
)

_logger.info("### Client starting")
client.connect()

_logger.info("### first read_coils to secure it works generally")
try:
rr = client.read_coils(1, 1, slave=1)
except ModbusException as exc:
_logger.error(exc)
raise RuntimeError(exc) from exc
if rr.isError():
raise RuntimeError("ERROR: read_coils returned an error!")
assert isinstance(rr, pdu_bit_read.ReadCoilsResponse)

_logger.info("### next write_coil to change a value")
try:
rr = client.write_coil(1, 17, slave=1)
except ModbusException as exc:
_logger.error(exc)
raise RuntimeError(exc) from exc
if rr.isError():
raise RuntimeError("ERROR: write_coil returned an error!")
assert isinstance(rr, pdu_bit_write.WriteSingleCoilResponse)

_logger.info("### finally read_coil to verify value")
try:
rr = client.read_coils(1, 1, slave=1)
except ModbusException as exc:
_logger.error(exc)
raise RuntimeError(exc) from exc
if rr.isError():
raise RuntimeError("ERROR: read_coils(2) returned an error!")
assert isinstance(rr, pdu_bit_read.ReadCoilsResponse)

client.close()
_logger.info("### End of Program")


if __name__ == "__main__":
run_sync_client()

0 comments on commit 5e3f8d5

Please sign in to comment.