issue-51-telegram work in progress

Browse Source
This commit is contained in:
Nigel Dokter 2023-02-11 09:39:44 +01:00
parent f5d402ad1b
commit 3ae06f88b7
2 changed files with 17 additions and 94 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

29
dsmr_parser/objects.py
View File

@ -34,45 +34,21 @@ class Telegram(object):
self._item_names.append(self._obis_name_mapping[obis_reference]) self._item_names.append(self._obis_name_mapping[obis_reference])
# Group Mbus related values into a MbusDevice object. # 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: if isinstance(dsmr_object, DSMRObject) and dsmr_object.is_mbus_reading:
channel_id = dsmr_object.obis_id_code[1] channel_id = dsmr_object.obis_id_code[1]
mbus_device = self._mbus_devices[channel_id] mbus_device = self._mbus_devices[channel_id]
mbus_device.add(obis_reference, dsmr_object) mbus_device.add(obis_reference, dsmr_object)
def get_mbus_devices(self): def get_mbus_devices(self):
""" """
Return MbusDevice objects which are used for water, heat and gas meters. Return MbusDevice objects which are used for water, heat and gas meters.
""" """
# TODO sort by channel ID return [d[1] for d in sorted(self._mbus_devices.items(), key=lambda x: x[0])]
return list(self._mbus_devices.values())
def get_mbus_device_by_channel(self, channel_id=None): def get_mbus_device_by_channel(self, channel_id=None):
return self._mbus_devices[channel_id] 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): def __getattr__(self, name):
""" will only get called for undefined attributes """ """ will only get called for undefined attributes """
obis_reference = self._reverse_obis_name_mapping[name] obis_reference = self._reverse_obis_name_mapping[name]
@ -126,6 +102,7 @@ class DSMRObject(object):
return obis_id == 0 and channel_id != 0 return obis_id == 0 and channel_id != 0
class MBusObject(DSMRObject): class MBusObject(DSMRObject):
@property @property

82
test/test_telegram.py
View File

@ -4,8 +4,7 @@ import pytz
from dsmr_parser import telegram_specifications from dsmr_parser import telegram_specifications
from dsmr_parser import obis_name_mapping 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 CosemObject, MbusDevice
from dsmr_parser.objects import MBusObject from dsmr_parser.objects import MBusObject
from dsmr_parser.objects import ProfileGenericObject from dsmr_parser.objects import ProfileGenericObject
from dsmr_parser.parsers import TelegramParser from dsmr_parser.parsers import TelegramParser
@ -327,84 +326,31 @@ class TelegramTest(unittest.TestCase):
self.assertEqual(len(telegram), 35) 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): def test_get_mbus_devices(self):
parser = TelegramParser(telegram_specifications.V5) parser = TelegramParser(telegram_specifications.V5)
telegram = parser.parse(TELEGRAM_V5_TWO_MBUS) telegram = parser.parse(TELEGRAM_V5_TWO_MBUS)
mbus_devices = telegram.get_mbus_devices() mbus_devices = telegram.get_mbus_devices()
print(mbus_devices)
self.assertEqual(len(mbus_devices), 2) self.assertEqual(len(mbus_devices), 2)
mbus_device_1 = mbus_devices[0] mbus_device_1 = mbus_devices[0]
self.assertEqual(type(mbus_device_1), MbusDevice) self.assertEqual(mbus_device_1.EQUIPMENT_IDENTIFIER_GAS.value, None)
print('mbus_device_1.HOURLY_GAS_METER_READING: ', mbus_device_1.HOURLY_GAS_METER_READING) print()
self.assertEqual(mbus_device_1.HOURLY_GAS_METER_READING.value, Decimal('0'))
mbus_device_2 = mbus_devices[1] mbus_device_2 = mbus_devices[1]
self.assertEqual(type(mbus_device_2), MbusDevice) self.assertEqual(mbus_device_2.EQUIPMENT_IDENTIFIER_GAS.value, '4730303339303031393336393930363139')
print('mbus_device_2.HOURLY_GAS_METER_READING: ', mbus_device_2.HOURLY_GAS_METER_READING) self.assertEqual(mbus_device_2.HOURLY_GAS_METER_READING.value, Decimal('246.138'))
# TODO
def test_get_mbus_device_by_channel(self): def test_get_mbus_device_by_channel(self):
parser = TelegramParser(telegram_specifications.V5) parser = TelegramParser(telegram_specifications.V5)
telegram = parser.parse(TELEGRAM_V5_TWO_MBUS) 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'))