From 14d9869fc31a2722daeac619fb737676390ff8d2 Mon Sep 17 00:00:00 2001 From: Nigel Dokter Date: Sat, 18 Feb 2023 19:55:27 +0100 Subject: [PATCH] issue-51-telegram feedback from lowdef --- dsmr_parser/objects.py | 56 ++++++++++++++++++--------------- test/objects/test_mbusdevice.py | 8 ++--- test/objects/test_telegram.py | 26 +++++---------- 3 files changed, 42 insertions(+), 48 deletions(-) diff --git a/dsmr_parser/objects.py b/dsmr_parser/objects.py index f4755f8..25c83f3 100644 --- a/dsmr_parser/objects.py +++ b/dsmr_parser/objects.py @@ -19,22 +19,22 @@ class Telegram(dict): [k for k,v in telegram] yields: ['P1_MESSAGE_HEADER', 'P1_MESSAGE_TIMESTAMP', 'EQUIPMENT_IDENTIFIER', ...] + + Note: Dict like usage is deprecated. The inheritance from dict is because of backwards compatibility. """ def __init__(self, *args, **kwargs): - self._mbus_channel_devices = {} self._item_names = [] + self._mbus_devices = [] super().__init__(*args, **kwargs) def add(self, obis_reference, dsmr_object): # Update name mapping used to get value by attribute. Example: telegram.P1_MESSAGE_HEADER - # Also keep track of the added names internally obis_name = obis_name_mapping.EN[obis_reference] setattr(self, obis_name, dsmr_object) if obis_name not in self._item_names: # TODO solve issue with repeating obis references self._item_names.append(obis_name) - # Group Mbus related values into a MbusDevice object. - # TODO MaxDemandParser (BELGIUM_MAXIMUM_DEMAND_13_MONTHS) returns a list + # TODO isinstance check: MaxDemandParser (BELGIUM_MAXIMUM_DEMAND_13_MONTHS) returns a list if isinstance(dsmr_object, DSMRObject) and dsmr_object.is_mbus_reading: self._add_mbus(obis_reference, dsmr_object) @@ -49,26 +49,25 @@ class Telegram(dict): """ channel_id = dsmr_object.obis_id_code[1] - # Create new MbusDevice for the first record on this channel_id - if channel_id not in self._mbus_channel_devices: - self._mbus_channel_devices[channel_id] = MbusDevice(channel_id=channel_id) + # Create new MbusDevice or update existing one as it's records are being added one by one. + mbus_device = self.get_mbus_device_by_channel(channel_id) + if not mbus_device: + mbus_device = MbusDevice(channel_id=channel_id) + self._mbus_devices.append(mbus_device) - mbus_device = self._mbus_channel_devices[channel_id] mbus_device.add(obis_reference, dsmr_object) - @property - def MBUS_DEVICES(self): - """ Return MbusDevice objects which are used for water, heat and gas meters. """ - return list(self._mbus_channel_devices.values()) + if not hasattr(self, 'MBUS_DEVICES'): + setattr(self, 'MBUS_DEVICES', self._mbus_devices) + self._item_names.append('MBUS_DEVICES') def get_mbus_device_by_channel(self, channel_id): """ :rtype: MbusDevice|None """ - return self._mbus_channel_devices.get(channel_id) - - def __len__(self): - return len(self._item_names) + for mbus_device in self._mbus_devices: + if mbus_device.channel_id == channel_id: + return mbus_device def __iter__(self): for attr in self._item_names: @@ -78,20 +77,22 @@ class Telegram(dict): def __str__(self): output = "" for attr, value in self: - output += "{}: \t {}\n".format(attr, str(value)) - - for channel_id, mbus_device in self._mbus_channel_devices.items(): - output += f'MBUS DEVICE (channel: {channel_id})\n' - for obis_name, value in mbus_device: - output += f'\t{obis_name}: \t {value} \n' + if attr == 'MBUS_DEVICES': + # Mbus devices are in a list + for mbus_device in value: + output += str(mbus_device) + else: + output += "{}: \t {}\n".format(attr, str(value)) return output def to_json(self): - telegram_data = {obis_name: json.loads(value.to_json()) for obis_name, value in self} + telegram_data = {obis_name: json.loads(value.to_json()) + for obis_name, value in self + if isinstance(value, DSMRObject)} telegram_data['MBUS_DEVICES'] = [ json.loads(mbus_device.to_json()) - for mbus_device in self._mbus_channel_devices.values() + for mbus_device in self._mbus_devices ] return json.dumps(telegram_data) @@ -112,6 +113,9 @@ class DSMRObject(object): return obis_id == 0 and channel_id != 0 + def to_json(self): + raise NotImplementedError + class MBusObject(DSMRObject): @@ -344,9 +348,9 @@ class MbusDevice: yield attr, value def __str__(self): - output = "CHANNEL_ID: \t {}\n".format(self.channel_id) + output = "MBUS DEVICE (channel {})\n".format(self.channel_id) for attr, value in self: - output += "{}: \t {}\n".format(attr, str(value)) + output += "\t{}: \t {}\n".format(attr, str(value)) return output def to_json(self): diff --git a/test/objects/test_mbusdevice.py b/test/objects/test_mbusdevice.py index 7caddbd..b92d4af 100644 --- a/test/objects/test_mbusdevice.py +++ b/test/objects/test_mbusdevice.py @@ -53,9 +53,9 @@ class MbusDeviceTest(unittest.TestCase): self.assertEqual( str(self.mbus_device), ( - 'CHANNEL_ID: 1\n' - 'DEVICE_TYPE: 3 [None]\n' - 'EQUIPMENT_IDENTIFIER_GAS: 4730303339303031393336393930363139 [None]\n' - 'HOURLY_GAS_METER_READING: 246.138 [m3] at 2020-04-26T20:30:01+00:00\n' + 'MBUS DEVICE (channel 1)\n' + '\tDEVICE_TYPE: 3 [None]\n' + '\tEQUIPMENT_IDENTIFIER_GAS: 4730303339303031393336393930363139 [None]\n' + '\tHOURLY_GAS_METER_READING: 246.138 [m3] at 2020-04-26T20:30:01+00:00\n' ) ) diff --git a/test/objects/test_telegram.py b/test/objects/test_telegram.py index 45c4423..17a6891 100644 --- a/test/objects/test_telegram.py +++ b/test/objects/test_telegram.py @@ -321,12 +321,6 @@ class TelegramTest(unittest.TestCase): assert item_names_tested_set == V4_name_set - def test_len(self): - parser = TelegramParser(telegram_specifications.V5) - telegram = parser.parse(TELEGRAM_V5_TWO_MBUS) - - self.assertEqual(len(telegram._item_names), 35) - def test_iter(self): parser = TelegramParser(telegram_specifications.V5) telegram = parser.parse(TELEGRAM_V5) @@ -373,13 +367,9 @@ class TelegramTest(unittest.TestCase): parser = TelegramParser(telegram_specifications.V5, apply_checksum_validation=False) telegram = parser.parse('') - self.assertEqual(telegram.MBUS_DEVICES, []) + self.assertFalse(hasattr(telegram, 'MBUS_DEVICES')) self.assertIsNone(telegram.get_mbus_device_by_channel(1)) - # Because of a bug related to incorrect use of defaultdict, - # test again for unwanted side effects - self.assertEqual(telegram.MBUS_DEVICES, []) - def test_to_json(self): parser = TelegramParser(telegram_specifications.V5) telegram = parser.parse(TELEGRAM_V5) @@ -474,6 +464,13 @@ class TelegramTest(unittest.TestCase): 'INSTANTANEOUS_CURRENT_L3: 0.86 [A]\n' 'TEXT_MESSAGE: None [None]\n' 'DEVICE_TYPE: 3 [None]\n' + 'MBUS DEVICE (channel 1)\n' + ' DEVICE_TYPE: 3 [None]\n' + ' EQUIPMENT_IDENTIFIER_GAS: 3232323241424344313233343536373839 [None]\n' + ' HOURLY_GAS_METER_READING: 0.107 [m3] at 2017-01-02T15:10:05+00:00\n' + 'MBUS DEVICE (channel 2)\n' + ' DEVICE_TYPE: 3 [None]\n' + ' EQUIPMENT_IDENTIFIER_GAS: None [None]\n' 'INSTANTANEOUS_ACTIVE_POWER_L1_POSITIVE: 0.070 [kW]\n' 'INSTANTANEOUS_ACTIVE_POWER_L2_POSITIVE: 0.032 [kW]\n' 'INSTANTANEOUS_ACTIVE_POWER_L3_POSITIVE: 0.142 [kW]\n' @@ -482,13 +479,6 @@ class TelegramTest(unittest.TestCase): 'INSTANTANEOUS_ACTIVE_POWER_L3_NEGATIVE: 0.000 [kW]\n' 'EQUIPMENT_IDENTIFIER_GAS: None [None]\n' 'HOURLY_GAS_METER_READING: 0.107 [m3] at 2017-01-02T15:10:05+00:00\n' - 'MBUS DEVICE (channel: 1)\n' - ' DEVICE_TYPE: 3 [None] \n' - ' EQUIPMENT_IDENTIFIER_GAS: 3232323241424344313233343536373839 [None] \n' - ' HOURLY_GAS_METER_READING: 0.107 [m3] at 2017-01-02T15:10:05+00:00 \n' - 'MBUS DEVICE (channel: 2)\n' - ' DEVICE_TYPE: 3 [None] \n' - ' EQUIPMENT_IDENTIFIER_GAS: None [None] \n' ) )