issue-51-telegram removed Telegram.__getattr__ and refactored __iter__ as well

Browse Source
This commit is contained in:
Nigel Dokter 2023-02-13 21:47:59 +01:00
parent c7b9966083
commit 516850481d
2 changed files with 22 additions and 32 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

39
dsmr_parser/objects.py
View File

@ -2,10 +2,11 @@ from collections import defaultdict
from decimal import Decimal from decimal import Decimal
from operator import attrgetter from operator import attrgetter
import dsmr_parser.obis_name_mapping
import datetime import datetime
import json import json
from dsmr_parser import obis_name_mapping
class Telegram(object): class Telegram(object):
""" """
@ -23,16 +24,11 @@ class Telegram(object):
self._telegram_data = defaultdict(list) self._telegram_data = defaultdict(list)
self._mbus_channel_devices = {} self._mbus_channel_devices = {}
# Reverse name mapping and attribute related:
self._obis_name_mapping = dsmr_parser.obis_name_mapping.EN
self._reverse_obis_name_mapping = dsmr_parser.obis_name_mapping.REVERSE_EN
self._item_names = []
def add(self, obis_reference, dsmr_object): def add(self, obis_reference, dsmr_object):
self._telegram_data[obis_reference].append(dsmr_object) self._telegram_data[obis_reference].append(dsmr_object)
# Update name mapping used to get value by attribute. Example: telegram.P1_MESSAGE_HEADER # Update name mapping used to get value by attribute. Example: telegram.P1_MESSAGE_HEADER
self._item_names.append(self._obis_name_mapping[obis_reference]) setattr(self, obis_name_mapping.EN[obis_reference], dsmr_object)
# Group Mbus related values into a MbusDevice object. # Group Mbus related values into a MbusDevice object.
# TODO MaxDemandParser (BELGIUM_MAXIMUM_DEMAND_13_MONTHS) returns a list # TODO MaxDemandParser (BELGIUM_MAXIMUM_DEMAND_13_MONTHS) returns a list
@ -65,13 +61,6 @@ class Telegram(object):
def get_mbus_device_by_channel(self, channel_id): def get_mbus_device_by_channel(self, channel_id):
return self._mbus_channel_devices.get(channel_id) return self._mbus_channel_devices.get(channel_id)
def __getattr__(self, name):
""" will only get called for undefined attributes """
obis_reference = self._reverse_obis_name_mapping[name]
value = self._telegram_data[obis_reference][0]
setattr(self, name, value)
return value
def __getitem__(self, obis_reference): def __getitem__(self, obis_reference):
""" """
Get value by key. Example: telegram[obis_references.P1_MESSAGE_HEADER] Get value by key. Example: telegram[obis_references.P1_MESSAGE_HEADER]
@ -89,9 +78,11 @@ class Telegram(object):
return len(self._telegram_data) return len(self._telegram_data)
def __iter__(self): def __iter__(self):
for attr in self._item_names: for obis_reference, values in self._telegram_data.items():
value = getattr(self, attr) reverse_obis_name = obis_name_mapping.EN[obis_reference]
yield attr, value value = values[0] # TODO might be considered legacy behavior?
yield reverse_obis_name, value
def __str__(self): def __str__(self):
output = "" output = ""
@ -330,20 +321,8 @@ class MbusDevice:
self.channel_id = channel_id self.channel_id = channel_id
self._telegram_data = {} self._telegram_data = {}
# OBIS name mapping related used by __getattr__
self._obis_name_mapping = dsmr_parser.obis_name_mapping.EN
self._reverse_obis_name_mapping = dsmr_parser.obis_name_mapping.REVERSE_EN
self._item_names = []
def add(self, obis_reference, dsmr_object): def add(self, obis_reference, dsmr_object):
self._telegram_data[obis_reference] = dsmr_object self._telegram_data[obis_reference] = dsmr_object
# Update name mapping used to get value by attribute. Example: device.HOURLY_GAS_METER_READING # Update name mapping used to get value by attribute. Example: device.HOURLY_GAS_METER_READING
self._item_names.append(self._obis_name_mapping[obis_reference]) setattr(self, obis_name_mapping.EN[obis_reference], dsmr_object)
def __getattr__(self, name):
""" will only get called for undefined attributes """
obis_reference = self._reverse_obis_name_mapping[name]
value = self._telegram_data[obis_reference]
setattr(self, name, value)
return value

15
test/test_telegram.py
View File

@ -2,13 +2,13 @@ import unittest
import datetime import datetime
import pytz import pytz
from dsmr_parser import telegram_specifications from dsmr_parser import telegram_specifications, obis_references
from dsmr_parser import obis_name_mapping from dsmr_parser import obis_name_mapping
from dsmr_parser.objects import CosemObject from dsmr_parser.objects import CosemObject
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
from test.example_telegrams import TELEGRAM_V4_2, TELEGRAM_V5_TWO_MBUS from test.example_telegrams import TELEGRAM_V4_2, TELEGRAM_V5_TWO_MBUS, TELEGRAM_V5
from decimal import Decimal from decimal import Decimal
@ -326,6 +326,17 @@ class TelegramTest(unittest.TestCase):
self.assertEqual(len(telegram), 35) self.assertEqual(len(telegram), 35)
def test_iter(self):
parser = TelegramParser(telegram_specifications.V5)
telegram = parser.parse(TELEGRAM_V5)
for obis_reference, 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])
self.assertEqual(dsmr_object.value, '50')
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)