Merged master into autodetecttelegram

.gitignore

@@ -3,7 +3,6 @@
 .tox
 .cache
 .venv
-.history
 *.egg-info
 /.project
 /.pydevproject

CHANGELOG.rst

@@ -1,6 +1,9 @@
 Change Log
 ----------
 
+**1.5.0** (2024-08-25)
+- Allow the telegram specification to optionally be autodetected (`PR #87 <https://github.com/ndokter/dsmr_parser/pull/87>`_ by `clonerswords <https://github.com/clonerswords>`_)
+
 **1.4.2** (2024-07-14)
 
 - Bump Github Actions to latest versions in favor of Node deprecations (`PR #159 <https://github.com/ndokter/dsmr_parser/pull/159>`_ by `dennissiemensma <https://github.com/dennissiemensma>`_)
@@ -167,7 +170,7 @@ Remove deprecated asyncio coroutine decorator (`PR #76 <https://github.com/ndokt
 **0.12** (2018-09-23)
 
 - Add serial settings for DSMR v5.0 (`PR #31 <https://github.com/ndokter/dsmr_parser/pull/31>`_).
-- Lux-creos-obis-1.8.0 (`PR #32 <https://github.com/ndokter/dsmr_parser/pull/32>`_). 
+- Lux-creos-obis-1.8.0 (`PR #32 <https://github.com/ndokter/dsmr_parser/pull/32>`_).
 
 **0.11** (2017-09-18)
 

dsmr_parser/exceptions.py

@@ -4,3 +4,7 @@ class ParseError(Exception):
 
 class InvalidChecksumError(ParseError):
     pass
+
+
+class TelegramSpecificationMatchError(ParseError):
+    pass

dsmr_parser/parsers.py

@@ -9,7 +9,7 @@ from dlms_cosem.connection import XDlmsApduFactory
 from dlms_cosem.protocol.xdlms import GeneralGlobalCipher
 
 from dsmr_parser.objects import MBusObject, MBusObjectPeak, CosemObject, ProfileGenericObject, Telegram
-from dsmr_parser.exceptions import ParseError, InvalidChecksumError
+from dsmr_parser.exceptions import ParseError, InvalidChecksumError, TelegramSpecificationMatchError
 from dsmr_parser.value_types import timestamp
 
 logger = logging.getLogger(__name__)
@@ -18,21 +18,17 @@ logger = logging.getLogger(__name__)
 class TelegramParser(object):
     crc16_tab = []
 
-    def __init__(self, telegram_specification, apply_checksum_validation=True):
+    def __init__(self, telegram_specification=None, apply_checksum_validation=True):
         """
-        :param telegram_specification: determines how the telegram is parsed
+        :param telegram_specification: determines how the telegram is parsed.
+            Will attempt to autodetect if omitted.
         :param apply_checksum_validation: validate checksum if applicable for
             telegram DSMR version (v4 and up).
         :type telegram_specification: dict
         """
         self.apply_checksum_validation = apply_checksum_validation
         self.telegram_specification = telegram_specification
-        # Regexes are compiled once to improve performance
-        self.telegram_specification_regexes = {
-            object["obis_reference"]: re.compile(object["obis_reference"], re.DOTALL | re.MULTILINE)
-            for object in self.telegram_specification['objects']
-        }
-        self._telegram_encryption_active = None
+        self._telegram_specification_regex = None
 
     def parse(self, telegram_data, encryption_key="", authentication_key="", throw_ex=False):  # noqa: C901
         """
@@ -47,11 +43,47 @@ class TelegramParser(object):
         :raises ParseError:
         :raises InvalidChecksumError:
         """
-        telegram_data = self.decrypt_telegram_data(
-            telegram_data=telegram_data,
-            encryption_key=encryption_key,
-            authentication_key=authentication_key
-        )
+        if not self.telegram_specification:
+            self.telegram_specification = \
+                match_telegram_specification(telegram_data)
+        if not self._telegram_specification_regex:
+            # Regexes are compiled once to improve performance
+            self._telegram_specification_regexes = {
+                object["obis_reference"]: re.compile(object["obis_reference"], re.DOTALL | re.MULTILINE)
+                for object in self.telegram_specification['objects']
+            }
+
+        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)
@@ -59,7 +91,7 @@ class TelegramParser(object):
         telegram = Telegram()
 
         for object in self.telegram_specification['objects']:
