Prepare version 0.34

Adds support for the Sagemcom T210-D-r smart meter

.github/workflows/tests.yml

@@ -12,10 +12,11 @@ jobs:
     strategy:
       matrix:
         python-version:
-          - 3.6
-          - 3.7
-          - 3.8
-          - 3.9
+          - '3.6'
+          - '3.7'
+          - '3.8'
+          - '3.9'
+          - '3.10'
 
     name: Python ${{ matrix.python-version }}
     steps:

CHANGELOG.rst

@@ -1,6 +1,17 @@
 Change Log
 ----------
 
+**0.34** (2022-10-19)
+
+- Adds support for the Sagemcom T210-D-r smart meter (`pull request #110 <https://github.com/ndokter/dsmr_parser/pull/110>`_).
+
+**0.33** (2022-04-20)
+
+- Test Python 3.10 in CI + legacy badge fix (`pull request #105 <https://github.com/ndokter/dsmr_parser/pull/105>`_).
+- Update telegram_specifications.py (`pull request #106 <https://github.com/ndokter/dsmr_parser/pull/106>`_).
+- Improve compatiblity with Belgian standard (`pull request #107 <https://github.com/ndokter/dsmr_parser/pull/107>`_).
+- Improve documentation asyncio (`pull request #63 <https://github.com/ndokter/dsmr_parser/pull/63>`_).
+
 **0.32** (2022-01-04)
 
 - Support DSMR data read via RFXtrx with integrated P1 reader (`pull request #98 <https://github.com/ndokter/dsmr_parser/pull/98>`_).

README.rst

@@ -4,8 +4,8 @@ DSMR Parser
 .. image:: https://img.shields.io/pypi/v/dsmr-parser.svg
     :target: https://pypi.python.org/pypi/dsmr-parser
 
-.. image:: https://travis-ci.org/ndokter/dsmr_parser.svg?branch=master
-    :target: https://travis-ci.org/ndokter/dsmr_parser
+.. image:: https://img.shields.io/github/workflow/status/ndokter/dsmr_parser/Tests/master
+    :target: https://github.com/ndokter/dsmr_parser/actions/workflows/tests.yml
 
 A library for parsing Dutch Smart Meter Requirements (DSMR) telegram data. It
 also includes client implementation to directly read and parse smart meter data.
@@ -14,7 +14,7 @@ also includes client implementation to directly read and parse smart meter data.
 Features
 --------
 
-DSMR Parser supports DSMR versions 2, 3, 4 and 5. It has been tested with Python 3.5, 3.6, 3.7, 3.8 and 3.9.
+DSMR Parser supports DSMR versions 2, 3, 4 and 5. See for the `currently supported/tested Python versions here <https://github.com/ndokter/dsmr_parser/blob/master/.github/workflows/tests.yml#L14>`_.
 
 
 Client module usage
@@ -39,10 +39,6 @@ process because the code is blocking (not asynchronous):
      for telegram in serial_reader.read():
          print(telegram)  # see 'Telegram object' docs below
 
-**AsyncIO client**
-
-To be documented.
-
 **Socket client**
 
 Read a remote serial port (for example using ser2net) and work with the parsed telegrams.
