issue-51-telegram work in progress

Browse Source
This commit is contained in:
Nigel Dokter 2023-02-07 19:48:56 +01:00
parent f92836da2f
commit ef66c8422c
3 changed files with 98 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
dsmr_parser/objects.py
View File

@ -27,15 +27,11 @@ class Telegram(object):
def add(self, obis_reference, dsmr_object):
self._telegram_data[obis_reference].append(dsmr_object)
# TODO experiment with api to see what is nice
def get(self, obis_reference, channel=None):
if channel is None:
return self._telegram_data[obis_reference]
def get_by_channel(self, obis_reference, channel):
try:
return next(filter(lambda x: x.channel == channel, self._telegram_data[obis_reference]))
except StopIteration:
return None
raise LookupError('No value found for OBIS reference "{}" on channel "{}"'.format(obis_reference, channel))
def __getattr__(self, name):
""" will only get called for undefined attributes """
@ -45,7 +41,11 @@ class Telegram(object):
return value
def __getitem__(self, obis_reference):
return self._telegram_data[obis_reference][0]
try:
return self._telegram_data[obis_reference][0]
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) # TODO: its nested now

45
test/example_telegrams.py
View File

@ -129,6 +129,51 @@ TELEGRAM_V5 = (
'!6EEE\r\n'
)
# V5 telegram with 2 MBUS devices
TELEGRAM_V5_TWO_MBUS = (
'/ISK5\\2M550T-1012\r\n'
'\r\n'
'1-3:0.2.8(50)\r\n'
'0-0:1.0.0(200426223325S)\r\n'
'0-0:96.1.1(4530303434303037333832323436303139)\r\n'
'1-0:1.8.1(002130.115*kWh)\r\n'
'1-0:1.8.2(000245.467*kWh)\r\n'
'1-0:2.8.1(000000.000*kWh)\r\n'
'1-0:2.8.2(000000.000*kWh)\r\n'
'0-0:96.14.0(0001)\r\n'
'1-0:1.7.0(00.111*kW)\r\n'
'1-0:2.7.0(00.000*kW)\r\n'
'0-0:96.7.21(00005)\r\n'
'0-0:96.7.9(00003)\r\n'
'1-0:99.97.0(1)(0-0:96.7.19)(190326095015W)(0000002014*s)\r\n'
'1-0:32.32.0(00001)\r\n'
'1-0:52.32.0(00001)\r\n'
'1-0:72.32.0(00192)\r\n'
'1-0:32.36.0(00001)\r\n'
'1-0:52.36.0(00001)\r\n'
'1-0:72.36.0(00001)\r\n'
'0-0:96.13.0()\r\n'
'1-0:32.7.0(229.9*V)\r\n'
'1-0:52.7.0(229.2*V)\r\n'
'1-0:72.7.0(222.9*V)\r\n'
'1-0:31.7.0(000*A)\r\n'
'1-0:51.7.0(000*A)\r\n'
'1-0:71.7.0(001*A)\r\n'
'1-0:21.7.0(00.056*kW)\r\n'
'1-0:41.7.0(00.000*kW)\r\n'
'1-0:61.7.0(00.055*kW)\r\n'
'1-0:22.7.0(00.000*kW)\r\n'
'1-0:42.7.0(00.000*kW)\r\n'
'1-0:62.7.0(00.000*kW)\r\n'
'0-1:24.1.0(003)\r\n'
'0-1:96.1.0()\r\n'
'0-1:24.2.1(700101010000W)(00000000)\r\n'
'0-2:24.1.0(003)\r\n'
'0-2:96.1.0(4730303339303031393336393930363139)\r\n'
'0-2:24.2.1(200426223001S)(00246.138*m3)\r\n'
'!56DD\r\n'
)
TELEGRAM_FLUVIUS_V171 = (
'/FLU5\253769484_A\r\n'
'\r\n'

47
test/test_telegram.py
View File

@ -4,11 +4,12 @@ import pytz
from dsmr_parser import telegram_specifications
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 MBusObject
from dsmr_parser.objects import ProfileGenericObject
from dsmr_parser.parsers import TelegramParser
from test.example_telegrams import TELEGRAM_V4_2
from test.example_telegrams import TELEGRAM_V4_2, TELEGRAM_V5_TWO_MBUS
from decimal import Decimal
@ -319,3 +320,47 @@ class TelegramTest(unittest.TestCase):
item_names_tested_set = set(self.item_names_tested)
assert item_names_tested_set == V4_name_set
def test_get(self):
parser = TelegramParser(telegram_specifications.V5)
telegram = parser.parse(TELEGRAM_V5_TWO_MBUS)
mbus_1 = telegram.get(obis.HOURLY_GAS_METER_READING, channel=1)
mbus_2 = telegram.get(obis.HOURLY_GAS_METER_READING, channel=2)
self.assertEqual(type(mbus_1), MBusObject)
self.assertEqual(mbus_1.channel, 1)
self.assertEqual(mbus_1.value, 0)
self.assertEqual(type(mbus_2), MBusObject)
self.assertEqual(mbus_2.channel, 2)
self.assertEqual(mbus_2.value, Decimal('246.138'))
def test_get_without_channel(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)
mbus = telegram.get(obis.HOURLY_GAS_METER_READING)
self.assertEqual(type(mbus), MBusObject)
self.assertEqual(mbus.channel, 1)
self.assertEqual(mbus.value, 0)
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_by_channel(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_by_channel('invalid_obis_reference', channel=1)