-            pattern = self.telegram_specification_regexes[object["obis_reference"]]
+            pattern = self._telegram_specification_regexes[object["obis_reference"]]
             matches = pattern.findall(telegram_data)
 
             # Some signatures are optional and may not be present,
@@ -86,42 +118,6 @@ class TelegramParser(object):
 
         return telegram
 
-    def decrypt_telegram_data(self, encryption_key, authentication_key, telegram_data):
-        """
-        Check if telegram data is encrypted and decrypt if applicable.
-        """
-        # if self._telegram_encryption_active is False:
-        #     # If encryption is not working, stop trying and logging warnings.
-        #     return telegram_data
-
-        if self.telegram_specification.get("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")
-            self._telegram_encryption_active = True
-        else:
-            try:
-                if unhexlify(telegram_data[0:2])[0] == GeneralGlobalCipher.TAG:
-                    logger.warning("Looks like a general_global_cipher frame "
-                                   "but telegram specification is not matching!")
-            except Exception:
-                pass
-            self._telegram_encryption_active = False
-
-        return telegram_data
-
     @staticmethod
     def validate_checksum(telegram):
         """
@@ -183,6 +179,37 @@ class TelegramParser(object):
         return crcValue
 
 
+def match_telegram_specification(telegram_data):
+    """
+    Find telegram specification that matches the telegram data by trying all
+    specifications.
+
+    Could be further optimized to check the actual 0.2.8 OBIS reference which
+    is available for DSMR version 4 and up.
+
+    :param str telegram_data: full telegram from start ('/') to checksum
+            ('!ABCD') including line endings in between the telegram's lines
+    :return: telegram specification
+    :rtype: dict
+    """
+    # Prevent circular import
+    from dsmr_parser import telegram_specifications
+
+    for specification in telegram_specifications.ALL:
+        try:
+            TelegramParser(specification).parse(telegram_data)
+        except ParseError:
+            pass
+        else:
+            return specification
+
+    raise TelegramSpecificationMatchError(
+        'Could automatically match telegram specification. Make sure the data'
+        'is not corrupt. Alternatively manually specify one.'
+    )
+
+
+
 class DSMRObjectParser(object):
     """
     Parses an object (can also be see as a 'line') from a telegram.

dsmr_parser/telegram_specifications.py

@@ -486,8 +486,6 @@ V5 = {
     ]
 }
 
-ALL = (V2_2, V3, V4, V5)
-
 BELGIUM_FLUVIUS = {
     'checksum_support': True,
     'objects': [
@@ -1368,3 +1366,17 @@ EON_HUNGARY = {
         }
     ]
 }
+
+ALL = (
+    V2_2,
+    V3,
+    V4,
+    V5,
+    BELGIUM_FLUVIUS,
+    LUXEMBOURG_SMARTY,
+    SWEDEN,
+    SAGEMCOM_T210_D_R,
+    AUSTRIA_ENERGIENETZE_STEIERMARK,
+    ISKRA_IE,
+    EON_HUNGARY
+)

setup.py