@@ -62,6 +58,117 @@ It should be run in a separate process because the code is blocking (not asynchr
      for telegram in socket_reader.read():
          print(telegram)  # see 'Telegram object' docs below
 
+**AsyncIO client**
+
+For a test run using a tcp server (lasting 20 seconds) use the following example:
+
+.. code-block:: python
+
+    import asyncio
+    import logging
+    from dsmr_parser import obis_references
+    from dsmr_parser.clients.protocol import create_dsmr_reader, create_tcp_dsmr_reader
+
+    logging.basicConfig(level=logging.INFO, format='%(message)s')
+
+    HOST = MY_HOST
+    PORT = MY_PORT
+    DSMR_VERSION = MY_DSMR_VERSION
+
+    logger = logging.getLogger('tcpclient')
+    logger.debug("Logger created")
+
+    def printTelegram(telegram):
+        logger.info(telegram)
+
+
+    async def main():
+        try:
+            logger.debug("Getting loop")
+            loop = asyncio.get_event_loop()
+            logger.debug("Creating reader")
+            await 	create_tcp_dsmr_reader(
+                                HOST,
+                                PORT,
+                                DSMR_VERSION,
+                                printTelegram,
+                                loop
+                                )
+            logger.debug("Reader created going to sleep now")					
+            await asyncio.sleep(20)
+            logger.info('Finished run')					
+        except Exception as e:
+            logger.error("Unexpected error: "+ e)
+
+    asyncio.run(main())
+
+Note the creation of a callback function to call when a telegram is received. In this case `printTelegram`. Normally the used loop is the one running.
+
+Currently the asyncio implementation does not support returning telegram objects directly as a `read_as_object()` for async tcp is currently not implemented.
+Moreover, the telegram passed to `telegram_callback(telegram)` is already parsed. Therefore we can't feed it into the telegram constructor directly as that expects unparsed telegrams
+
+However, if we construct a mock TelegramParser that just returns the already parsed object we can work around this. An example is below:
+
+.. code-block:: python
+
+    import asyncio
+    import logging
+    #from dsmr_parser import obis_references
+    #from dsmr_parser import telegram_specifications
+    #from dsmr_parser.clients.protocol import create_dsmr_reader, create_tcp_dsmr_reader
+    #from dsmr_parser.objects import Telegram
+
+    logging.basicConfig(level=logging.INFO, format='%(message)s')
+
+    HOST = MY_HOST
+    PORT = MY_PORT
+    DSMR_VERSION = MY_DSMR_VERSION
+
+    logger = logging.getLogger('tcpclient')
+    logger.debug("Logger created")
+
+    class mockTelegramParser(object):
+
+        def parse(self, telegram):
+            return telegram
+
+    telegram_parser = mockTelegramParser()
+
+    def printTelegram(telegram):
+        try:
+            logger.info(Telegram(telegram, telegram_parser, telegram_specifications.V4))
+        except InvalidChecksumError as e:
+            logger.warning(str(e))
+        except ParseError as e:
+            logger.error('Failed to parse telegram: %s', e)
+
+
+    async def main():
+        try:
+            logger.debug("Getting loop")
+            loop = asyncio.get_event_loop()
+            logger.debug("Creating reader")
+            await 	create_tcp_dsmr_reader(
+                                HOST,
+                                PORT,
+                                DSMR_VERSION,
+                                printTelegram,
+                                loop
+                                )
+            logger.debug("Reader created going to sleep now")					
+            while True:
+                await asyncio.sleep(1)
+        except Exception as e:
+            logger.error("Unexpected error: "+ e)
+            raise
+
+    if __name__ == '__main__':
+        try:
+            asyncio.run(main())
+        except (KeyboardInterrupt, SystemExit):
+            logger.info('Closing down...')					
+        except Exception as e:
+            logger.error("Unexpected error: "+ e)
 
 Parsing module usage
 --------------------
@@ -70,39 +177,39 @@ into a dictionary.
 
 .. code-block:: python
 
-     from dsmr_parser import telegram_specifications
-     from dsmr_parser.parsers import TelegramParser
+    from dsmr_parser import telegram_specifications
+    from dsmr_parser.parsers import TelegramParser
 
-     # String is formatted in separate lines for readability.
-     telegram_str = (
-         '/ISk5\\2MT382-1000\r\n'
-         '\r\n'
-         '0-0:96.1.1(4B384547303034303436333935353037)\r\n'
-         '1-0:1.8.1(12345.678*kWh)\r\n'
-         '1-0:1.8.2(12345.678*kWh)\r\n'
-         '1-0:2.8.1(12345.678*kWh)\r\n'
-         '1-0:2.8.2(12345.678*kWh)\r\n'
-         '0-0:96.14.0(0002)\r\n'
-         '1-0:1.7.0(001.19*kW)\r\n'
-         '1-0:2.7.0(000.00*kW)\r\n'
-         '0-0:17.0.0(016*A)\r\n'
-         '0-0:96.3.10(1)\r\n'
-         '0-0:96.13.1(303132333435363738)\r\n'
-         '0-0:96.13.0(303132333435363738393A3B3C3D3E3F303132333435363738393A3B3C3D3E'
-         '3F303132333435363738393A3B3C3D3E3F303132333435363738393A3B3C3D3E3F30313233'
-         '3435363738393A3B3C3D3E3F)\r\n'
-         '0-1:96.1.0(3232323241424344313233343536373839)\r\n'
-         '0-1:24.1.0(03)\r\n'
-         '0-1:24.3.0(090212160000)(00)(60)(1)(0-1:24.2.1)(m3)\r\n'
-         '(00001.001)\r\n'
-         '0-1:24.4.0(1)\r\n'
-         '!\r\n'
-     )
+    # String is formatted in separate lines for readability.
+    telegram_str = (
+        '/ISk5\\2MT382-1000\r\n'
+        '\r\n'
+        '0-0:96.1.1(4B384547303034303436333935353037)\r\n'
+        '1-0:1.8.1(12345.678*kWh)\r\n'
+        '1-0:1.8.2(12345.678*kWh)\r\n'
+        '1-0:2.8.1(12345.678*kWh)\r\n'
+        '1-0:2.8.2(12345.678*kWh)\r\n'
+        '0-0:96.14.0(0002)\r\n'
+        '1-0:1.7.0(001.19*kW)\r\n'
+        '1-0:2.7.0(000.00*kW)\r\n'
+        '0-0:17.0.0(016*A)\r\n'
+        '0-0:96.3.10(1)\r\n'
+        '0-0:96.13.1(303132333435363738)\r\n'
+        '0-0:96.13.0(303132333435363738393A3B3C3D3E3F303132333435363738393A3B3C3D3E'
+        '3F303132333435363738393A3B3C3D3E3F303132333435363738393A3B3C3D3E3F30313233'
+        '3435363738393A3B3C3D3E3F)\r\n'
+        '0-1:96.1.0(3232323241424344313233343536373839)\r\n'
+        '0-1:24.1.0(03)\r\n'
+        '0-1:24.3.0(090212160000)(00)(60)(1)(0-1:24.2.1)(m3)\r\n'
+        '(00001.001)\r\n'
+        '0-1:24.4.0(1)\r\n'
+        '!\r\n'
+    )
 
-     parser = TelegramParser(telegram_specifications.V3)
-    
-     telegram = parser.parse(telegram_str)
-     print(telegram)  # see 'Telegram object' docs below
+    parser = TelegramParser(telegram_specifications.V3)
+
+    telegram = parser.parse(telegram_str)
+    print(telegram)  # see 'Telegram object' docs below
 
 Telegram dictionary
 -------------------
@@ -155,7 +262,7 @@ Example to get some of the values:
      gas_reading = telegram[obis_references.HOURLY_GAS_METER_READING]
 
     # See dsmr_reader.obis_references for all readable telegram values.
-    # Note that the avilable values differ per DSMR version.
+    # Note that the available values differ per DSMR version.
 
 Telegram as an Object
 ---------------------

dsmr_parser/obis_name_mapping.py

@@ -50,7 +50,9 @@ EN = {
     obis.ACTUAL_TRESHOLD_ELECTRICITY: 'ACTUAL_TRESHOLD_ELECTRICITY',
     obis.ACTUAL_SWITCH_POSITION: 'ACTUAL_SWITCH_POSITION',
     obis.VALVE_POSITION_GAS: 'VALVE_POSITION_GAS',
-    obis.BELGIUM_HOURLY_GAS_METER_READING: 'BELGIUM_HOURLY_GAS_METER_READING',
+    obis.BELGIUM_5MIN_GAS_METER_READING: 'BELGIUM_5MIN_GAS_METER_READING',
+    obis.BELGIUM_MAX_POWER_PER_PHASE: 'BELGIUM_MAX_POWER_PER_PHASE',
+    obis.BELGIUM_MAX_CURRENT_PER_PHASE: 'BELGIUM_MAX_CURRENT_PER_PHASE',
     obis.LUXEMBOURG_EQUIPMENT_IDENTIFIER: 'LUXEMBOURG_EQUIPMENT_IDENTIFIER',
     obis.Q3D_EQUIPMENT_IDENTIFIER: 'Q3D_EQUIPMENT_IDENTIFIER',
     obis.Q3D_EQUIPMENT_STATE: 'Q3D_EQUIPMENT_STATE',

dsmr_parser/obis_references.py

@@ -12,6 +12,14 @@ ELECTRICITY_USED_TARIFF_1 = r'\d-\d:1\.8\.1.+?\r\n'
 ELECTRICITY_USED_TARIFF_2 = r'\d-\d:1\.8\.2.+?\r\n'
 ELECTRICITY_DELIVERED_TARIFF_1 = r'\d-\d:2\.8\.1.+?\r\n'
 ELECTRICITY_DELIVERED_TARIFF_2 = r'\d-\d:2\.8\.2.+?\r\n'
+CURRENT_REACTIVE_EXPORTED = r'\d-\d:3\.7\.0.+?\r\n'
+ELECTRICITY_REACTIVE_IMPORTED_TOTAL = r'\d-\d:3\.8\.0.+?\r\n'
+ELECTRICITY_REACTIVE_IMPORTED_TARIFF_1 = r'\d-\d:3\.8\.1.+?\r\n'
+ELECTRICITY_REACTIVE_IMPORTED_TARIFF_2 = r'\d-\d:3\.8\.2.+?\r\n'
+CURRENT_REACTIVE_IMPORTED = r'\d-\d:4\.7\.0.+?\r\n'
+ELECTRICITY_REACTIVE_EXPORTED_TOTAL = r'\d-\d:4\.8\.0.+?\r\n'
+ELECTRICITY_REACTIVE_EXPORTED_TARIFF_1 = r'\d-\d:4\.8\.1.+?\r\n'
+ELECTRICITY_REACTIVE_EXPORTED_TARIFF_2 = r'\d-\d:4\.8\.2.+?\r\n'
 ELECTRICITY_ACTIVE_TARIFF = r'\d-\d:96\.14\.0.+?\r\n'
 EQUIPMENT_IDENTIFIER = r'\d-\d:96\.1\.1.+?\r\n'
 CURRENT_ELECTRICITY_USAGE = r'\d-\d:1\.7\.0.+?\r\n'
@@ -65,7 +73,9 @@ ELECTRICITY_IMPORTED_TOTAL = r'\d-\d:1\.8\.0.+?\r\n'  # Total imported energy re
 ELECTRICITY_EXPORTED_TOTAL = r'\d-\d:2\.8\.0.+?\r\n'  # Total exported energy register (P-)
 
 # International non generalized additions (country specific) / risk for necessary refactoring
-BELGIUM_HOURLY_GAS_METER_READING = r'\d-\d:24\.2\.3.+?\r\n'  # Different code, same format.
+BELGIUM_5MIN_GAS_METER_READING = r'\d-\d:24\.2\.3.+?\r\n'  # Different code, same format.
+BELGIUM_MAX_POWER_PER_PHASE = r'\d-\d:17\.0\.0.+?\r\n'  # Applicable when power limitation is active
+BELGIUM_MAX_CURRENT_PER_PHASE = r'\d-\d:31\.4\.0.+?\r\n'  # Applicable when current limitation is active
 LUXEMBOURG_EQUIPMENT_IDENTIFIER = r'\d-\d:42\.0\.0.+?\r\n'  # Logical device name
 Q3D_EQUIPMENT_IDENTIFIER = r'\d-\d:0\.0\.0.+?\r\n'  # Logical device name
 Q3D_EQUIPMENT_STATE = r'\d-\d:96\.5\.5.+?\r\n'  # Device state (hexadecimal)

dsmr_parser/parsers.py

@@ -1,8 +1,12 @@
 import logging
 import re
+from binascii import unhexlify
 
 from ctypes import c_ushort
 
+from dlms_cosem.connection import XDlmsApduFactory
+from dlms_cosem.protocol.xdlms import GeneralGlobalCipher
+
 from dsmr_parser.objects import MBusObject, CosemObject, ProfileGenericObject
 from dsmr_parser.exceptions import ParseError, InvalidChecksumError
 
@@ -22,14 +26,15 @@ class TelegramParser(object):
         self.telegram_specification = telegram_specification
         self.apply_checksum_validation = apply_checksum_validation
 
-    def parse(self, telegram_data):
+    def parse(self, telegram_data, encryption_key="", authentication_key=""):  # noqa: C901
         """
         Parse telegram from string to dict.
-
         The telegram str type makes python 2.x integration easier.
 
         :param str telegram_data: full telegram from start ('/') to checksum
             ('!ABCD') including line endings in between the telegram's lines
+        :param str encryption_key: encryption key
+        :param str authentication_key: authentication key
         :rtype: dict
         :returns: Shortened example:
             {
@@ -43,6 +48,38 @@ class TelegramParser(object):
         :raises InvalidChecksumError:
         """
 
+        if "general_global_cipher" in self.telegram_specification:
+            if self.telegram_specification["general_global_cipher"]:
+                enc_key = unhexlify(encryption_key)
+                auth_key = unhexlify(authentication_key)
+                telegram_data = unhexlify(telegram_data)
+                apdu = XDlmsApduFactory.apdu_from_bytes(apdu_bytes=telegram_data)
+                if apdu.security_control.security_suite != 0:
+                    logger.warning("Untested security suite")
+                if apdu.security_control.authenticated and not apdu.security_control.encrypted:
+                    logger.warning("Untested authentication only")
+                if not apdu.security_control.authenticated and not apdu.security_control.encrypted:
+                    logger.warning("Untested not encrypted or authenticated")
+                if apdu.security_control.compressed:
+                    logger.warning("Untested compression")
+                if apdu.security_control.broadcast_key:
+                    logger.warning("Untested broadcast key")
+                telegram_data = apdu.to_plain_apdu(enc_key, auth_key).decode("ascii")
+            else:
+                try:
+                    if unhexlify(telegram_data[0:2])[0] == GeneralGlobalCipher.TAG:
+                        raise RuntimeError("Looks like a general_global_cipher frame "
+                                           "but telegram specification is not matching!")
+                except Exception:
+                    pass
+        else:
+            try:
+                if unhexlify(telegram_data[0:2])[0] == GeneralGlobalCipher.TAG:
+                    raise RuntimeError(
+                        "Looks like a general_global_cipher frame but telegram specification is not matching!")
+            except Exception:
+                pass
+
         if self.apply_checksum_validation \
                 and self.telegram_specification['checksum_support']:
             self.validate_checksum(telegram_data)

dsmr_parser/telegram_specifications.py

@@ -144,10 +144,14 @@ ALL = (V2_2, V3, V4, V5)
 
 BELGIUM_FLUVIUS = deepcopy(V5)
 BELGIUM_FLUVIUS['objects'].update({
-    obis.BELGIUM_HOURLY_GAS_METER_READING: MBusParser(
+    obis.BELGIUM_5MIN_GAS_METER_READING: MBusParser(
         ValueParser(timestamp),
         ValueParser(Decimal)
-    )
+    ),
+    obis.BELGIUM_MAX_POWER_PER_PHASE: CosemParser(ValueParser(Decimal)),
+    obis.BELGIUM_MAX_CURRENT_PER_PHASE: CosemParser(ValueParser(Decimal)),
+    obis.ACTUAL_SWITCH_POSITION: CosemParser(ValueParser(str)),
+    obis.VALVE_POSITION_GAS: CosemParser(ValueParser(str)),
 })
 
 LUXEMBOURG_SMARTY = deepcopy(V5)
@@ -197,3 +201,33 @@ Q3D = {
         obis.Q3D_EQUIPMENT_SERIALNUMBER: CosemParser(ValueParser(str)),
     },
 }
+
+
+SAGEMCOM_T210_D_R = {
+    "general_global_cipher": True,
+    "checksum_support": True,
+    'objects': {
+        obis.P1_MESSAGE_HEADER: CosemParser(ValueParser(str)),
+        obis.P1_MESSAGE_TIMESTAMP: CosemParser(ValueParser(timestamp)),
+        obis.ELECTRICITY_IMPORTED_TOTAL: CosemParser(ValueParser(Decimal)),
+        obis.ELECTRICITY_USED_TARIFF_1: CosemParser(ValueParser(Decimal)),
+        obis.ELECTRICITY_USED_TARIFF_2: CosemParser(ValueParser(Decimal)),
+        obis.CURRENT_ELECTRICITY_USAGE: CosemParser(ValueParser(Decimal)),
+
+        obis.ELECTRICITY_REACTIVE_EXPORTED_TOTAL: CosemParser(ValueParser(Decimal)),
+        obis.ELECTRICITY_REACTIVE_EXPORTED_TARIFF_1: CosemParser(ValueParser(Decimal)),
+        obis.ELECTRICITY_REACTIVE_EXPORTED_TARIFF_2: CosemParser(ValueParser(Decimal)),
+        obis.CURRENT_REACTIVE_IMPORTED: CosemParser(ValueParser(Decimal)),
+
+        obis.ELECTRICITY_EXPORTED_TOTAL: CosemParser(ValueParser(Decimal)),
+        obis.ELECTRICITY_DELIVERED_TARIFF_1: CosemParser(ValueParser(Decimal)),
+        obis.ELECTRICITY_DELIVERED_TARIFF_2: CosemParser(ValueParser(Decimal)),
+        obis.CURRENT_ELECTRICITY_DELIVERY: CosemParser(ValueParser(Decimal)),
+
+        obis.ELECTRICITY_REACTIVE_IMPORTED_TOTAL: CosemParser(ValueParser(Decimal)),
+        obis.ELECTRICITY_REACTIVE_IMPORTED_TARIFF_1: CosemParser(ValueParser(Decimal)),
+        obis.ELECTRICITY_REACTIVE_IMPORTED_TARIFF_2: CosemParser(ValueParser(Decimal)),
+        obis.CURRENT_REACTIVE_EXPORTED: CosemParser(ValueParser(Decimal)),
+    }
+}
+AUSTRIA_ENERGIENETZE_STEIERMARK = SAGEMCOM_T210_D_R

setup.py

@@ -7,13 +7,14 @@ setup(
     author_email='nigel@nldr.net',
     license='MIT',
     url='https://github.com/ndokter/dsmr_parser',
-    version='0.32',
+    version='0.34',
     packages=find_packages(exclude=('test', 'test.*')),
     install_requires=[
         'pyserial>=3,<4',
         'pyserial-asyncio<1',
         'pytz',
-        'Tailer==0.4.1'
+        'Tailer==0.4.1',
+        'dlms_cosem==21.3.2'
     ],
     entry_points={
         'console_scripts': ['dsmr_console=dsmr_parser.__main__:console']

test/example_telegrams.py

@@ -169,3 +169,27 @@ TELEGRAM_ESY5Q3DA1004_V304 = (
     '  25818685\r\n'
     'DE0000000000000000000000000000003\r\n'
 )
+
+TELEGRAM_SAGEMCOM_T210_D_R = (
+    '/EST5\\253710000_A\r\n'
+    '\r\n'
+    '1-3:0.2.8(50)\r\n'
+    '0-0:1.0.0(221006155014S)\r\n'
+    '1-0:1.8.0(006545766*Wh)\r\n'
+    '1-0:1.8.1(005017120*Wh)\r\n'
+    '1-0:1.8.2(001528646*Wh)\r\n'
+    '1-0:1.7.0(000000286*W)\r\n'
+    '1-0:2.8.0(000000058*Wh)\r\n'
+    '1-0:2.8.1(000000000*Wh)\r\n'
+    '1-0:2.8.2(000000058*Wh)\r\n'
+    '1-0:2.7.0(000000000*W)\r\n'
+    '1-0:3.8.0(000000747*varh)\r\n'
+    '1-0:3.8.1(000000000*varh)\r\n'
+    '1-0:3.8.2(000000747*varh)\r\n'
+    '1-0:3.7.0(000000000*var)\r\n'
+    '1-0:4.8.0(003897726*varh)\r\n'
+    '1-0:4.8.1(002692848*varh)\r\n'
+    '1-0:4.8.2(001204878*varh)\r\n'
+    '1-0:4.7.0(000000166*var)\r\n'
+    '!7EF9\r\n'
+)

test/test_parse_sagemcom_t210_d_r.py

@@ -0,0 +1,107 @@
+from binascii import unhexlify
+from copy import deepcopy
+
+import unittest
+
+from dlms_cosem.exceptions import DecryptionError
+from dlms_cosem.protocol.xdlms import GeneralGlobalCipher
+from dlms_cosem.security import SecurityControlField, encrypt
+
+from dsmr_parser import telegram_specifications
+from dsmr_parser.exceptions import ParseError
+from dsmr_parser.parsers import TelegramParser
+from test.example_telegrams import TELEGRAM_SAGEMCOM_T210_D_R
+
+
+class TelegramParserEncryptedTest(unittest.TestCase):
+    """ Test parsing of a DSML encypted DSMR v5.x telegram. """
+    DUMMY_ENCRYPTION_KEY = "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"
+    DUMMY_AUTHENTICATION_KEY = "BBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB"
+
+    def __generate_encrypted(self, security_suite=0, authenticated=True, encrypted=True):
+        security_control = SecurityControlField(
+            security_suite=security_suite, authenticated=authenticated, encrypted=encrypted
+        )
+        encryption_key = unhexlify(self.DUMMY_ENCRYPTION_KEY)
+        authentication_key = unhexlify(self.DUMMY_AUTHENTICATION_KEY)
+        system_title = "SYSTEMID".encode("ascii")
+        invocation_counter = int.from_bytes(bytes.fromhex("10000001"), "big")
+        plain_data = TELEGRAM_SAGEMCOM_T210_D_R.encode("ascii")
+
+        encrypted = encrypt(
+            security_control=security_control,
+            key=encryption_key,
+            auth_key=authentication_key,
+            system_title=system_title,
+            invocation_counter=invocation_counter,
+            plain_text=plain_data,
+        )
+
+        full_frame = bytearray(GeneralGlobalCipher.TAG.to_bytes(1, "big", signed=False))
+        full_frame.extend(len(system_title).to_bytes(1, "big", signed=False))
+        full_frame.extend(system_title)
+        full_frame.extend([0x82])  # Length of the following length bytes
+        # https://github.com/pwitab/dlms-cosem/blob/739f81a58e5f07663a512d4a128851333a0ed5e6/dlms_cosem/a_xdr.py#L33
+
+        security_control = security_control.to_bytes()
+        invocation_counter = invocation_counter.to_bytes(4, "big", signed=False)
+        full_frame.extend((len(encrypted)
+                           + len(invocation_counter)
+                           + len(security_control)).to_bytes(2, "big", signed=False))
+        full_frame.extend(security_control)
+        full_frame.extend(invocation_counter)
+        full_frame.extend(encrypted)
+
+        return full_frame
+
+    def test_parse(self):
+        parser = TelegramParser(telegram_specifications.SAGEMCOM_T210_D_R)
+        result = parser.parse(self.__generate_encrypted().hex(),
+                              self.DUMMY_ENCRYPTION_KEY,
+                              self.DUMMY_AUTHENTICATION_KEY)
+        self.assertEqual(len(result), 18)
+
+    def test_damaged_frame(self):
+        # If the frame is damaged decrypting fails (crc is technically not needed)
+        parser = TelegramParser(telegram_specifications.SAGEMCOM_T210_D_R)
+
+        generated = self.__generate_encrypted()
+        generated[150] = 0x00
+        generated = generated.hex()
+
+        with self.assertRaises(DecryptionError):
+            parser.parse(generated, self.DUMMY_ENCRYPTION_KEY, self.DUMMY_AUTHENTICATION_KEY)
+
+    def test_plain(self):
+        # If a plain request is parsed with "general_global_cipher": True it fails
+        parser = TelegramParser(telegram_specifications.SAGEMCOM_T210_D_R)
+
+        with self.assertRaises(Exception):
+            parser.parse(TELEGRAM_SAGEMCOM_T210_D_R, self.DUMMY_ENCRYPTION_KEY, self.DUMMY_AUTHENTICATION_KEY)
+
+    def test_general_global_cipher_not_specified(self):
+        # If a GGC frame is detected but general_global_cipher is not set it fails
+        parser = TelegramParser(telegram_specifications.SAGEMCOM_T210_D_R)
+        parser = deepcopy(parser)  # We do not want to change the module value
+        parser.telegram_specification['general_global_cipher'] = False
+
+        with self.assertRaises(ParseError):
+            parser.parse(self.__generate_encrypted().hex(), self.DUMMY_ENCRYPTION_KEY, self.DUMMY_AUTHENTICATION_KEY)
+
+    def test_only_encrypted(self):
+        # Not implemented by dlms_cosem
+        parser = TelegramParser(telegram_specifications.SAGEMCOM_T210_D_R)
+
+        only_auth = self.__generate_encrypted(0, authenticated=False, encrypted=True).hex()
+
+        with self.assertRaises(ValueError):
+            parser.parse(only_auth, self.DUMMY_ENCRYPTION_KEY)
+
+    def test_only_auth(self):
+        # Not implemented by dlms_cosem
+        parser = TelegramParser(telegram_specifications.SAGEMCOM_T210_D_R)
+
+        only_auth = self.__generate_encrypted(0, authenticated=True, encrypted=False).hex()
+
+        with self.assertRaises(ValueError):
+            parser.parse(only_auth, authentication_key=self.DUMMY_AUTHENTICATION_KEY)

tox.ini

@@ -5,6 +5,7 @@ deps=
   pylama
   pytest-asyncio
   pytest-mock
+  dlms_cosem
 commands=
   py.test --cov=dsmr_parser test {posargs}
   pylama dsmr_parser test