From ad25fd4182e703c6076a99caa3f4eff6e5796b67 Mon Sep 17 00:00:00 2001 From: Nigel Dokter Date: Wed, 15 Feb 2023 11:55:28 +0100 Subject: [PATCH] issue-51-telegram work in progress --- dsmr_parser/objects.py | 62 ++++++++++++---- test/objects/__init__.py | 0 test/objects/test_mbusdevice.py | 50 +++++++++++++ .../{ => objects}/test_parser_corner_cases.py | 0 test/{ => objects}/test_telegram.py | 74 ++++++++++++++++++- 5 files changed, 167 insertions(+), 19 deletions(-) create mode 100644 test/objects/__init__.py create mode 100644 test/objects/test_mbusdevice.py rename test/{ => objects}/test_parser_corner_cases.py (100%) rename test/{ => objects}/test_telegram.py (79%) diff --git a/dsmr_parser/objects.py b/dsmr_parser/objects.py index aa40a4c..a4e4ca6 100644 --- a/dsmr_parser/objects.py +++ b/dsmr_parser/objects.py @@ -23,12 +23,17 @@ class Telegram(object): def __init__(self): self._telegram_data = defaultdict(list) self._mbus_channel_devices = {} + self._item_names = [] def add(self, obis_reference, dsmr_object): self._telegram_data[obis_reference].append(dsmr_object) # Update name mapping used to get value by attribute. Example: telegram.P1_MESSAGE_HEADER - setattr(self, obis_name_mapping.EN[obis_reference], dsmr_object) + # 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 @@ -63,26 +68,26 @@ class Telegram(object): def __getitem__(self, obis_reference): """ - Get value by key. Example: telegram[obis_references.P1_MESSAGE_HEADER] + Deprecated method to get obis_reference by key. Exists for backwards compatibility - For Mbus devices like gas and water meters, it's better to use get_mbus_devices and get_mbus_device_by_channel. - This key approach will only fetch the first found value and therefor might not be accurate. + Example: telegram[obis_references.P1_MESSAGE_HEADER] """ try: + # TODO use _telegram_data here or else TelegramParserFluviusTest.test_parse breaks return self._telegram_data[obis_reference][0] + # obis_name = obis_name_mapping.EN[obis_reference] + # return getattr(self, obis_name) 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) + return len(self._item_names) def __iter__(self): - for obis_reference, values in self._telegram_data.items(): - reverse_obis_name = obis_name_mapping.EN[obis_reference] - value = values[0] # TODO might be considered legacy behavior? - - yield reverse_obis_name, value + for attr in self._item_names: + value = getattr(self, attr) + yield attr, value def __str__(self): output = "" @@ -91,7 +96,13 @@ class Telegram(object): return output def to_json(self): - return json.dumps(dict([[attr, json.loads(value.to_json())] for attr, value in self])) + telegram_data = {obis_name: json.loads(value.to_json()) for obis_name, value in self} + telegram_data['MBUS_DEVICES'] = [ + json.loads(mbus_device.to_json()) + for mbus_device in self._mbus_channel_devices.values() + ] + + return json.dumps(telegram_data) class DSMRObject(object): @@ -319,10 +330,31 @@ class MbusDevice: def __init__(self, channel_id): self.channel_id = channel_id - self._telegram_data = {} + self._item_names = [] def add(self, obis_reference, dsmr_object): - self._telegram_data[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) + self._item_names.append(obis_name) - # Update name mapping used to get value by attribute. Example: device.HOURLY_GAS_METER_READING - setattr(self, obis_name_mapping.EN[obis_reference], dsmr_object) + def __len__(self): + return len(self._item_names) + + def __iter__(self): + for attr in self._item_names: + value = getattr(self, attr) + yield attr, value + + def __str__(self): + output = "" + for attr, value in self: + output += "{}: \t {}\n".format(attr, str(value)) + return output + + def to_json(self): + data = {obis_name: json.loads(value.to_json()) for obis_name, value in self} + data['CHANNEL_ID'] = self.channel_id + + return json.dumps(data) diff --git a/test/objects/__init__.py b/test/objects/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/test/objects/test_mbusdevice.py b/test/objects/test_mbusdevice.py new file mode 100644 index 0000000..4cc20fb --- /dev/null +++ b/test/objects/test_mbusdevice.py @@ -0,0 +1,50 @@ +from decimal import Decimal + +import json +import unittest + +from dsmr_parser import telegram_specifications, obis_references +from dsmr_parser.objects import MbusDevice + + +class MbusDeviceTest(unittest.TestCase): + + def setUp(self): + v5_objects = telegram_specifications.V5['objects'] + + device_type_parser = v5_objects[obis_references.DEVICE_TYPE] + device_type = device_type_parser.parse('0-2:24.1.0(003)\r\n') + + equipment_parser = v5_objects[obis_references.EQUIPMENT_IDENTIFIER_GAS] + equipment = equipment_parser.parse('0-2:96.1.0(4730303339303031393336393930363139)\r\n') + + gas_reading_parser = v5_objects[obis_references.HOURLY_GAS_METER_READING] + gas_reading = gas_reading_parser.parse('0-2:24.2.1(200426223001S)(00246.138*m3)\r\n') + + mbus_device = MbusDevice(channel_id=1) + mbus_device.add(obis_references.DEVICE_TYPE, device_type) + mbus_device.add(obis_references.EQUIPMENT_IDENTIFIER_GAS, equipment) + mbus_device.add(obis_references.HOURLY_GAS_METER_READING, gas_reading) + + self.mbus_device = mbus_device + + def test_attributes(self): + self.assertEqual(self.mbus_device.DEVICE_TYPE.value, 3) + self.assertEqual(self.mbus_device.DEVICE_TYPE.unit, None) + + self.assertEqual(self.mbus_device.EQUIPMENT_IDENTIFIER_GAS.value, + '4730303339303031393336393930363139') + self.assertEqual(self.mbus_device.EQUIPMENT_IDENTIFIER_GAS.unit, None) + + self.assertEqual(self.mbus_device.HOURLY_GAS_METER_READING.value, Decimal('246.138')) + self.assertEqual(self.mbus_device.HOURLY_GAS_METER_READING.unit, 'm3') + + def test_to_json(self): + self.assertEqual( + json.loads(self.mbus_device.to_json()), + { + 'CHANNEL_ID': 1, + 'DEVICE_TYPE': {'value': 3, 'unit': None}, + 'EQUIPMENT_IDENTIFIER_GAS': {'value': '4730303339303031393336393930363139', 'unit': None}, + 'HOURLY_GAS_METER_READING': {'datetime': '2020-04-26T22:30:01+02:00', 'value': 246.138, 'unit': 'm3'}} + ) diff --git a/test/test_parser_corner_cases.py b/test/objects/test_parser_corner_cases.py similarity index 100% rename from test/test_parser_corner_cases.py rename to test/objects/test_parser_corner_cases.py diff --git a/test/test_telegram.py b/test/objects/test_telegram.py similarity index 79% rename from test/test_telegram.py rename to test/objects/test_telegram.py index c33332d..9e90491 100644 --- a/test/test_telegram.py +++ b/test/objects/test_telegram.py @@ -1,3 +1,4 @@ +import json import unittest import datetime import pytz @@ -324,17 +325,17 @@ class TelegramTest(unittest.TestCase): parser = TelegramParser(telegram_specifications.V5) telegram = parser.parse(TELEGRAM_V5_TWO_MBUS) - self.assertEqual(len(telegram), 35) + self.assertEqual(len(telegram._item_names), 35) def test_iter(self): parser = TelegramParser(telegram_specifications.V5) telegram = parser.parse(TELEGRAM_V5) - for obis_reference, dsmr_object in telegram: + for obis_name, dsmr_object in telegram: break - # Verify that the iterator works for at least on evalue - self.assertEqual(obis_reference, obis_name_mapping.EN[obis_references.P1_MESSAGE_HEADER]) + # Verify that the iterator works for at least one value + self.assertEqual(obis_name, obis_name_mapping.EN[obis_references.P1_MESSAGE_HEADER]) self.assertEqual(dsmr_object.value, '50') def test_get_mbus_devices(self): @@ -378,3 +379,68 @@ class TelegramTest(unittest.TestCase): # Because of a bug related to incorrect use of defaultdict, # test again for unwanted side effects self.assertEqual(telegram.get_mbus_devices(), []) + + def test_to_json(self): + parser = TelegramParser(telegram_specifications.V5) + telegram = parser.parse(TELEGRAM_V5) + json_data = json.loads(telegram.to_json()) + + self.assertEqual( + json_data, + {'CURRENT_ELECTRICITY_DELIVERY': {'unit': 'kW', 'value': 0.0}, + 'CURRENT_ELECTRICITY_USAGE': {'unit': 'kW', 'value': 0.244}, + 'DEVICE_TYPE': {'unit': None, 'value': 3}, + 'ELECTRICITY_ACTIVE_TARIFF': {'unit': None, 'value': '0002'}, + 'ELECTRICITY_DELIVERED_TARIFF_1': {'unit': 'kWh', 'value': 2.444}, + 'ELECTRICITY_DELIVERED_TARIFF_2': {'unit': 'kWh', 'value': 0.0}, + 'ELECTRICITY_USED_TARIFF_1': {'unit': 'kWh', 'value': 4.426}, + 'ELECTRICITY_USED_TARIFF_2': {'unit': 'kWh', 'value': 2.399}, + 'EQUIPMENT_IDENTIFIER': {'unit': None, + 'value': '4B384547303034303436333935353037'}, + 'EQUIPMENT_IDENTIFIER_GAS': {'unit': None, 'value': None}, + 'HOURLY_GAS_METER_READING': {'datetime': '2017-01-02T16:10:05+01:00', + 'unit': 'm3', + 'value': 0.107}, + 'INSTANTANEOUS_ACTIVE_POWER_L1_NEGATIVE': {'unit': 'kW', 'value': 0.0}, + 'INSTANTANEOUS_ACTIVE_POWER_L1_POSITIVE': {'unit': 'kW', 'value': 0.07}, + 'INSTANTANEOUS_ACTIVE_POWER_L2_NEGATIVE': {'unit': 'kW', 'value': 0.0}, + 'INSTANTANEOUS_ACTIVE_POWER_L2_POSITIVE': {'unit': 'kW', 'value': 0.032}, + 'INSTANTANEOUS_ACTIVE_POWER_L3_NEGATIVE': {'unit': 'kW', 'value': 0.0}, + 'INSTANTANEOUS_ACTIVE_POWER_L3_POSITIVE': {'unit': 'kW', 'value': 0.142}, + 'INSTANTANEOUS_CURRENT_L1': {'unit': 'A', 'value': 0.48}, + 'INSTANTANEOUS_CURRENT_L2': {'unit': 'A', 'value': 0.44}, + 'INSTANTANEOUS_CURRENT_L3': {'unit': 'A', 'value': 0.86}, + 'INSTANTANEOUS_VOLTAGE_L1': {'unit': 'V', 'value': 230.0}, + 'INSTANTANEOUS_VOLTAGE_L2': {'unit': 'V', 'value': 230.0}, + 'INSTANTANEOUS_VOLTAGE_L3': {'unit': 'V', 'value': 229.0}, + 'LONG_POWER_FAILURE_COUNT': {'unit': None, 'value': 0}, + 'MBUS_DEVICES': [{'CHANNEL_ID': 1, + 'DEVICE_TYPE': {'unit': None, 'value': 3}, + 'EQUIPMENT_IDENTIFIER_GAS': {'unit': None, + 'value': '3232323241424344313233343536373839'}, + 'HOURLY_GAS_METER_READING': {'datetime': '2017-01-02T16:10:05+01:00', + 'unit': 'm3', + 'value': 0.107}}, + {'CHANNEL_ID': 2, + 'DEVICE_TYPE': {'unit': None, 'value': 3}, + 'EQUIPMENT_IDENTIFIER_GAS': {'unit': None, 'value': None}}], + 'P1_MESSAGE_HEADER': {'unit': None, 'value': '50'}, + 'P1_MESSAGE_TIMESTAMP': {'unit': None, 'value': '2017-01-02T19:20:02+01:00'}, + 'POWER_EVENT_FAILURE_LOG': {'buffer': [], + 'buffer_length': 0, + 'buffer_type': '0-0:96.7.19'}, + 'SHORT_POWER_FAILURE_COUNT': {'unit': None, 'value': 13}, + 'TEXT_MESSAGE': {'unit': None, 'value': None}, + 'VOLTAGE_SAG_L1_COUNT': {'unit': None, 'value': 0}, + 'VOLTAGE_SAG_L2_COUNT': {'unit': None, 'value': 0}, + 'VOLTAGE_SAG_L3_COUNT': {'unit': None, 'value': 0}, + 'VOLTAGE_SWELL_L1_COUNT': {'unit': None, 'value': 0}, + 'VOLTAGE_SWELL_L2_COUNT': {'unit': None, 'value': 0}, + 'VOLTAGE_SWELL_L3_COUNT': {'unit': None, 'value': 0}} + ) + + def test_getitem(self): + parser = TelegramParser(telegram_specifications.V5) + telegram = parser.parse(TELEGRAM_V5) + + self.assertEqual(telegram[obis_references.P1_MESSAGE_HEADER].value, '50')