You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@plc4x.apache.org by hu...@apache.org on 2024/08/29 13:10:27 UTC
(plc4x) branch feat/plc4py/modbus_write created (now d2c9435e4f)
This is an automated email from the ASF dual-hosted git repository.
hutcheb pushed a change to branch feat/plc4py/modbus_write
in repository https://gitbox.apache.org/repos/asf/plc4x.git
at d2c9435e4f feat(plc4py): get started on some modbus writing
This branch includes the following new commits:
new d2c9435e4f feat(plc4py): get started on some modbus writing
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(plc4x) 01/01: feat(plc4py): get started on some modbus writing
Posted by hu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hutcheb pushed a commit to branch feat/plc4py/modbus_write
in repository https://gitbox.apache.org/repos/asf/plc4x.git
commit d2c9435e4f93e0be812e4208e8fd9f6c7d69dd6e
Author: hutcheb <be...@gmail.com>
AuthorDate: Thu Aug 29 21:10:05 2024 +0800
feat(plc4py): get started on some modbus writing
---
.../templates/python/data-io-template.python.ftlh | 6 +-
plc4py/plc4py/drivers/modbus/ModbusConnection.py | 10 +-
plc4py/plc4py/drivers/modbus/ModbusDevice.py | 52 +++++++++-
.../plc4py/protocols/modbus/readwrite/DataItem.py | 16 +--
.../protocols/simulated/readwrite/DataItem.py | 72 ++++++-------
plc4py/setup.py | 2 +-
.../drivers/modbus/test_modbus_connection.py | 112 +++++++++++++++++++--
7 files changed, 210 insertions(+), 60 deletions(-)
diff --git a/code-generation/language-python/src/main/resources/templates/python/data-io-template.python.ftlh b/code-generation/language-python/src/main/resources/templates/python/data-io-template.python.ftlh
index 38464d206c..4e0ee3b537 100644
--- a/code-generation/language-python/src/main/resources/templates/python/data-io-template.python.ftlh
+++ b/code-generation/language-python/src/main/resources/templates/python/data-io-template.python.ftlh
@@ -76,7 +76,7 @@ class ${type.name}:
<#if discriminatorType.isEnumTypeReference()>
${helper.getLanguageTypeNameForTypeReference(discriminatorType)}.${helper.toParseExpression(dataIoTypeDefinition.switchField.orElseThrow(), discriminatorType, discriminatorValueTerm, parserArguments)}
<#else>
- ${helper.camelCaseToSnakeCase(helper.toParseExpression(dataIoTypeDefinition.switchField.orElseThrow(), discriminatorType, discriminatorValueTerm, parserArguments))}
+ ${helper.toParseExpression(dataIoTypeDefinition.switchField.orElseThrow(), discriminatorType, discriminatorValueTerm, parserArguments)}
</#if>
<#sep> and </#sep>
</#list>
@@ -131,7 +131,7 @@ class ${type.name}:
)
</...@compress>
- <#-- A terminated array keeps on reading data as long as the termination expression evaluates to false -->
+ <#-- A terminated array keeps on reading data as long as the termination expression evaluates to False -->
<#elseif arrayField.isTerminatedArrayField()>
# Terminated array
${arrayField.name}: ${helper.getNonPrimitiveLanguageTypeNameForField(arrayField)} = new LinkedList<>()
@@ -387,7 +387,7 @@ class ${type.name}:
<#if case.name == "BOOL">
while write_buffer.getPos() < len(write_buffer.get_data()):
- write_buffer.write_bit(false)
+ write_buffer.write_bit(False)
</#if>
<#break>
<#case "const">
diff --git a/plc4py/plc4py/drivers/modbus/ModbusConnection.py b/plc4py/plc4py/drivers/modbus/ModbusConnection.py
index 81656baa75..ac964974b7 100644
--- a/plc4py/plc4py/drivers/modbus/ModbusConnection.py
+++ b/plc4py/plc4py/drivers/modbus/ModbusConnection.py
@@ -25,7 +25,7 @@ from plc4py.api.messages.PlcRequest import (
PlcReadRequest,
PlcRequest,
PlcWriteRequest,
- ReadRequestBuilder,
+ ReadRequestBuilder, WriteRequestBuilder,
)
from plc4py.api.messages.PlcResponse import (
PlcResponse,
@@ -39,7 +39,7 @@ from plc4py.drivers.modbus.ModbusProtocol import ModbusProtocol
from plc4py.drivers.modbus.ModbusTag import ModbusTagBuilder
from plc4py.drivers.PlcDriverLoader import PlcDriverLoader
from plc4py.spi.messages.PlcReader import DefaultPlcReader
-from plc4py.spi.messages.PlcRequest import DefaultReadRequestBuilder
+from plc4py.spi.messages.PlcRequest import DefaultReadRequestBuilder, DefaultWriteRequestBuilder
from plc4py.spi.messages.PlcWriter import DefaultPlcWriter
from plc4py.spi.transport.Plc4xBaseTransport import Plc4xBaseTransport
from plc4py.spi.transport.TCPTransport import TCPTransport
@@ -136,6 +136,12 @@ class ModbusConnection(PlcConnection, DefaultPlcReader, DefaultPlcWriter):
"""
return DefaultReadRequestBuilder(ModbusTagBuilder)
+ def write_request_builder(self) -> WriteRequestBuilder:
+ """
+ :return: write request builder.
+ """
+ return DefaultWriteRequestBuilder(ModbusTagBuilder)
+
class ModbusDriver(PlcDriver):
def __init__(self):
diff --git a/plc4py/plc4py/drivers/modbus/ModbusDevice.py b/plc4py/plc4py/drivers/modbus/ModbusDevice.py
index 08b9e7a8e8..8cc11ee7f4 100644
--- a/plc4py/plc4py/drivers/modbus/ModbusDevice.py
+++ b/plc4py/plc4py/drivers/modbus/ModbusDevice.py
@@ -56,6 +56,11 @@ from plc4py.spi.messages.utils.ResponseItem import ResponseItem
from plc4py.spi.values.PlcValues import PlcList, PlcNull
from plc4py.utils.GenericTypes import AtomicInteger, ByteOrder
+from protocols.modbus.readwrite.ModbusPDUWriteMultipleCoilsRequest import ModbusPDUWriteMultipleCoilsRequest
+from protocols.modbus.readwrite.ModbusPDUWriteMultipleHoldingRegistersRequest import \
+ ModbusPDUWriteMultipleHoldingRegistersRequest, ModbusPDUWriteMultipleHoldingRegistersRequestBuilder
+from protocols.modbus.readwrite.ModbusPDUWriteSingleCoilRequest import ModbusPDUWriteSingleCoilRequest
+
@dataclass
class ModbusDevice:
@@ -156,4 +161,49 @@ class ModbusDevice:
"""
Writes one field from the Modbus Device
"""
- pass
+ if len(request.tags) > 1:
+ raise NotImplementedError(
+ "The Modbus driver only supports writing single tags at once"
+ )
+ if len(request.tags) == 0:
+ raise PlcRuntimeException("No tags have been specified to write")
+ tag = request.tags[request.tag_names[0]]
+ logging.debug(f"Writing tag {str(tag)} from Modbus Device")
+
+ # Create future to be returned when a value is returned
+ loop = asyncio.get_running_loop()
+ message_future = loop.create_future()
+
+ if isinstance(tag, ModbusTagCoil):
+ pdu = ModbusPDUWriteMultipleCoilsRequest(tag.address, tag.quantity, [v for k, v in request.values])
+ elif isinstance(tag, ModbusTagDiscreteInput):
+ raise PlcRuntimeException("Modbus doesn't support writing to discrete inputs")
+ elif isinstance(tag, ModbusTagInputRegister):
+ raise PlcRuntimeException("Modbus doesn't support writing to input registers")
+ elif isinstance(tag, ModbusTagHoldingRegister):
+ pdu = ModbusPDUWriteMultipleHoldingRegistersRequestBuilder(tag.address, tag.quantity, request.values).build()
+ else:
+ raise NotImplementedError(
+ "Modbus tag type not implemented " + str(tag.__class__)
+ )
+
+ adu = ModbusTcpADU(
+ False,
+ self._transaction_generator.increment(),
+ self._configuration.unit_identifier,
+ pdu,
+ )
+ write_buffer = WriteBufferByteBased(adu.length_in_bytes(), ByteOrder.BIG_ENDIAN)
+ adu.serialize(write_buffer)
+
+ protocol = transport.protocol
+ protocol.write_wait_for_response(
+ write_buffer.get_bytes(),
+ transport,
+ adu.transaction_identifier,
+ message_future,
+ )
+
+ await message_future
+ result = message_future.result()
+ pass
\ No newline at end of file
diff --git a/plc4py/plc4py/protocols/modbus/readwrite/DataItem.py b/plc4py/plc4py/protocols/modbus/readwrite/DataItem.py
index 1d2e54494a..2eec738436 100644
--- a/plc4py/plc4py/protocols/modbus/readwrite/DataItem.py
+++ b/plc4py/plc4py/protocols/modbus/readwrite/DataItem.py
@@ -58,7 +58,7 @@ class DataItem:
if (
data_type == ModbusDataType.BOOL
and number_of_values == int(1)
- and big_endian == true
+ and big_endian == True
): # BOOL
# Reserved Field (Compartmentalized so the "reserved" variable can't leak)
@@ -79,7 +79,7 @@ class DataItem:
if (
data_type == ModbusDataType.BOOL
and number_of_values == int(1)
- and big_endian == false
+ and big_endian == False
): # BOOL
# Reserved Field (Compartmentalized so the "reserved" variable can't leak)
@@ -120,7 +120,7 @@ class DataItem:
if (
data_type == ModbusDataType.BYTE
and number_of_values == int(1)
- and big_endian == true
+ and big_endian == True
): # BYTE
# Reserved Field (Compartmentalized so the "reserved" variable can't leak)
@@ -141,7 +141,7 @@ class DataItem:
if (
data_type == ModbusDataType.BYTE
and number_of_values == int(1)
- and big_endian == false
+ and big_endian == False
): # BYTE
# Simple Field (value)
@@ -189,7 +189,7 @@ class DataItem:
if (
data_type == ModbusDataType.SINT
and number_of_values == int(1)
- and big_endian == true
+ and big_endian == True
): # SINT
# Reserved Field (Compartmentalized so the "reserved" variable can't leak)
@@ -210,7 +210,7 @@ class DataItem:
if (
data_type == ModbusDataType.SINT
and number_of_values == int(1)
- and big_endian == false
+ and big_endian == False
): # SINT
# Simple Field (value)
@@ -287,7 +287,7 @@ class DataItem:
if (
data_type == ModbusDataType.USINT
and number_of_values == int(1)
- and big_endian == true
+ and big_endian == True
): # USINT
# Reserved Field (Compartmentalized so the "reserved" variable can't leak)
@@ -308,7 +308,7 @@ class DataItem:
if (
data_type == ModbusDataType.USINT
and number_of_values == int(1)
- and big_endian == false
+ and big_endian == False
): # USINT
# Simple Field (value)
diff --git a/plc4py/plc4py/protocols/simulated/readwrite/DataItem.py b/plc4py/plc4py/protocols/simulated/readwrite/DataItem.py
index e90c85a1c7..f5a2dc2bbf 100644
--- a/plc4py/plc4py/protocols/simulated/readwrite/DataItem.py
+++ b/plc4py/plc4py/protocols/simulated/readwrite/DataItem.py
@@ -48,13 +48,13 @@ import math
class DataItem:
@staticmethod
def static_parse(read_buffer: ReadBuffer, data_type: str, number_of_values: int):
- if data_type == "_bool" and number_of_values == int(1): # BOOL
+ if data_type == "BOOL" and number_of_values == int(1): # BOOL
# Simple Field (value)
value: bool = read_buffer.read_bit("")
return PlcBOOL(value)
- if data_type == "_bool": # List
+ if data_type == "BOOL": # List
# Array field (value)
# Count array
item_count: int = int(number_of_values)
@@ -63,13 +63,13 @@ class DataItem:
value.append(PlcBOOL(bool(read_buffer.read_bit(""))))
return PlcList(value)
- if data_type == "_byte" and number_of_values == int(1): # BYTE
+ if data_type == "BYTE" and number_of_values == int(1): # BYTE
# Simple Field (value)
value: int = read_buffer.read_unsigned_short(8, logical_name="")
return PlcBYTE(value)
- if data_type == "_byte": # List
+ if data_type == "BYTE": # List
# Array field (value)
# Count array
item_count: int = int(number_of_values)
@@ -80,13 +80,13 @@ class DataItem:
)
return PlcList(value)
- if data_type == "_word" and number_of_values == int(1): # WORD
+ if data_type == "WORD" and number_of_values == int(1): # WORD
# Simple Field (value)
value: int = read_buffer.read_unsigned_int(16, logical_name="")
return PlcWORD(value)
- if data_type == "_word": # List
+ if data_type == "WORD": # List
# Array field (value)
# Count array
item_count: int = int(number_of_values)
@@ -97,13 +97,13 @@ class DataItem:
)
return PlcList(value)
- if data_type == "_dword" and number_of_values == int(1): # DWORD
+ if data_type == "DWORD" and number_of_values == int(1): # DWORD
# Simple Field (value)
value: int = read_buffer.read_unsigned_long(32, logical_name="")
return PlcDWORD(value)
- if data_type == "_dword": # List
+ if data_type == "DWORD": # List
# Array field (value)
# Count array
item_count: int = int(number_of_values)
@@ -114,13 +114,13 @@ class DataItem:
)
return PlcList(value)
- if data_type == "_lword" and number_of_values == int(1): # LWORD
+ if data_type == "LWORD" and number_of_values == int(1): # LWORD
# Simple Field (value)
value: int = read_buffer.read_unsigned_long(64, logical_name="")
return PlcLWORD(value)
- if data_type == "_lword": # List
+ if data_type == "LWORD": # List
# Array field (value)
# Count array
item_count: int = int(number_of_values)
@@ -131,13 +131,13 @@ class DataItem:
)
return PlcList(value)
- if data_type == "_sint" and number_of_values == int(1): # SINT
+ if data_type == "SINT" and number_of_values == int(1): # SINT
# Simple Field (value)
value: int = read_buffer.read_signed_byte(8, logical_name="")
return PlcSINT(value)
- if data_type == "_sint": # List
+ if data_type == "SINT": # List
# Array field (value)
# Count array
item_count: int = int(number_of_values)
@@ -148,13 +148,13 @@ class DataItem:
)
return PlcList(value)
- if data_type == "_int" and number_of_values == int(1): # INT
+ if data_type == "INT" and number_of_values == int(1): # INT
# Simple Field (value)
value: int = read_buffer.read_short(16, logical_name="")
return PlcINT(value)
- if data_type == "_int": # List
+ if data_type == "INT": # List
# Array field (value)
# Count array
item_count: int = int(number_of_values)
@@ -163,13 +163,13 @@ class DataItem:
value.append(PlcINT(int(read_buffer.read_short(16, logical_name=""))))
return PlcList(value)
- if data_type == "_dint" and number_of_values == int(1): # DINT
+ if data_type == "DINT" and number_of_values == int(1): # DINT
# Simple Field (value)
value: int = read_buffer.read_int(32, logical_name="")
return PlcDINT(value)
- if data_type == "_dint": # List
+ if data_type == "DINT": # List
# Array field (value)
# Count array
item_count: int = int(number_of_values)
@@ -178,13 +178,13 @@ class DataItem:
value.append(PlcDINT(int(read_buffer.read_int(32, logical_name=""))))
return PlcList(value)
- if data_type == "_lint" and number_of_values == int(1): # LINT
+ if data_type == "LINT" and number_of_values == int(1): # LINT
# Simple Field (value)
value: int = read_buffer.read_long(64, logical_name="")
return PlcLINT(value)
- if data_type == "_lint": # List
+ if data_type == "LINT": # List
# Array field (value)
# Count array
item_count: int = int(number_of_values)
@@ -193,13 +193,13 @@ class DataItem:
value.append(PlcLINT(int(read_buffer.read_long(64, logical_name=""))))
return PlcList(value)
- if data_type == "_usint" and number_of_values == int(1): # USINT
+ if data_type == "USINT" and number_of_values == int(1): # USINT
# Simple Field (value)
value: int = read_buffer.read_unsigned_short(8, logical_name="")
return PlcUSINT(value)
- if data_type == "_usint": # List
+ if data_type == "USINT": # List
# Array field (value)
# Count array
item_count: int = int(number_of_values)
@@ -210,13 +210,13 @@ class DataItem:
)
return PlcList(value)
- if data_type == "_uint" and number_of_values == int(1): # UINT
+ if data_type == "UINT" and number_of_values == int(1): # UINT
# Simple Field (value)
value: int = read_buffer.read_unsigned_int(16, logical_name="")
return PlcUINT(value)
- if data_type == "_uint": # List
+ if data_type == "UINT": # List
# Array field (value)
# Count array
item_count: int = int(number_of_values)
@@ -227,13 +227,13 @@ class DataItem:
)
return PlcList(value)
- if data_type == "_udint" and number_of_values == int(1): # UDINT
+ if data_type == "UDINT" and number_of_values == int(1): # UDINT
# Simple Field (value)
value: int = read_buffer.read_unsigned_long(32, logical_name="")
return PlcUDINT(value)
- if data_type == "_udint": # List
+ if data_type == "UDINT": # List
# Array field (value)
# Count array
item_count: int = int(number_of_values)
@@ -244,13 +244,13 @@ class DataItem:
)
return PlcList(value)
- if data_type == "_ulint" and number_of_values == int(1): # ULINT
+ if data_type == "ULINT" and number_of_values == int(1): # ULINT
# Simple Field (value)
value: int = read_buffer.read_unsigned_long(64, logical_name="")
return PlcULINT(value)
- if data_type == "_ulint": # List
+ if data_type == "ULINT": # List
# Array field (value)
# Count array
item_count: int = int(number_of_values)
@@ -261,13 +261,13 @@ class DataItem:
)
return PlcList(value)
- if data_type == "_real" and number_of_values == int(1): # REAL
+ if data_type == "REAL" and number_of_values == int(1): # REAL
# Simple Field (value)
value: float = read_buffer.read_float(32, logical_name="")
return PlcREAL(value)
- if data_type == "_real": # List
+ if data_type == "REAL": # List
# Array field (value)
# Count array
item_count: int = int(number_of_values)
@@ -278,13 +278,13 @@ class DataItem:
)
return PlcList(value)
- if data_type == "_lreal" and number_of_values == int(1): # LREAL
+ if data_type == "LREAL" and number_of_values == int(1): # LREAL
# Simple Field (value)
value: float = read_buffer.read_double(64, logical_name="")
return PlcLREAL(value)
- if data_type == "_lreal": # List
+ if data_type == "LREAL": # List
# Array field (value)
# Count array
item_count: int = int(number_of_values)
@@ -295,13 +295,13 @@ class DataItem:
)
return PlcList(value)
- if data_type == "_char" and number_of_values == int(1): # CHAR
+ if data_type == "CHAR" and number_of_values == int(1): # CHAR
# Simple Field (value)
value: str = read_buffer.read_str(8, logical_name="", encoding="")
return PlcCHAR(value)
- if data_type == "_char": # List
+ if data_type == "CHAR": # List
# Array field (value)
# Count array
item_count: int = int(number_of_values)
@@ -314,13 +314,13 @@ class DataItem:
)
return PlcList(value)
- if data_type == "_wchar" and number_of_values == int(1): # WCHAR
+ if data_type == "WCHAR" and number_of_values == int(1): # WCHAR
# Simple Field (value)
value: str = read_buffer.read_str(16, logical_name="", encoding="")
return PlcWCHAR(value)
- if data_type == "_wchar": # List
+ if data_type == "WCHAR": # List
# Array field (value)
# Count array
item_count: int = int(number_of_values)
@@ -333,13 +333,13 @@ class DataItem:
)
return PlcList(value)
- if data_type == "_string": # STRING
+ if data_type == "STRING": # STRING
# Simple Field (value)
value: str = read_buffer.read_str(255, logical_name="", encoding="")
return PlcSTRING(value)
- if data_type == "_wstring": # STRING
+ if data_type == "WSTRING": # STRING
# Simple Field (value)
value: str = read_buffer.read_str(255, logical_name="", encoding="")
diff --git a/plc4py/setup.py b/plc4py/setup.py
index f245afe23f..253ccd7a30 100644
--- a/plc4py/setup.py
+++ b/plc4py/setup.py
@@ -21,7 +21,7 @@ from setuptools import setup, find_packages
setup(
name="plc4py",
- version="0.11a0",
+ version="0.13",
description="Plc4py The Python Industrial IOT Adapter",
classifiers=[
"Development Status :: 3 - Alpha",
diff --git a/plc4py/tests/unit/plc4py/drivers/modbus/test_modbus_connection.py b/plc4py/tests/unit/plc4py/drivers/modbus/test_modbus_connection.py
index 1fd9c49e6c..62a98fb0bb 100644
--- a/plc4py/tests/unit/plc4py/drivers/modbus/test_modbus_connection.py
+++ b/plc4py/tests/unit/plc4py/drivers/modbus/test_modbus_connection.py
@@ -24,6 +24,8 @@ from plc4py.PlcDriverManager import PlcDriverManager
from plc4py.api.value.PlcValue import PlcResponseCode
import logging
+from plc4py.spi.values.PlcValues import PlcINT
+
logger = logging.getLogger("testing")
@@ -46,7 +48,66 @@ async def manual_test_plc_driver_modbus_connect():
@pytest.mark.asyncio
@pytest.mark.xfail
-async def test_plc_driver_modbus_read():
+async def test_plc_driver_modbus_read_coil():
+ """
+ Test reading data from a Modbus PLC.
+ """
+ log = logging.getLogger(__name__)
+
+ # Initialize the PlcDriverManager
+ driver_manager = PlcDriverManager()
+
+ # Establish a connection to the Modbus PLC
+ async with driver_manager.connection("modbus://192.168.174.128:502") as connection:
+ with connection.read_request_builder() as builder:
+ builder.add_item("Random Tag", "0x00001[9]")
+ request = builder.build()
+ response = await connection.execute(request)
+ value = response.tags["Random Tag"].value
+
+
+@pytest.mark.asyncio
+@pytest.mark.xfail
+async def test_plc_driver_modbus_read_coil_array():
+ """
+ Test reading data from a Modbus PLC.
+ """
+ log = logging.getLogger(__name__)
+
+ # Initialize the PlcDriverManager
+ driver_manager = PlcDriverManager()
+
+ # Establish a connection to the Modbus PLC
+ async with driver_manager.connection("modbus://192.168.174.128:502") as connection:
+ with connection.read_request_builder() as builder:
+ builder.add_item("Random Tag", "0x00001[10]")
+ request = builder.build()
+ response = await connection.execute(request)
+ value = response.tags["Random Tag"].value
+
+
+@pytest.mark.asyncio
+async def test_plc_driver_modbus_read_contact_array():
+ """
+ Test reading data from a Modbus PLC.
+ """
+ log = logging.getLogger(__name__)
+
+ # Initialize the PlcDriverManager
+ driver_manager = PlcDriverManager()
+
+ # Establish a connection to the Modbus PLC
+ async with driver_manager.connection("modbus://192.168.174.128:502") as connection:
+ with connection.read_request_builder() as builder:
+ builder.add_item("Random Tag", "1x00001[10]")
+ request = builder.build()
+ response = await connection.execute(request)
+ value = response.tags["Random Tag"].value
+ pass
+
+
+@pytest.mark.asyncio
+async def test_plc_driver_modbus_read_input_register_array():
"""
Test reading data from a Modbus PLC.
"""
@@ -56,17 +117,50 @@ async def test_plc_driver_modbus_read():
driver_manager = PlcDriverManager()
# Establish a connection to the Modbus PLC
- async with driver_manager.connection("modbus://127.0.0.1:5020") as connection:
+ async with driver_manager.connection("modbus://192.168.174.128:502") as connection:
with connection.read_request_builder() as builder:
- builder.add_item("Random Tag", "4x00001[10]")
+ builder.add_item("Random Tag", "3x00001")
request = builder.build()
+ response = await connection.execute(request)
+ value = response.tags["Random Tag"].value
+ pass
+
+
+@pytest.mark.asyncio
+async def test_plc_driver_modbus_read_holding_array():
+ """
+ Test reading data from a Modbus PLC.
+ """
+ log = logging.getLogger(__name__)
- # Execute the read request
- for _ in range(100):
- future = connection.execute(request)
+ # Initialize the PlcDriverManager
+ driver_manager = PlcDriverManager()
- response = await future
+ # Establish a connection to the Modbus PLC
+ async with driver_manager.connection("modbus://192.168.174.128:502") as connection:
+ with connection.read_request_builder() as builder:
+ builder.add_item("Random Tag", "4x00001")
+ request = builder.build()
+ response = await connection.execute(request)
value = response.tags["Random Tag"].value
- log.info("Read tag 4x00001[10] - %s", value)
+ pass
+
+
+@pytest.mark.asyncio
+async def test_plc_driver_modbus_write_holding():
+ """
+ Test reading data from a Modbus PLC.
+ """
+ log = logging.getLogger(__name__)
- pass
+ # Initialize the PlcDriverManager
+ driver_manager = PlcDriverManager()
+
+ # Establish a connection to the Modbus PLC
+ async with driver_manager.connection("modbus://192.168.174.128:502") as connection:
+ with connection.write_request_builder() as builder:
+ builder.add_item("Random Tag", "4x00001", 2)
+ request = builder.build()
+ response = await connection.execute(request)
+ value = response.tags["Random Tag"]
+ pass
\ No newline at end of file