From 0593d4172d6932b1a6f4f89f44cedb110a5f116c Mon Sep 17 00:00:00 2001
From: Nigel Dokter <mail@nldr.net>
Date: Sun, 12 Feb 2023 18:06:01 +0100
Subject: [PATCH] issue-51-telegram fix bug when no mbus devices are present
---
dsmr_parser/objects.py | 13 +++++++++----
test/test_telegram.py | 11 +++++++++++
2 files changed, 20 insertions(+), 4 deletions(-)
diff --git a/dsmr_parser/objects.py b/dsmr_parser/objects.py
index 1ae26bc..b6455e0 100644
--- a/dsmr_parser/objects.py
+++ b/dsmr_parser/objects.py
@@ -36,9 +36,12 @@ class Telegram(object):
# Group Mbus related values into a MbusDevice object.
# 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)
+ self._add_mbus(obis_reference, dsmr_object)
+
+ def _add_mbus(self, obis_reference, dsmr_object):
+ 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):
"""
@@ -47,7 +50,9 @@ class Telegram(object):
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):
- return self._mbus_devices[channel_id]
+ # Check key, because defaultdict would otherwise instantiate an empty MbusDevice
+ if channel_id in self._mbus_devices:
+ return self._mbus_devices[channel_id]
def __getattr__(self, name):
""" will only get called for undefined attributes """
diff --git a/test/test_telegram.py b/test/test_telegram.py
index bb60b8a..20c79be 100644
--- a/test/test_telegram.py
+++ b/test/test_telegram.py
@@ -356,3 +356,14 @@ class TelegramTest(unittest.TestCase):
self.assertEqual(mbus_device_2.DEVICE_TYPE.value, 3)
self.assertEqual(mbus_device_2.EQUIPMENT_IDENTIFIER_GAS.value, '4730303339303031393336393930363139')
self.assertEqual(mbus_device_2.HOURLY_GAS_METER_READING.value, Decimal('246.138'))
+
+ def test_without_mbus_devices(self):
+ parser = TelegramParser(telegram_specifications.V5, apply_checksum_validation=False)
+ telegram = parser.parse('')
+
+ self.assertEqual(telegram.get_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.get_mbus_devices(), [])