@@ -7,7 +7,7 @@ setup(
     author_email='mail@nldr.net',
     license='MIT',
     url='https://github.com/ndokter/dsmr_parser',
-    version='1.4.2',
+    version='1.5.0',
     packages=find_packages(exclude=('test', 'test.*')),
     install_requires=[
         'pyserial>=3,<4',

test/clients/test_filereader.py

@@ -1,21 +0,0 @@
-import unittest
-import tempfile
-
-from dsmr_parser.clients.filereader import FileReader
-from dsmr_parser.telegram_specifications import V5
-from test.example_telegrams import TELEGRAM_V5
-
-
-class FileReaderTest(unittest.TestCase):
-    def test_read_as_object(self):
-        with tempfile.NamedTemporaryFile() as file:
-            with open(file.name, "w") as f:
-                f.write(TELEGRAM_V5)
-
-            telegrams = []
-            reader = FileReader(file=file.name, telegram_specification=V5)
-            # Call
-            for telegram in reader.read_as_object():
-                telegrams.append(telegram)
-
-            self.assertEqual(len(telegrams), 1)

test/clients/test_rfxtrx_protocol.py

@@ -1,77 +0,0 @@
-from unittest.mock import Mock
-
-import unittest
-
-from dsmr_parser import obis_references as obis
-from dsmr_parser.clients.rfxtrx_protocol import create_rfxtrx_dsmr_protocol, PACKETTYPE_DSMR, SUBTYPE_P1
-from dsmr_parser.objects import Telegram
-
-TELEGRAM_V2_2 = (
-    '/ISk5\2MT382-1004\r\n'
-    '\r\n'
-    '0-0:96.1.1(00000000000000)\r\n'
-    '1-0:1.8.1(00001.001*kWh)\r\n'
-    '1-0:1.8.2(00001.001*kWh)\r\n'
-    '1-0:2.8.1(00001.001*kWh)\r\n'
-    '1-0:2.8.2(00001.001*kWh)\r\n'
-    '0-0:96.14.0(0001)\r\n'
-    '1-0:1.7.0(0001.01*kW)\r\n'
-    '1-0:2.7.0(0000.00*kW)\r\n'
-    '0-0:17.0.0(0999.00*kW)\r\n'
-    '0-0:96.3.10(1)\r\n'
-    '0-0:96.13.1()\r\n'
-    '0-0:96.13.0()\r\n'
-    '0-1:24.1.0(3)\r\n'
-    '0-1:96.1.0(000000000000)\r\n'
-    '0-1:24.3.0(161107190000)(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'
-)
-
-OTHER_RF_PACKET = b'\x03\x01\x02\x03'
-
-
-def encode_telegram_as_RF_packets(telegram):
-    data = b''
-
-    for line in telegram.split('\n'):
-        packet_data = (line + '\n').encode('ascii')
-        packet_header = bytes(bytearray([
-            len(packet_data) + 3,  # excluding length byte
-            PACKETTYPE_DSMR,
-            SUBTYPE_P1,
-            0  # seq num (ignored)
-        ]))
-
-        data += packet_header + packet_data
-        # other RF packets can pass by on the line
-        data += OTHER_RF_PACKET
-
-    return data
-
-
-class RFXtrxProtocolTest(unittest.TestCase):
-
-    def setUp(self):
-        new_protocol, _ = create_rfxtrx_dsmr_protocol('2.2',
-                                                      telegram_callback=Mock(),
-                                                      keep_alive_interval=1)
-        self.protocol = new_protocol()
-
-    def test_complete_packet(self):
-        """Protocol should assemble incoming lines into complete packet."""
-
-        data = encode_telegram_as_RF_packets(TELEGRAM_V2_2)
-        # send data broken up in two parts
-        self.protocol.data_received(data[0:200])
-        self.protocol.data_received(data[200:])
-
-        telegram = self.protocol.telegram_callback.call_args_list[0][0][0]
-        assert isinstance(telegram, Telegram)
-
-        assert float(telegram[obis.CURRENT_ELECTRICITY_USAGE].value) == 1.01
-        assert telegram[obis.CURRENT_ELECTRICITY_USAGE].unit == 'kW'
-
-        assert float(telegram[obis.GAS_METER_READING].value) == 1.001
-        assert telegram[obis.GAS_METER_READING].unit == 'm3'

test/test_match_telegram_specification.py

@@ -0,0 +1,29 @@
+import unittest
+
+from dsmr_parser.exceptions import TelegramSpecificationMatchError
+from dsmr_parser.parsers import match_telegram_specification
+from dsmr_parser import telegram_specifications
+from test import example_telegrams
+
+
+class MatchTelegramSpecificationTest(unittest.TestCase):
+
+    def test_v2_2(self):
+        assert match_telegram_specification(example_telegrams.TELEGRAM_V2_2) \
+           == telegram_specifications.V2_2
+
+    def test_v3(self):
+        assert match_telegram_specification(example_telegrams.TELEGRAM_V3) \
+           == telegram_specifications.V3
+
+    def test_v4_2(self):
+        assert match_telegram_specification(example_telegrams.TELEGRAM_V4_2) \
+           == telegram_specifications.V4
+
+    def test_v5(self):
+        assert match_telegram_specification(example_telegrams.TELEGRAM_V5) \
+           == telegram_specifications.V5
+
+    def test_malformed_telegram(self):
+        with self.assertRaises(TelegramSpecificationMatchError):
+            match_telegram_specification(example_telegrams.TELEGRAM_V5[:-4])

test/test_parse_v2_2.py

@@ -12,6 +12,15 @@ from test.example_telegrams import TELEGRAM_V2_2
 class TelegramParserV2_2Test(unittest.TestCase):
     """ Test parsing of a DSMR v2.2 telegram. """
 
+    def test_telegram_specification_matching(self):
+        parser = TelegramParser()
+        parser.parse(TELEGRAM_V2_2)
+
+        self.assertEqual(
+            parser.telegram_specification,
+            telegram_specifications.V2_2
+        )
+
     def test_parse(self):
         parser = TelegramParser(telegram_specifications.V2_2)
         try:

test/test_parse_v3.py

@@ -12,6 +12,15 @@ from test.example_telegrams import TELEGRAM_V3
 class TelegramParserV3Test(unittest.TestCase):
     """ Test parsing of a DSMR v3 telegram. """
 
+    def test_telegram_specification_matching(self):
+        parser = TelegramParser()
+        parser.parse(TELEGRAM_V3)
+
+        self.assertEqual(
+            parser.telegram_specification,
+            telegram_specifications.V3
+        )
+
     def test_parse(self):
         parser = TelegramParser(telegram_specifications.V3)
         try:

test/test_parse_v4_2.py

@@ -15,6 +15,15 @@ from test.example_telegrams import TELEGRAM_V4_2
 class TelegramParserV4_2Test(unittest.TestCase):
     """ Test parsing of a DSMR v4.2 telegram. """
 
+    def test_telegram_specification_matching(self):
+        parser = TelegramParser()
+        parser.parse(TELEGRAM_V4_2)
+
+        self.assertEqual(
+            parser.telegram_specification,
+            telegram_specifications.V4
+        )
+
     def test_parse(self):
         parser = TelegramParser(telegram_specifications.V4)
         try:

test/test_parse_v5.py

@@ -15,13 +15,22 @@ from test.example_telegrams import TELEGRAM_V5
 class TelegramParserV5Test(unittest.TestCase):
     """ Test parsing of a DSMR v5.x telegram. """
 
+    def test_telegram_specification_matching(self):
+        parser = TelegramParser()
+        parser.parse(TELEGRAM_V5)
+
+        self.assertEqual(
+            parser.telegram_specification,
+            telegram_specifications.V5
+        )
+
     def test_parse(self):
         parser = TelegramParser(telegram_specifications.V5)
         try:
             telegram = parser.parse(TELEGRAM_V5, throw_ex=True)
         except Exception as ex:
             assert False, f"parse trigged an exception {ex}"
-
+        print('test: ', type(telegram.P1_MESSAGE_HEADER), telegram.P1_MESSAGE_HEADER.__dict__)
         # P1_MESSAGE_HEADER (1-3:0.2.8)
         assert isinstance(telegram.P1_MESSAGE_HEADER, CosemObject)
         assert telegram.P1_MESSAGE_HEADER.unit is None