From 3ae06f88b7d040d8a8033ad239482ac3d60f1281 Mon Sep 17 00:00:00 2001 From: Nigel Dokter Date: Sat, 11 Feb 2023 09:39:44 +0100 Subject: [PATCH] issue-51-telegram work in progress --- dsmr_parser/objects.py | 29 ++------------- test/test_telegram.py | 82 ++++++++---------------------------------- 2 files changed, 17 insertions(+), 94 deletions(-) diff --git a/dsmr_parser/objects.py b/dsmr_parser/objects.py index ac2014d..008da34 100644 --- a/dsmr_parser/objects.py +++ b/dsmr_parser/objects.py @@ -34,45 +34,21 @@ class Telegram(object): self._item_names.append(self._obis_name_mapping[obis_reference]) # Group Mbus related values into a MbusDevice object. - # TODO sometimes this is a list due to BELGIUM_MAXIMUM_DEMAND_13_MONTHS + # TODO MaxDemandParser (BELGIUM_MAXIMUM_DEMAND_13_MONTHS) returns a list if isinstance(dsmr_object, DSMRObject) and dsmr_object.is_mbus_reading: channel_id = dsmr_object.obis_id_code[1] mbus_device = self._mbus_devices[channel_id] mbus_device.add(obis_reference, dsmr_object) - def get_mbus_devices(self): """ Return MbusDevice objects which are used for water, heat and gas meters. """ - # TODO sort by channel ID - return list(self._mbus_devices.values()) + return [d[1] for d in sorted(self._mbus_devices.items(), key=lambda x: x[0])] def get_mbus_device_by_channel(self, channel_id=None): return self._mbus_devices[channel_id] - # # TODO devices groeperen. alle values van dat channel daar in groeperen en wrappen in device object gebruik makende van device id - # def get(self, obis_reference, channel=None): - # """ - # Get values by OBIS reference (regex). If multiple values exist a list is returned, unless filtering by channel. - # May assume that values are sorted by channel. - # """ - # if channel is None: - # try: - # values = self._telegram_data[obis_reference] - # except KeyError: - # raise LookupError('No value found for OBIS reference "{}"'.format(obis_reference)) - # - # if len(values) == 1: - # return values[0] - # else: - # return values - # - # try: - # return [v for v in self._telegram_data[obis_reference] if v.channel == channel][0] - # except IndexError: - # 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 """ obis_reference = self._reverse_obis_name_mapping[name] @@ -126,6 +102,7 @@ class DSMRObject(object): return obis_id == 0 and channel_id != 0 + class MBusObject(DSMRObject): @property diff --git a/test/test_telegram.py b/test/test_telegram.py index a380b9e..c289314 100644 --- a/test/test_telegram.py +++ b/test/test_telegram.py @@ -4,8 +4,7 @@ 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, MbusDevice +from dsmr_parser.objects import CosemObject from dsmr_parser.objects import MBusObject from dsmr_parser.objects import ProfileGenericObject from dsmr_parser.parsers import TelegramParser @@ -327,84 +326,31 @@ class TelegramTest(unittest.TestCase): self.assertEqual(len(telegram), 35) - # def test_get(self): - # """ Retrieve MBUS device without supplying channel which fetches all (two) records found. """ - # parser = TelegramParser(telegram_specifications.V5) - # telegram = parser.parse(TELEGRAM_V5_TWO_MBUS) - # - # # A single value is returned for the current electricity usage - # electricity_used_value = telegram.get(obis.CURRENT_ELECTRICITY_USAGE) - # self.assertEqual(type(electricity_used_value), CosemObject) - # self.assertEqual(electricity_used_value.channel, 0) - # self.assertEqual(electricity_used_value.value, Decimal('0.111')) - # - # # Multiple values are returned for the gas reading - # gas_values = telegram.get(obis.HOURLY_GAS_METER_READING) - # self.assertEqual(len(gas_values), 2) - # - # gas_value_1 = gas_values[0] - # self.assertEqual(type(gas_value_1), MBusObject) - # self.assertEqual(gas_value_1.channel, 1) - # self.assertEqual(gas_value_1.value, 0) - # - # gas_value_2 = gas_values[1] - # self.assertEqual(type(gas_value_2), MBusObject) - # self.assertEqual(gas_value_2.channel, 2) - # self.assertEqual(gas_value_2.value, Decimal('246.138')) - # - # def test_get_with_channel(self): - # parser = TelegramParser(telegram_specifications.V5) - # telegram = parser.parse(TELEGRAM_V5_TWO_MBUS) - # - # gas_value_1 = telegram.get(obis.HOURLY_GAS_METER_READING, channel=1) - # gas_value_2 = telegram.get(obis.HOURLY_GAS_METER_READING, channel=2) - # - # self.assertEqual(type(gas_value_1), MBusObject) - # self.assertEqual(gas_value_1.channel, 1) - # self.assertEqual(gas_value_1.value, 0) - # - # self.assertEqual(type(gas_value_2), MBusObject) - # self.assertEqual(gas_value_2.channel, 2) - # self.assertEqual(gas_value_2.value, Decimal('246.138')) - # - # 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(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('invalid_obis_reference', channel=1) - - # TODO def test_get_mbus_devices(self): parser = TelegramParser(telegram_specifications.V5) telegram = parser.parse(TELEGRAM_V5_TWO_MBUS) mbus_devices = telegram.get_mbus_devices() - print(mbus_devices) + self.assertEqual(len(mbus_devices), 2) mbus_device_1 = mbus_devices[0] - self.assertEqual(type(mbus_device_1), MbusDevice) - print('mbus_device_1.HOURLY_GAS_METER_READING: ', mbus_device_1.HOURLY_GAS_METER_READING) + self.assertEqual(mbus_device_1.EQUIPMENT_IDENTIFIER_GAS.value, None) + print() + self.assertEqual(mbus_device_1.HOURLY_GAS_METER_READING.value, Decimal('0')) mbus_device_2 = mbus_devices[1] - self.assertEqual(type(mbus_device_2), MbusDevice) - print('mbus_device_2.HOURLY_GAS_METER_READING: ', mbus_device_2.HOURLY_GAS_METER_READING) + self.assertEqual(mbus_device_2.EQUIPMENT_IDENTIFIER_GAS.value, '4730303339303031393336393930363139') + self.assertEqual(mbus_device_2.HOURLY_GAS_METER_READING.value, Decimal('246.138')) - # TODO def test_get_mbus_device_by_channel(self): parser = TelegramParser(telegram_specifications.V5) telegram = parser.parse(TELEGRAM_V5_TWO_MBUS) - print('by channel: ', telegram.get_mbus_device_by_channel(2).HOURLY_GAS_METER_READING) + mbus_device_1 = telegram.get_mbus_device_by_channel(1) + self.assertEqual(mbus_device_1.EQUIPMENT_IDENTIFIER_GAS.value, None) + self.assertEqual(mbus_device_1.HOURLY_GAS_METER_READING.value, Decimal('0')) + + mbus_device_2 = telegram.get_mbus_device_by_channel(2) + self.assertEqual(mbus_device_2.EQUIPMENT_IDENTIFIER_GAS.value, '4730303339303031393336393930363139') + self.assertEqual(mbus_device_2.HOURLY_GAS_METER_READING.value, Decimal('246.138'))