From ef66c8422c9f3a08e8808fa2e3ccf02bb42728ed Mon Sep 17 00:00:00 2001 From: Nigel Dokter Date: Tue, 7 Feb 2023 19:48:56 +0100 Subject: [PATCH] issue-51-telegram work in progress --- dsmr_parser/objects.py | 14 ++++++------ test/example_telegrams.py | 45 +++++++++++++++++++++++++++++++++++++ test/test_telegram.py | 47 ++++++++++++++++++++++++++++++++++++++- 3 files changed, 98 insertions(+), 8 deletions(-) diff --git a/dsmr_parser/objects.py b/dsmr_parser/objects.py index b649ca1..4fbe0a9 100644 --- a/dsmr_parser/objects.py +++ b/dsmr_parser/objects.py @@ -27,15 +27,11 @@ class Telegram(object): def add(self, obis_reference, dsmr_object): self._telegram_data[obis_reference].append(dsmr_object) - # TODO experiment with api to see what is nice - def get(self, obis_reference, channel=None): - if channel is None: - return self._telegram_data[obis_reference] - + def get_by_channel(self, obis_reference, channel): try: return next(filter(lambda x: x.channel == channel, self._telegram_data[obis_reference])) except StopIteration: - return None + raise LookupError('No value found for OBIS reference "{}" on channel "{}"'.format(obis_reference, channel)) def __getattr__(self, name): """ will only get called for undefined attributes """ @@ -45,7 +41,11 @@ class Telegram(object): return value def __getitem__(self, obis_reference): - return self._telegram_data[obis_reference][0] + try: + return self._telegram_data[obis_reference][0] + except IndexError: + # The index error is an internal detail. The KeyError is expected as a user. + raise KeyError def __len__(self): return len(self._telegram_data) # TODO: its nested now diff --git a/test/example_telegrams.py b/test/example_telegrams.py index 5a350ed..36ea45e 100644 --- a/test/example_telegrams.py +++ b/test/example_telegrams.py @@ -129,6 +129,51 @@ TELEGRAM_V5 = ( '!6EEE\r\n' ) +# V5 telegram with 2 MBUS devices +TELEGRAM_V5_TWO_MBUS = ( + '/ISK5\\2M550T-1012\r\n' + '\r\n' + '1-3:0.2.8(50)\r\n' + '0-0:1.0.0(200426223325S)\r\n' + '0-0:96.1.1(4530303434303037333832323436303139)\r\n' + '1-0:1.8.1(002130.115*kWh)\r\n' + '1-0:1.8.2(000245.467*kWh)\r\n' + '1-0:2.8.1(000000.000*kWh)\r\n' + '1-0:2.8.2(000000.000*kWh)\r\n' + '0-0:96.14.0(0001)\r\n' + '1-0:1.7.0(00.111*kW)\r\n' + '1-0:2.7.0(00.000*kW)\r\n' + '0-0:96.7.21(00005)\r\n' + '0-0:96.7.9(00003)\r\n' + '1-0:99.97.0(1)(0-0:96.7.19)(190326095015W)(0000002014*s)\r\n' + '1-0:32.32.0(00001)\r\n' + '1-0:52.32.0(00001)\r\n' + '1-0:72.32.0(00192)\r\n' + '1-0:32.36.0(00001)\r\n' + '1-0:52.36.0(00001)\r\n' + '1-0:72.36.0(00001)\r\n' + '0-0:96.13.0()\r\n' + '1-0:32.7.0(229.9*V)\r\n' + '1-0:52.7.0(229.2*V)\r\n' + '1-0:72.7.0(222.9*V)\r\n' + '1-0:31.7.0(000*A)\r\n' + '1-0:51.7.0(000*A)\r\n' + '1-0:71.7.0(001*A)\r\n' + '1-0:21.7.0(00.056*kW)\r\n' + '1-0:41.7.0(00.000*kW)\r\n' + '1-0:61.7.0(00.055*kW)\r\n' + '1-0:22.7.0(00.000*kW)\r\n' + '1-0:42.7.0(00.000*kW)\r\n' + '1-0:62.7.0(00.000*kW)\r\n' + '0-1:24.1.0(003)\r\n' + '0-1:96.1.0()\r\n' + '0-1:24.2.1(700101010000W)(00000000)\r\n' + '0-2:24.1.0(003)\r\n' + '0-2:96.1.0(4730303339303031393336393930363139)\r\n' + '0-2:24.2.1(200426223001S)(00246.138*m3)\r\n' + '!56DD\r\n' +) + TELEGRAM_FLUVIUS_V171 = ( '/FLU5\253769484_A\r\n' '\r\n' diff --git a/test/test_telegram.py b/test/test_telegram.py index bf7a841..708651e 100644 --- a/test/test_telegram.py +++ b/test/test_telegram.py @@ -4,11 +4,12 @@ import pytz from dsmr_parser import telegram_specifications from dsmr_parser import obis_name_mapping +from dsmr_parser import obis_references as obis from dsmr_parser.objects import CosemObject from dsmr_parser.objects import MBusObject from dsmr_parser.objects import ProfileGenericObject from dsmr_parser.parsers import TelegramParser -from test.example_telegrams import TELEGRAM_V4_2 +from test.example_telegrams import TELEGRAM_V4_2, TELEGRAM_V5_TWO_MBUS from decimal import Decimal @@ -319,3 +320,47 @@ class TelegramTest(unittest.TestCase): item_names_tested_set = set(self.item_names_tested) assert item_names_tested_set == V4_name_set + + def test_get(self): + parser = TelegramParser(telegram_specifications.V5) + telegram = parser.parse(TELEGRAM_V5_TWO_MBUS) + + mbus_1 = telegram.get(obis.HOURLY_GAS_METER_READING, channel=1) + mbus_2 = telegram.get(obis.HOURLY_GAS_METER_READING, channel=2) + + self.assertEqual(type(mbus_1), MBusObject) + self.assertEqual(mbus_1.channel, 1) + self.assertEqual(mbus_1.value, 0) + + self.assertEqual(type(mbus_2), MBusObject) + self.assertEqual(mbus_2.channel, 2) + self.assertEqual(mbus_2.value, Decimal('246.138')) + + def test_get_without_channel(self): + """ Retrieve MBUS device without supplying channel which fetches the first MBUS record found """ + parser = TelegramParser(telegram_specifications.V5) + telegram = parser.parse(TELEGRAM_V5_TWO_MBUS) + + mbus = telegram.get(obis.HOURLY_GAS_METER_READING) + + self.assertEqual(type(mbus), MBusObject) + self.assertEqual(mbus.channel, 1) + self.assertEqual(mbus.value, 0) + + def test_get_unknown_value(self): + """ Retrieve MBUS device without supplying channel which fetches the first MBUS record found """ + parser = TelegramParser(telegram_specifications.V5) + telegram = parser.parse(TELEGRAM_V5_TWO_MBUS) + + # Test valid OBIS reference with wrong channel + with self.assertRaises(LookupError) as exception_context: + telegram.get_by_channel(obis.HOURLY_GAS_METER_READING, channel=123) + + self.assertEqual( + str(exception_context.exception), + 'No value found for OBIS reference "\d-\d:24\.2\.1.+?\\r\\n" on channel "123"' + ) + + # Test invalid OBIS reference + with self.assertRaises(LookupError): + telegram.get_by_channel('invalid_obis_reference', channel=1) \ No newline at end of file