From b4ac8c3bab2bc4741b4e455a3a887c14639428b1 Mon Sep 17 00:00:00 2001 From: Thomas Hoek Date: Mon, 30 Jun 2025 13:36:54 +0200 Subject: [PATCH 1/4] draft: asyncio support --- dlms_cosem/async_client.py | 270 +++++++++++++++++++ dlms_cosem/asyncio.py | 223 +++++++++++++++ dlms_cosem/client.py | 15 +- tests/test_async_tcp_transport.py | 86 ++++++ tests/test_clients/test_async_dlms_client.py | 57 ++++ 5 files changed, 645 insertions(+), 6 deletions(-) create mode 100644 dlms_cosem/async_client.py create mode 100644 dlms_cosem/asyncio.py create mode 100644 tests/test_async_tcp_transport.py create mode 100644 tests/test_clients/test_async_dlms_client.py diff --git a/dlms_cosem/async_client.py b/dlms_cosem/async_client.py new file mode 100644 index 0000000..576a259 --- /dev/null +++ b/dlms_cosem/async_client.py @@ -0,0 +1,270 @@ +import contextlib +from typing import Optional, List +from collections.abc import AsyncGenerator + +import attr +import structlog + +from dlms_cosem import cosem, dlms_data, enumerations, exceptions, state, utils +from dlms_cosem.security import AuthenticationMethodManager +from dlms_cosem.asyncio import AsyncDlmsTransport +from dlms_cosem.connection import DlmsConnection, DlmsConnectionSettings +from dlms_cosem.cosem.selective_access import RangeDescriptor +from dlms_cosem.protocol import acse, xdlms +from dlms_cosem.protocol.xdlms import ConfirmedServiceError + +LOG = structlog.get_logger() + + +class DataResultError(Exception): + """ Error retrieveing data""" + + +class ActionError(Exception): + """Error performing an action""" + + +class HLSError(Exception): + """error in HLS procedure""" + + +@attr.s(auto_attribs=True) +class AsyncDlmsClient: + transport: AsyncDlmsTransport + authentication: AuthenticationMethodManager + encryption_key: Optional[bytes] = attr.ib(default=None) + authentication_key: Optional[bytes] = attr.ib(default=None) + security_suite: Optional[int] = attr.ib(default=0) + dedicated_ciphering: bool = attr.ib(default=False) + block_transfer: bool = attr.ib(default=False) + max_pdu_size: int = attr.ib(default=65535) + client_system_title: Optional[bytes] = attr.ib(default=None) + client_initial_invocation_counter: int = attr.ib(default=0) + meter_initial_invocation_counter: int = attr.ib(default=0) + timeout: int = attr.ib(default=10) + connection_settings: Optional[DlmsConnectionSettings] = attr.ib(default=None) + + dlms_connection: DlmsConnection = attr.ib( + default=attr.Factory( + lambda self: DlmsConnection( + client_system_title=self.client_system_title, + authentication=self.authentication, + global_encryption_key=self.encryption_key, + global_authentication_key=self.authentication_key, + use_dedicated_ciphering=self.dedicated_ciphering, + use_block_transfer=self.block_transfer, + security_suite=self.security_suite, + max_pdu_size=self.max_pdu_size, + client_invocation_counter=self.client_initial_invocation_counter, + meter_invocation_counter=self.meter_initial_invocation_counter, + settings=self.connection_settings, + ), + takes_self=True, + ) + ) + + @contextlib.asynccontextmanager + async def session(self) -> AsyncGenerator["AsyncDlmsClient", None]: + await self.connect() + try: + await self.associate() + yield self + await self.release_association() + finally: + await self.disconnect() + + async def get( + self, + cosem_attribute: cosem.CosemAttribute, + access_descriptor: Optional[RangeDescriptor] = None, + ) -> bytes: + await self.send( + xdlms.GetRequestNormal( + cosem_attribute=cosem_attribute, access_selection=access_descriptor + ) + ) + all_data_received = False + data = bytearray() + while not all_data_received: + get_response = self.next_event() + if isinstance(get_response, xdlms.GetResponseNormal): + data.extend(get_response.data) + all_data_received = True + continue + if isinstance(get_response, xdlms.GetResponseWithBlock): + data.extend(get_response.data) + await self.send( + xdlms.GetRequestNext( + invoke_id_and_priority=get_response.invoke_id_and_priority, + block_number=get_response.block_number, + ) + ) + continue + if isinstance(get_response, xdlms.GetResponseLastBlock): + data.extend(get_response.data) + all_data_received = True + continue + + if isinstance(get_response, xdlms.GetResponseLastBlockWithError): + raise DataResultError( + f"Error in blocktransfer of GET response: {get_response.error!r}" + ) + + if isinstance(get_response, xdlms.GetResponseNormalWithError): + raise DataResultError( + f"Could not perform GET request: {get_response.error!r}" + ) + + return bytes(data) + + async def get_many( + self, cosem_attributes_with_selection: List[cosem.CosemAttributeWithSelection] + ): + """ + Make a GET.WITH_LIST call. Get many items in one request. + """ + out = xdlms.GetRequestWithList( + cosem_attributes_with_selection=cosem_attributes_with_selection + ) + await self.send(out) + response = self.next_event() + if isinstance(response, xdlms.ExceptionResponse): + raise exceptions.DlmsClientException( + f"Received an Exception response with state error: " + f"{response.state_error.name} and service error: " + f"{response.service_error.name}" + ) + return response + + async def set(self, cosem_attribute: cosem.CosemAttribute, data: bytes): + await self.send(xdlms.SetRequestNormal(cosem_attribute=cosem_attribute, data=data)) + return self.next_event() + + async def action(self, method: cosem.CosemMethod, data: bytes): + await self.send(xdlms.ActionRequestNormal(cosem_method=method, data=data)) + response = self.next_event() + + if isinstance(response, xdlms.ActionResponseNormalWithError): + raise ActionError(response.error.name) + elif isinstance(response, xdlms.ActionResponseNormalWithData): + if response.status != enumerations.ActionResultStatus.SUCCESS: + raise ActionError(f"Unsuccessful ActionRequest: {response.status.name}") + return response.data + else: + if response.status != enumerations.ActionResultStatus.SUCCESS: + raise ActionError(f"Unsuccessful ActionRequest: {response.status.name}") + return + + async def associate( + self, + association_request: Optional[acse.ApplicationAssociationRequest] = None, + ) -> acse.ApplicationAssociationResponse: + + # the aarq can be overridden or the standard one from the connection is used. + aarq = association_request or self.dlms_connection.get_aarq() + + await self.send(aarq) + response = self.next_event() + # we could have received an exception from the meter. + if isinstance(response, xdlms.ExceptionResponse): + raise exceptions.DlmsClientException( + f"DLMS Exception: {response.state_error!r}:{response.service_error!r}" + ) + # the association might not be accepted by the meter + if isinstance(response, acse.ApplicationAssociationResponse): + if response.result is not enumerations.AssociationResult.ACCEPTED: + # there could be an error suppled with the reject. + extra_error = None + if response.user_information: + if isinstance( + response.user_information.content, ConfirmedServiceError + ): + extra_error = response.user_information.content.error + raise exceptions.DlmsClientException( + f"Unable to perform Association: {response.result!r} and " + f"{response.result_source_diagnostics!r}, extra info: {extra_error}" + ) + else: + raise exceptions.LocalDlmsProtocolError( + "Did not receive an AARE after sending AARQ" + ) + + if self.should_send_hls_reply(): + + # TODO: wrap hls logic in method + try: + hls_response = await self.send_hls_reply() + except ActionError as e: + raise HLSError from e + + if not hls_response: + raise HLSError("No HLS data in response") + + hls_data = utils.parse_as_dlms_data(hls_response) + + if not hls_data: + raise HLSError("Did not receive any HLS response data") + + if not self.dlms_connection.authentication.hls_meter_data_is_valid( + hls_data, self.dlms_connection + ): + raise HLSError( + f"Meter did not respond with correct challenge calculation" + ) + + return response + + def should_send_hls_reply(self) -> bool: + return ( + self.dlms_connection.state.current_state + == state.SHOULD_SEND_HLS_SEVER_CHALLENGE_RESULT + ) + + async def send_hls_reply(self) -> Optional[bytes]: + return await self.action( + method=cosem.CosemMethod( + enumerations.CosemInterface.ASSOCIATION_LN, + cosem.Obis(0, 0, 40, 0, 0), + 1, + ), + data=dlms_data.OctetStringData( + self.dlms_connection.authentication.hls_generate_reply_data( + self.dlms_connection + ) + ).to_bytes(), + ) + + async def release_association(self) -> Optional[acse.ReleaseResponse]: + + rlrq = self.dlms_connection.get_rlrq() + try: + await self.send(rlrq) + rlre = self.next_event() + return rlre + except exceptions.NoRlrqRlreError: + return None + + async def connect(self): + await self.transport.connect() + + async def disconnect(self): + await self.transport.disconnect() + + async def send(self, *events): + for event in events: + data = self.dlms_connection.send(event) + response_bytes = await self.transport.send_request(data) + self.dlms_connection.receive_data(response_bytes) + + def next_event(self): + event = self.dlms_connection.next_event() + return event + + @property + def client_invocation_counter(self) -> int: + return self.dlms_connection.client_invocation_counter + + @client_invocation_counter.setter + def client_invocation_counter(self, ic: int): + self.dlms_connection.client_invocation_counter = ic + diff --git a/dlms_cosem/asyncio.py b/dlms_cosem/asyncio.py new file mode 100644 index 0000000..e7007e2 --- /dev/null +++ b/dlms_cosem/asyncio.py @@ -0,0 +1,223 @@ +from __future__ import annotations # noqa + +import asyncio +import socket +import sys +from typing import Optional, Tuple + +from dlms_cosem.protocol.wrappers import WrapperHeader, WrapperProtocolDataUnit + +if sys.version_info < (3, 8): + from typing_extensions import Protocol +else: + from typing import Protocol + +import attr + +from dlms_cosem import exceptions + +from typing import Optional, TYPE_CHECKING + +import structlog + +if TYPE_CHECKING: + pass + +LOG = structlog.get_logger() + +LLC_COMMAND_HEADER = b"\xe6\xe6\x00" +LLC_RESPONSE_HEADER = b"\xe6\xe7\x00" + + +class AsyncIoImplementation: + writer: Optional[asyncio.StreamWriter] = None + reader: Optional[asyncio.StreamReader] = None + + async def connect(self) -> None: ... + + async def disconnect(self) -> None: ... + + async def send(self, data: bytes) -> None: ... + + async def recv(self, amount: int) -> bytes: ... + + async def recv_until(self, end: bytes) -> bytes: ... + + +class AsyncDlmsTransport(Protocol): + """ + Protocol for a class that should be used for transport. + """ + + client_logical_address: int + server_logical_address: int + io: AsyncIoImplementation + timeout: int + + async def connect(self) -> None: ... + + async def disconnect(self) -> None: ... + + async def send_request(self, bytes_to_send: bytes) -> bytes: ... + + +@attr.s(auto_attribs=True) +class AsyncTcpTransport(AsyncDlmsTransport): + """ + A TCP transport. + """ + + client_logical_address: int + server_logical_address: int + io: AsyncIoImplementation + timeout: int = attr.ib(default=10) + + def wrap(self, bytes_to_wrap: bytes) -> bytes: + """ + When sending data over TCP or UDP it is necessary to wrap the data in the in + the DLMS IP wrapper. This is so the server (meter) knows where the data is + intended and how long the message is since there is no "final/poll" bit like + in HDLC. + """ + header = WrapperHeader( + source_wport=self.client_logical_address, + destination_wport=self.server_logical_address, + length=len(bytes_to_wrap), + ) + return WrapperProtocolDataUnit(bytes_to_wrap, header).to_bytes() + + async def connect(self): + await self.io.connect() + + async def disconnect(self): + await self.io.disconnect() + + async def send_request(self, bytes_to_send: bytes) -> bytes: + """ + Sends a whole DLMS APDU wrapped in the DLMS IP Wrapper. + """ + wrapped = self.wrap(bytes_to_send) + LOG.debug("Sending data", data=wrapped, transport=self) + await self.io.send(self.wrap(bytes_to_send)) + + return await self.recv_response() + + async def recv_response(self) -> bytes: + """ + Receives a whole DLMS APDU. Gets the total length from the DLMS IP Wrapper. + """ + header_data = await self.io.recv(8) + header = WrapperHeader.from_bytes(header_data) + data = await self.io.recv(header.length) + + LOG.debug("Received data", data=header_data + data, transport=self) + + return data + + + +@attr.s(auto_attribs=True) +class AsyncTcpIO(AsyncIoImplementation): + """ + A TCP transport using Blocking I/O. + """ + + host: str + port: int + timeout: int = attr.ib(default=10) + #TODO implement SSL support + #ssl_handshake_timeout: int = attr.ib(default=10) + #ssl_shutdown_timeout: int = attr.ib(default=10) + + reader: Optional[asyncio.StreamReader] = attr.ib(default=None) + writer: Optional[asyncio.StreamWriter] = attr.ib(default=None) + + @property + def is_connected(self) -> bool: + """ + Returns True if the transport is connected. + """ + return self.writer is not None and not self.writer.is_closing() + + @property + def address(self) -> Tuple[str, int]: + return self.host, self.port + + async def connect(self): + """ + Create a new socket and set it on the transport + """ + if self.writer: + raise RuntimeError(f"There is already an active connection to {self.address}") + + try: + self.reader, self.writer = await asyncio.wait_for( + asyncio.open_connection( + host=self.host, + port=self.port, + #TODO implement SSL support + #ssl=True, + #ssl_handshake_timeout=self.ssl_handshake_timeout, + #ssl_shutdown_timeout=self.ssl_shutdown_timeout, + ), + timeout=self.timeout, + ) + except ( + OSError, + IOError, + socket.timeout, + socket.error, + ConnectionRefusedError, + ) as e: + raise exceptions.CommunicationError("Unable to connect socket") from e + #LOG.info(f"Connected to {self.address}") + + async def disconnect(self): + """ + Close socket and remove it from the transport. No-op if the socket is already + closed. + """ + # only disconnect if there is a writer. + if not self.writer or self.writer.is_closing(): + return + try: + self.writer.close() + await self.writer.wait_closed() + except (OSError, IOError, socket.timeout, socket.error) as e: + raise exceptions.CommunicationError from e + finally: + self.writer = self.reader = None + #LOG.info(f"Connection to {self.address} is closed") + + async def send(self, data: bytes): + """ + Sends a whole DLMS APDU wrapped in the DLMS IP Wrapper. + """ + if not self.writer or self.writer.is_closing(): + raise RuntimeError("TCP transport not connected.") + try: + self.writer.write(data) + await self.writer.drain() + except (OSError, IOError, socket.timeout, socket.error) as e: + raise exceptions.CommunicationError("Could no send data") from e + + async def recv(self, amount: int = 1) -> bytes: + """ + Receives a whole DLMS APDU. Gets the total length from the DLMS IP Wrapper. + """ + if not self.reader: + raise RuntimeError("TCP transport not connected.") + data = b"" + while len(data) < amount: + try: + data += await self.reader.read(amount - len(data)) + except (OSError, IOError, socket.timeout, socket.error) as e: + raise exceptions.CommunicationError("Could not receive data") from e + return data + + async def recv_until(self, end: bytes) -> bytes: + data = b"" + while not data.endswith(end): + data += await self.recv() + return data + diff --git a/dlms_cosem/client.py b/dlms_cosem/client.py index b07a074..1d37128 100644 --- a/dlms_cosem/client.py +++ b/dlms_cosem/client.py @@ -1,5 +1,6 @@ import contextlib -from typing import * +from typing import Optional, List +from collections.abc import Generator import attr import structlog @@ -63,12 +64,14 @@ class DlmsClient: ) @contextlib.contextmanager - def session(self) -> "DlmsClient": + def session(self) -> Generator["DlmsClient", None, None]: self.connect() - self.associate() - yield self - self.release_association() - self.disconnect() + try: + self.associate() + yield self + self.release_association() + finally: + self.disconnect() def get( self, diff --git a/tests/test_async_tcp_transport.py b/tests/test_async_tcp_transport.py new file mode 100644 index 0000000..27f6337 --- /dev/null +++ b/tests/test_async_tcp_transport.py @@ -0,0 +1,86 @@ +import pytest + +from dlms_cosem.asyncio import AsyncTcpIO, AsyncTcpTransport +from dlms_cosem.exceptions import CommunicationError + + +import socket + +@pytest.fixture +def open_socket(): + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: + # Bind it to address 0.0.0.0 with port 0 + s.bind(('0.0.0.0', 0)) + # Get the port number assigned by the system + s.listen(1) + yield s + + +@pytest.fixture +def open_port(open_socket): + # Return the port number of the open socket + return open_socket.getsockname()[1] + +@pytest.fixture +def closed_port(open_socket): + port = open_socket.getsockname()[1] + open_socket.close() + return port + + +class TestAsyncAsyncTcpTransport: + + host = "localhost" + client_logical_address = 1 + server_logical_address = 1 + + @pytest.mark.asyncio + async def test_can_connect(self, open_port): + io = AsyncTcpIO(host=self.host, port=open_port) + transport = AsyncTcpTransport( + self.client_logical_address, self.server_logical_address, io + ) + await transport.connect() + assert transport.io.writer and not transport.io.writer.is_closing() + assert transport.io.reader + + @pytest.mark.asyncio + async def test_connect_on_connected_raises(self, open_port): + io = AsyncTcpIO(host=self.host, port=open_port) + transport = AsyncTcpTransport( + self.client_logical_address, self.server_logical_address, io + ) + await transport.connect() + with pytest.raises(RuntimeError): + await transport.connect() + + @pytest.mark.asyncio + async def test_cant_connect_raises_communications_error(self, closed_port): + io = AsyncTcpIO(host=self.host, port=closed_port) + transport = AsyncTcpTransport( + self.client_logical_address, self.server_logical_address, io + ) + with pytest.raises(CommunicationError): + await transport.connect() + + @pytest.mark.asyncio + async def test_disconnect(self, open_port): + io = AsyncTcpIO(host=self.host, port=open_port) + transport = AsyncTcpTransport( + self.client_logical_address, self.server_logical_address, io + ) + await transport.connect() + await transport.disconnect() + assert transport.io.writer is None or transport.io.writer.is_closing() + assert not transport.io.reader + + @pytest.mark.asyncio + async def test_disconnect_is_noop_if_disconnected(self, open_port): + io = AsyncTcpIO(host=self.host, port=open_port) + transport = AsyncTcpTransport( + self.client_logical_address, self.server_logical_address, io + ) + await transport.connect() + await transport.disconnect() + await transport.disconnect() + assert transport.io.writer is None or transport.io.writer.is_closing() diff --git a/tests/test_clients/test_async_dlms_client.py b/tests/test_clients/test_async_dlms_client.py new file mode 100644 index 0000000..4d6d8b6 --- /dev/null +++ b/tests/test_clients/test_async_dlms_client.py @@ -0,0 +1,57 @@ +from dlms_cosem.connection import DlmsConnectionSettings +from dlms_cosem.security import NoSecurityAuthentication +from dlms_cosem.async_client import AsyncDlmsClient +from dlms_cosem.asyncio import AsyncTcpTransport, AsyncTcpIO + + +class TestAsyncDlmsClient: + def test_client_invocation_counter_property(self): + transport = AsyncTcpTransport( + io=AsyncTcpIO(host="localhost", port=4059), + client_logical_address=1, + server_logical_address=1, + ) + client = AsyncDlmsClient( + client_initial_invocation_counter=500, + transport=transport, + authentication=NoSecurityAuthentication(), + ) + + assert client.client_invocation_counter == 500 + + def test_client_invocation_counter_setter(self): + transport = AsyncTcpTransport( + io=AsyncTcpIO(host="localhost", port=4059), + client_logical_address=1, + server_logical_address=1, + ) + client = AsyncDlmsClient( + client_initial_invocation_counter=500, + transport=transport, + authentication=NoSecurityAuthentication(), + ) + client.client_invocation_counter = 1000 + assert client.client_invocation_counter == 1000 + assert client.dlms_connection.client_invocation_counter == 1000 + + +class TestAsyncDlmsClientWithConnectionSettings: + + def test_can_get_settings_from_client(self): + settings = DlmsConnectionSettings(empty_system_title_in_general_glo_ciphering=True) + + transport = AsyncTcpTransport( + io=AsyncTcpIO(host="localhost", port=4059), + client_logical_address=1, + server_logical_address=1, + ) + + + client = AsyncDlmsClient( + client_initial_invocation_counter=500, + transport=transport, + authentication=NoSecurityAuthentication(), + connection_settings=settings + ) + + assert client.dlms_connection.settings.empty_system_title_in_general_glo_ciphering == True From 7f78301539361d4ab19fe1fe9cc808a60b9e2be1 Mon Sep 17 00:00:00 2001 From: Thomas Hoek Date: Mon, 30 Jun 2025 15:22:34 +0200 Subject: [PATCH 2/4] revert: restore dlms_cosem/client.py to original to prevent merge conflicts later on --- dlms_cosem/client.py | 15 ++++++--------- 1 file changed, 6 insertions(+), 9 deletions(-) diff --git a/dlms_cosem/client.py b/dlms_cosem/client.py index 1d37128..b07a074 100644 --- a/dlms_cosem/client.py +++ b/dlms_cosem/client.py @@ -1,6 +1,5 @@ import contextlib -from typing import Optional, List -from collections.abc import Generator +from typing import * import attr import structlog @@ -64,14 +63,12 @@ class DlmsClient: ) @contextlib.contextmanager - def session(self) -> Generator["DlmsClient", None, None]: + def session(self) -> "DlmsClient": self.connect() - try: - self.associate() - yield self - self.release_association() - finally: - self.disconnect() + self.associate() + yield self + self.release_association() + self.disconnect() def get( self, From a065e855bc68c61ac41dc22741b5af32e425b00e Mon Sep 17 00:00:00 2001 From: Thomas Hoek Date: Mon, 30 Jun 2025 15:35:03 +0200 Subject: [PATCH 3/4] feat: dlms with async tcp example --- dlms_cosem/async_client.py | 4 +- dlms_cosem/asyncio.py | 8 +- examples/dlms_with_async_tcp_example.py | 165 ++++++++++++++++++++++++ 3 files changed, 171 insertions(+), 6 deletions(-) create mode 100644 examples/dlms_with_async_tcp_example.py diff --git a/dlms_cosem/async_client.py b/dlms_cosem/async_client.py index 576a259..2e4a288 100644 --- a/dlms_cosem/async_client.py +++ b/dlms_cosem/async_client.py @@ -6,7 +6,7 @@ import structlog from dlms_cosem import cosem, dlms_data, enumerations, exceptions, state, utils -from dlms_cosem.security import AuthenticationMethodManager +from dlms_cosem.security import AuthenticationMethodManager, NoSecurityAuthentication from dlms_cosem.asyncio import AsyncDlmsTransport from dlms_cosem.connection import DlmsConnection, DlmsConnectionSettings from dlms_cosem.cosem.selective_access import RangeDescriptor @@ -31,7 +31,7 @@ class HLSError(Exception): @attr.s(auto_attribs=True) class AsyncDlmsClient: transport: AsyncDlmsTransport - authentication: AuthenticationMethodManager + authentication: AuthenticationMethodManager = attr.ib(default=NoSecurityAuthentication()) encryption_key: Optional[bytes] = attr.ib(default=None) authentication_key: Optional[bytes] = attr.ib(default=None) security_suite: Optional[int] = attr.ib(default=0) diff --git a/dlms_cosem/asyncio.py b/dlms_cosem/asyncio.py index e7007e2..d606953 100644 --- a/dlms_cosem/asyncio.py +++ b/dlms_cosem/asyncio.py @@ -64,7 +64,7 @@ async def send_request(self, bytes_to_send: bytes) -> bytes: ... @attr.s(auto_attribs=True) class AsyncTcpTransport(AsyncDlmsTransport): """ - A TCP transport. + An async TCP transport. """ client_logical_address: int @@ -119,7 +119,7 @@ async def recv_response(self) -> bytes: @attr.s(auto_attribs=True) class AsyncTcpIO(AsyncIoImplementation): """ - A TCP transport using Blocking I/O. + A TCP transport using asyncio """ host: str @@ -170,7 +170,7 @@ async def connect(self): ConnectionRefusedError, ) as e: raise exceptions.CommunicationError("Unable to connect socket") from e - #LOG.info(f"Connected to {self.address}") + LOG.info(f"Connected to {self.address}") async def disconnect(self): """ @@ -187,7 +187,7 @@ async def disconnect(self): raise exceptions.CommunicationError from e finally: self.writer = self.reader = None - #LOG.info(f"Connection to {self.address} is closed") + LOG.info(f"Connection to {self.address} is closed") async def send(self, data: bytes): """ diff --git a/examples/dlms_with_async_tcp_example.py b/examples/dlms_with_async_tcp_example.py new file mode 100644 index 0000000..c881603 --- /dev/null +++ b/examples/dlms_with_async_tcp_example.py @@ -0,0 +1,165 @@ +import asyncio +import logging +from pprint import pprint +from time import sleep +import sys + +from dateutil import parser as dateparser + +from dlms_cosem import a_xdr, cosem, enumerations +from dlms_cosem.security import ( + HighLevelSecurityGmacAuthentication, +) +from dlms_cosem.async_client import AsyncDlmsClient +from dlms_cosem.asyncio import AsyncTcpIO, AsyncTcpTransport +from dlms_cosem.cosem import selective_access +from dlms_cosem.cosem.selective_access import RangeDescriptor +from dlms_cosem.parsers import ProfileGenericBufferParser +from dlms_cosem.protocol.xdlms.conformance import Conformance + +# set up logging so you get a bit nicer printout of what is happening. +logging.basicConfig( + level=logging.DEBUG, + format="%(asctime)s,%(msecs)d : %(levelname)s : %(message)s", + datefmt="%H:%M:%S", +) + +c = Conformance( + general_protection=False, + general_block_transfer=False, + delta_value_encoding=False, + attribute_0_supported_with_set=False, + priority_management_supported=False, + attribute_0_supported_with_get=False, + block_transfer_with_get_or_read=True, + block_transfer_with_set_or_write=False, + block_transfer_with_action=True, + multiple_references=True, + data_notification=False, + access=False, + get=True, + set=True, + selective_access=True, + event_notification=False, + action=True, +) + +encryption_key = bytes.fromhex("990EB3136F283EDB44A79F15F0BFCC21") +authentication_key = bytes.fromhex("EC29E2F4BD7D697394B190827CE3DD9A") +auth = enumerations.AuthenticationMechanism.HLS_GMAC + + +async def main(): + """Example of using the AsyncDlmsClient with TCP transport. + + Establish + 1) an unsecured session to read public attributes and + 2) a secured session to read attributes that require authentication """ + if len(sys.argv) < 3: + print("Usage: python dlms_with_async_tcp_example.py <") + sys.exit(1) + + host = sys.argv[1] + port = int(sys.argv[2]) + + tcp_io = AsyncTcpIO(host=host, port=port) + public_tcp_transport = AsyncTcpTransport( + client_logical_address=16, + server_logical_address=1, + io=tcp_io, + ) + public_client = AsyncDlmsClient( + transport=public_tcp_transport + ) + + # Try a session without security + async with public_client.session() as client: + + response_data = await client.get( + cosem.CosemAttribute( + interface=enumerations.CosemInterface.DATA, + instance=cosem.Obis(0, 0, 0x2B, 1, 0), + attribute=2, + ) + ) + data_decoder = a_xdr.AXdrDecoder( + encoding_conf=a_xdr.EncodingConf( + attributes=[a_xdr.Sequence(attribute_name="data")] + ) + ) + invocation_counter = data_decoder.decode(response_data)["data"] + print(f"meter_initial_invocation_counter = {invocation_counter}") + + # we are not reusing the socket as of now. We just need to give the meter some time to + # close the connection on its side + sleep(2) + + tcp_io = AsyncTcpIO(host=host, port=port) + management_tcp_transport = AsyncTcpTransport( + client_logical_address=1, + server_logical_address=1, + io=tcp_io, + ) + + management_client = AsyncDlmsClient( + transport=management_tcp_transport, + authentication=HighLevelSecurityGmacAuthentication(challenge_length=32), + encryption_key=encryption_key, + authentication_key=authentication_key, + client_initial_invocation_counter=invocation_counter + 1, + ) + + # Try a session with security + async with management_client.session() as client: + + profile = await client.get( + cosem.CosemAttribute( + interface=enumerations.CosemInterface.PROFILE_GENERIC, + instance=cosem.Obis(1, 0, 99, 1, 0), + attribute=2, + ), + access_descriptor=RangeDescriptor( + restricting_object=selective_access.CaptureObject( + cosem_attribute=cosem.CosemAttribute( + interface=enumerations.CosemInterface.CLOCK, + instance=cosem.Obis.from_string("0.0.1.0.0.255"), + attribute=2, + ), + data_index=0, + ), + from_value=dateparser.parse("2022-01-01T00:00:00-02:00"), + to_value=dateparser.parse("2022-01-02T00:00:00-01:00"), + ), + ) + + parser = ProfileGenericBufferParser( + capture_objects=[ + cosem.CosemAttribute( + interface=enumerations.CosemInterface.CLOCK, + instance=cosem.Obis(0, 0, 1, 0, 0, 255), + attribute=2, + ), + cosem.CosemAttribute( + interface=enumerations.CosemInterface.DATA, + instance=cosem.Obis(0, 0, 96, 10, 1, 255), + attribute=2, + ), + cosem.CosemAttribute( + interface=enumerations.CosemInterface.REGISTER, + instance=cosem.Obis(1, 0, 1, 8, 0, 255), + attribute=2, + ), + cosem.CosemAttribute( + interface=enumerations.CosemInterface.REGISTER, + instance=cosem.Obis(1, 0, 2, 8, 0, 255), + attribute=2, + ), + ], + capture_period=60, + ) + result = parser.parse_bytes(profile) + pprint(result) + + +if __name__ == '__main__': + asyncio.run(main()) From 9296b3452e00587eac71d1cb161822bdffe31265 Mon Sep 17 00:00:00 2001 From: Thomas Hoek Date: Tue, 1 Jul 2025 12:41:30 +0200 Subject: [PATCH 4/4] mod(tests): add pytest-asyncio dependency for async unit tests --- setup.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/setup.py b/setup.py index 57fb915..cf8c769 100644 --- a/setup.py +++ b/setup.py @@ -30,7 +30,7 @@ ] DOC_PACKAGES = ["mkdocs", "mkdocs-material"] -TEST_PACKAGES = ["pytest", "pytest-cov", "pytest-sugar"] +TEST_PACKAGES = ["pytest", "pytest-cov", "pytest-sugar", "pytest-asyncio"] DEV_PACKAGES = ["pre-commit"] + DOC_PACKAGES + TEST_PACKAGES EXTRAS = {