diff --git a/.gitignore b/.gitignore index 4dfc343..6789bb2 100644 --- a/.gitignore +++ b/.gitignore @@ -2,11 +2,13 @@ *.pyc .tox .cache +.venv *.egg-info /.project /.pydevproject /.coverage build/ dist/ +venv/ *.*~ *~ \ No newline at end of file diff --git a/CHANGELOG.rst b/CHANGELOG.rst index 0fda0e3..46a4645 100644 --- a/CHANGELOG.rst +++ b/CHANGELOG.rst @@ -1,5 +1,19 @@ Change Log ---------- +**0.20** (2020-05-12) + +- All objects can now print their values +- Add parser + object for generic profile + +**0.19** (2020-05-03) + +- Add following missing elements to telegram specification v4: + - SHORT_POWER_FAILURE_COUNT, + - INSTANTANEOUS_CURRENT_L1, + - INSTANTANEOUS_CURRENT_L2, + - INSTANTANEOUS_CURRENT_L3 +- Add missing tests + fix small test bugs +- Complete telegram object v4 parse test **0.18** (2020-01-28) @@ -36,7 +50,7 @@ Change Log **0.10** (2017-06-05) -- bugix: don't force full telegram signatures (`pull request #25 `_) +- bugfix: don't force full telegram signatures (`pull request #25 `_) - removed unused code for automatic telegram detection as this needs reworking after the fix mentioned above - InvalidChecksumError's are logged as warning instead of error diff --git a/dsmr_parser/objects.py b/dsmr_parser/objects.py index e313cd5..ce48a01 100644 --- a/dsmr_parser/objects.py +++ b/dsmr_parser/objects.py @@ -1,4 +1,5 @@ import dsmr_parser.obis_name_mapping +import datetime class Telegram(object): @@ -48,7 +49,7 @@ class Telegram(object): def __str__(self): output = "" for attr, value in self: - output += "{}: \t {} \t[{}]\n".format(attr, str(value.value), str(value.unit)) + output += "{}: \t {}\n".format(attr, str(value)) return output @@ -73,7 +74,7 @@ class MBusObject(DSMRObject): # TODO object, but let the parse set them differently? So don't use # TODO hardcoded indexes here. if len(self.values) != 2: # v2 - return self.values[5]['value'] + return self.values[6]['value'] else: return self.values[1]['value'] @@ -83,10 +84,14 @@ class MBusObject(DSMRObject): # TODO object, but let the parse set them differently? So don't use # TODO hardcoded indexes here. if len(self.values) != 2: # v2 - return self.values[4]['value'] + return self.values[5]['value'] else: return self.values[1]['unit'] + def __str__(self): + output = "{}\t[{}] at {}".format(str(self.value), str(self.unit), str(self.datetime.astimezone().isoformat())) + return output + class CosemObject(DSMRObject): @@ -98,6 +103,50 @@ class CosemObject(DSMRObject): def unit(self): return self.values[0]['unit'] + def __str__(self): + print_value = self.value + if isinstance(self.value, datetime.datetime): + print_value = self.value.astimezone().isoformat() + output = "{}\t[{}]".format(str(print_value), str(self.unit)) + return output -class ProfileGeneric(DSMRObject): - pass # TODO implement + +class ProfileGenericObject(DSMRObject): + """ + Represents all data in a GenericProfile value. + All buffer values are returned as a list of MBusObjects, + containing the datetime (timestamp) and the value. + """ + + def __init__(self, values): + super().__init__(values) + self._buffer_list = None + + @property + def buffer_length(self): + return self.values[0]['value'] + + @property + def buffer_type(self): + return self.values[1]['value'] + + @property + def buffer(self): + if self._buffer_list is None: + self._buffer_list = [] + values_offset = 2 + for i in range(self.buffer_length): + offset = values_offset + i*2 + self._buffer_list.append(MBusObject([self.values[offset], self.values[offset + 1]])) + return self._buffer_list + + def __str__(self): + output = "\t buffer length: {}\n".format(self.buffer_length) + output += "\t buffer type: {}".format(self.buffer_type) + for buffer_value in self.buffer: + timestamp = buffer_value.datetime + if isinstance(timestamp, datetime.datetime): + timestamp = str(timestamp.astimezone().isoformat()) + output += "\n\t event occured at: {}".format(timestamp) + output += "\t for: {} [{}]".format(buffer_value.value, buffer_value.unit) + return output diff --git a/dsmr_parser/parsers.py b/dsmr_parser/parsers.py index d9aeb5a..2c7c017 100644 --- a/dsmr_parser/parsers.py +++ b/dsmr_parser/parsers.py @@ -3,7 +3,7 @@ import re from ctypes import c_ushort -from dsmr_parser.objects import MBusObject, CosemObject +from dsmr_parser.objects import MBusObject, CosemObject, ProfileGenericObject from dsmr_parser.exceptions import ParseError, InvalidChecksumError logger = logging.getLogger(__name__) @@ -123,19 +123,28 @@ class DSMRObjectParser(object): def __init__(self, *value_formats): self.value_formats = value_formats + def _is_line_wellformed(self, line, values): + # allows overriding by child class + return (values and (len(values) == len(self.value_formats))) + + def _parse_values(self, values): + # allows overriding by child class + return [self.value_formats[i].parse(value) + for i, value in enumerate(values)] + def _parse(self, line): # Match value groups, but exclude the parentheses - pattern = re.compile(r'((?<=\()[0-9a-zA-Z\.\*]{0,}(?=\)))+') + pattern = re.compile(r'((?<=\()[0-9a-zA-Z\.\*\-\:]{0,}(?=\)))') + values = re.findall(pattern, line) + if not self._is_line_wellformed(line, values): + raise ParseError("Invalid '%s' line for '%s'", line, self) + # Convert empty value groups to None for clarity. values = [None if value == '' else value for value in values] - if not values or len(values) != len(self.value_formats): - raise ParseError("Invalid '%s' line for '%s'", line, self) - - return [self.value_formats[i].parse(value) - for i, value in enumerate(values)] + return self._parse_values(values) class MBusParser(DSMRObjectParser): @@ -204,9 +213,34 @@ class ProfileGenericParser(DSMRObjectParser): 8) Buffer value 2 (oldest entry of buffer attribute without unit) 9) Unit of buffer values (Unit of capture objects attribute) """ + def __init__(self, buffer_types, head_parsers, parsers_for_unidentified): + self.value_formats = head_parsers + self.buffer_types = buffer_types + self.parsers_for_unidentified = parsers_for_unidentified + + def _is_line_wellformed(self, line, values): + if values and (len(values) >= 2) and (values[0].isdigit()): + buffer_length = int(values[0]) + return (buffer_length <= 10) and (len(values) == (buffer_length * 2 + 2)) + else: + return False + + def _parse_values(self, values): + buffer_length = int(values[0]) + buffer_value_obis_ID = values[1] + if (buffer_length > 0): + if buffer_value_obis_ID in self.buffer_types: + bufferValueParsers = self.buffer_types[buffer_value_obis_ID] + else: + bufferValueParsers = self.parsers_for_unidentified + # add the parsers for the encountered value type z times + for _ in range(buffer_length): + self.value_formats.extend(bufferValueParsers) + + return [self.value_formats[i].parse(value) for i, value in enumerate(values)] def parse(self, line): - raise NotImplementedError() + return ProfileGenericObject(self._parse(line)) class ValueParser(object): diff --git a/dsmr_parser/profile_generic_specifications.py b/dsmr_parser/profile_generic_specifications.py new file mode 100644 index 0000000..a52416c --- /dev/null +++ b/dsmr_parser/profile_generic_specifications.py @@ -0,0 +1,10 @@ +from dsmr_parser.parsers import ValueParser +from dsmr_parser.value_types import timestamp + +PG_FAILURE_EVENT = r'0-0:96.7.19' + +PG_HEAD_PARSERS = [ValueParser(int), ValueParser(str)] +PG_UNIDENTIFIED_BUFFERTYPE_PARSERS = [ValueParser(str), ValueParser(str)] +BUFFER_TYPES = { + PG_FAILURE_EVENT: [ValueParser(timestamp), ValueParser(int)] + } diff --git a/dsmr_parser/telegram_specifications.py b/dsmr_parser/telegram_specifications.py index a42806f..1341ded 100644 --- a/dsmr_parser/telegram_specifications.py +++ b/dsmr_parser/telegram_specifications.py @@ -2,9 +2,9 @@ from decimal import Decimal from copy import deepcopy from dsmr_parser import obis_references as obis -from dsmr_parser.parsers import CosemParser, ValueParser, MBusParser +from dsmr_parser.parsers import CosemParser, ValueParser, MBusParser, ProfileGenericParser from dsmr_parser.value_types import timestamp - +from dsmr_parser.profile_generic_specifications import BUFFER_TYPES, PG_HEAD_PARSERS, PG_UNIDENTIFIED_BUFFERTYPE_PARSERS """ dsmr_parser.telegram_specifications @@ -37,8 +37,9 @@ V2_2 = { ValueParser(int), ValueParser(int), ValueParser(int), - ValueParser(str), - ValueParser(Decimal), + ValueParser(str), # obis ref + ValueParser(str), # unit, position 5 + ValueParser(Decimal), # meter reading, position 6 ), } } @@ -58,8 +59,12 @@ V4 = { obis.ELECTRICITY_ACTIVE_TARIFF: CosemParser(ValueParser(str)), obis.CURRENT_ELECTRICITY_USAGE: CosemParser(ValueParser(Decimal)), obis.CURRENT_ELECTRICITY_DELIVERY: CosemParser(ValueParser(Decimal)), + obis.SHORT_POWER_FAILURE_COUNT: CosemParser(ValueParser(int)), obis.LONG_POWER_FAILURE_COUNT: CosemParser(ValueParser(int)), - # POWER_EVENT_FAILURE_LOG: ProfileGenericParser(), TODO + obis.POWER_EVENT_FAILURE_LOG: + ProfileGenericParser(BUFFER_TYPES, + PG_HEAD_PARSERS, + PG_UNIDENTIFIED_BUFFERTYPE_PARSERS), obis.VOLTAGE_SAG_L1_COUNT: CosemParser(ValueParser(int)), obis.VOLTAGE_SAG_L2_COUNT: CosemParser(ValueParser(int)), obis.VOLTAGE_SAG_L3_COUNT: CosemParser(ValueParser(int)), @@ -69,6 +74,9 @@ V4 = { obis.TEXT_MESSAGE_CODE: CosemParser(ValueParser(int)), obis.TEXT_MESSAGE: CosemParser(ValueParser(str)), obis.DEVICE_TYPE: CosemParser(ValueParser(int)), + obis.INSTANTANEOUS_CURRENT_L1: CosemParser(ValueParser(Decimal)), + obis.INSTANTANEOUS_CURRENT_L2: CosemParser(ValueParser(Decimal)), + obis.INSTANTANEOUS_CURRENT_L3: CosemParser(ValueParser(Decimal)), obis.INSTANTANEOUS_ACTIVE_POWER_L1_POSITIVE: CosemParser(ValueParser(Decimal)), obis.INSTANTANEOUS_ACTIVE_POWER_L2_POSITIVE: CosemParser(ValueParser(Decimal)), obis.INSTANTANEOUS_ACTIVE_POWER_L3_POSITIVE: CosemParser(ValueParser(Decimal)), @@ -99,7 +107,10 @@ V5 = { obis.CURRENT_ELECTRICITY_DELIVERY: CosemParser(ValueParser(Decimal)), obis.LONG_POWER_FAILURE_COUNT: CosemParser(ValueParser(int)), obis.SHORT_POWER_FAILURE_COUNT: CosemParser(ValueParser(int)), - # POWER_EVENT_FAILURE_LOG: ProfileGenericParser(), TODO + obis.POWER_EVENT_FAILURE_LOG: + ProfileGenericParser(BUFFER_TYPES, + PG_HEAD_PARSERS, + PG_UNIDENTIFIED_BUFFERTYPE_PARSERS), obis.VOLTAGE_SAG_L1_COUNT: CosemParser(ValueParser(int)), obis.VOLTAGE_SAG_L2_COUNT: CosemParser(ValueParser(int)), obis.VOLTAGE_SAG_L3_COUNT: CosemParser(ValueParser(int)), diff --git a/setup.py b/setup.py index 9c47c51..c925b4d 100644 --- a/setup.py +++ b/setup.py @@ -6,7 +6,7 @@ setup( author='Nigel Dokter', author_email='nigel@nldr.net', url='https://github.com/ndokter/dsmr_parser', - version='0.18', + version='0.20', packages=find_packages(), install_requires=[ 'pyserial>=3,<4', diff --git a/test/test_parse_v4_2.py b/test/test_parse_v4_2.py index 681783b..cab34f7 100644 --- a/test/test_parse_v4_2.py +++ b/test/test_parse_v4_2.py @@ -80,6 +80,12 @@ class TelegramParserV4_2Test(unittest.TestCase): assert isinstance(result[obis.CURRENT_ELECTRICITY_DELIVERY].value, Decimal) assert result[obis.CURRENT_ELECTRICITY_DELIVERY].value == Decimal('0') + # SHORT_POWER_FAILURE_COUNT (1-0:96.7.21) + assert isinstance(result[obis.SHORT_POWER_FAILURE_COUNT], CosemObject) + assert result[obis.SHORT_POWER_FAILURE_COUNT].unit is None + assert isinstance(result[obis.SHORT_POWER_FAILURE_COUNT].value, int) + assert result[obis.SHORT_POWER_FAILURE_COUNT].value == 15 + # LONG_POWER_FAILURE_COUNT (96.7.9) assert isinstance(result[obis.LONG_POWER_FAILURE_COUNT], CosemObject) assert result[obis.LONG_POWER_FAILURE_COUNT].unit is None @@ -132,8 +138,26 @@ class TelegramParserV4_2Test(unittest.TestCase): assert result[obis.TEXT_MESSAGE].unit is None assert result[obis.TEXT_MESSAGE].value is None + # INSTANTANEOUS_CURRENT_L1 (1-0:31.7.0) + assert isinstance(result[obis.INSTANTANEOUS_CURRENT_L1], CosemObject) + assert result[obis.INSTANTANEOUS_CURRENT_L1].unit == 'A' + assert isinstance(result[obis.INSTANTANEOUS_CURRENT_L1].value, Decimal) + assert result[obis.INSTANTANEOUS_CURRENT_L1].value == Decimal('0') + + # INSTANTANEOUS_CURRENT_L2 (1-0:51.7.0) + assert isinstance(result[obis.INSTANTANEOUS_CURRENT_L2], CosemObject) + assert result[obis.INSTANTANEOUS_CURRENT_L2].unit == 'A' + assert isinstance(result[obis.INSTANTANEOUS_CURRENT_L2].value, Decimal) + assert result[obis.INSTANTANEOUS_CURRENT_L2].value == Decimal('6') + + # INSTANTANEOUS_CURRENT_L3 (1-0:71.7.0) + assert isinstance(result[obis.INSTANTANEOUS_CURRENT_L3], CosemObject) + assert result[obis.INSTANTANEOUS_CURRENT_L3].unit == 'A' + assert isinstance(result[obis.INSTANTANEOUS_CURRENT_L3].value, Decimal) + assert result[obis.INSTANTANEOUS_CURRENT_L3].value == Decimal('2') + # DEVICE_TYPE (0-x:24.1.0) - assert isinstance(result[obis.TEXT_MESSAGE], CosemObject) + assert isinstance(result[obis.DEVICE_TYPE], CosemObject) assert result[obis.DEVICE_TYPE].unit is None assert isinstance(result[obis.DEVICE_TYPE].value, int) assert result[obis.DEVICE_TYPE].value == 3 diff --git a/test/test_parse_v5.py b/test/test_parse_v5.py index e9cfbc1..67d7cd8 100644 --- a/test/test_parse_v5.py +++ b/test/test_parse_v5.py @@ -171,7 +171,7 @@ class TelegramParserV5Test(unittest.TestCase): assert result[obis.TEXT_MESSAGE].value is None # DEVICE_TYPE (0-x:24.1.0) - assert isinstance(result[obis.TEXT_MESSAGE], CosemObject) + assert isinstance(result[obis.DEVICE_TYPE], CosemObject) assert result[obis.DEVICE_TYPE].unit is None assert isinstance(result[obis.DEVICE_TYPE].value, int) assert result[obis.DEVICE_TYPE].value == 3 diff --git a/test/test_telegram.py b/test/test_telegram.py index ea85704..90b8eff 100644 --- a/test/test_telegram.py +++ b/test/test_telegram.py @@ -1,21 +1,322 @@ import unittest +import datetime +import pytz from dsmr_parser import telegram_specifications +from dsmr_parser import obis_name_mapping from dsmr_parser.objects import CosemObject +from dsmr_parser.objects import MBusObject from dsmr_parser.objects import Telegram +from dsmr_parser.objects import ProfileGenericObject from dsmr_parser.parsers import TelegramParser from test.example_telegrams import TELEGRAM_V4_2 +from decimal import Decimal class TelegramTest(unittest.TestCase): """ Test instantiation of Telegram object """ + def __init__(self, *args, **kwargs): + self.item_names_tested = [] + super(TelegramTest, self).__init__(*args, **kwargs) + + def verify_telegram_item(self, telegram, testitem_name, object_type, unit_val, value_type, value_val): + testitem = eval("telegram.{}".format(testitem_name)) + assert isinstance(testitem, object_type) + assert testitem.unit == unit_val + assert isinstance(testitem.value, value_type) + assert testitem.value == value_val + self.item_names_tested.append(testitem_name) + def test_instantiate(self): parser = TelegramParser(telegram_specifications.V4) telegram = Telegram(TELEGRAM_V4_2, parser, telegram_specifications.V4) # P1_MESSAGE_HEADER (1-3:0.2.8) - testitem = telegram.P1_MESSAGE_HEADER - assert isinstance(testitem, CosemObject) - assert testitem.unit is None - assert testitem.value == '42' + self.verify_telegram_item(telegram, + 'P1_MESSAGE_HEADER', + object_type=CosemObject, + unit_val=None, + value_type=str, + value_val='42') + + # P1_MESSAGE_TIMESTAMP (0-0:1.0.0) + self.verify_telegram_item(telegram, + 'P1_MESSAGE_TIMESTAMP', + CosemObject, + unit_val=None, + value_type=datetime.datetime, + value_val=datetime.datetime(2016, 11, 13, 19, 57, 57, tzinfo=pytz.UTC)) + + # ELECTRICITY_USED_TARIFF_1 (1-0:1.8.1) + self.verify_telegram_item(telegram, + 'ELECTRICITY_USED_TARIFF_1', + object_type=CosemObject, + unit_val='kWh', + value_type=Decimal, + value_val=Decimal('1581.123')) + + # ELECTRICITY_USED_TARIFF_2 (1-0:1.8.2) + self.verify_telegram_item(telegram, + 'ELECTRICITY_USED_TARIFF_2', + object_type=CosemObject, + unit_val='kWh', + value_type=Decimal, + value_val=Decimal('1435.706')) + + # ELECTRICITY_DELIVERED_TARIFF_1 (1-0:2.8.1) + self.verify_telegram_item(telegram, + 'ELECTRICITY_DELIVERED_TARIFF_1', + object_type=CosemObject, + unit_val='kWh', + value_type=Decimal, + value_val=Decimal('0')) + + # ELECTRICITY_DELIVERED_TARIFF_2 (1-0:2.8.2) + self.verify_telegram_item(telegram, + 'ELECTRICITY_DELIVERED_TARIFF_2', + object_type=CosemObject, + unit_val='kWh', + value_type=Decimal, + value_val=Decimal('0')) + + # ELECTRICITY_ACTIVE_TARIFF (0-0:96.14.0) + self.verify_telegram_item(telegram, + 'ELECTRICITY_ACTIVE_TARIFF', + object_type=CosemObject, + unit_val=None, + value_type=str, + value_val='0002') + + # EQUIPMENT_IDENTIFIER (0-0:96.1.1) + self.verify_telegram_item(telegram, + 'EQUIPMENT_IDENTIFIER', + object_type=CosemObject, + unit_val=None, + value_type=str, + value_val='3960221976967177082151037881335713') + + # CURRENT_ELECTRICITY_USAGE (1-0:1.7.0) + self.verify_telegram_item(telegram, + 'CURRENT_ELECTRICITY_USAGE', + object_type=CosemObject, + unit_val='kW', + value_type=Decimal, + value_val=Decimal('2.027')) + + # CURRENT_ELECTRICITY_DELIVERY (1-0:2.7.0) + self.verify_telegram_item(telegram, + 'CURRENT_ELECTRICITY_DELIVERY', + object_type=CosemObject, + unit_val='kW', + value_type=Decimal, + value_val=Decimal('0')) + + # SHORT_POWER_FAILURE_COUNT (1-0:96.7.21) + self.verify_telegram_item(telegram, + 'SHORT_POWER_FAILURE_COUNT', + object_type=CosemObject, + unit_val=None, + value_type=int, + value_val=15) + + # LONG_POWER_FAILURE_COUNT (96.7.9) + self.verify_telegram_item(telegram, + 'LONG_POWER_FAILURE_COUNT', + object_type=CosemObject, + unit_val=None, + value_type=int, + value_val=7) + + # VOLTAGE_SAG_L1_COUNT (1-0:32.32.0) + self.verify_telegram_item(telegram, + 'VOLTAGE_SAG_L1_COUNT', + object_type=CosemObject, + unit_val=None, + value_type=int, + value_val=0) + + # VOLTAGE_SAG_L2_COUNT (1-0:52.32.0) + self.verify_telegram_item(telegram, + 'VOLTAGE_SAG_L2_COUNT', + object_type=CosemObject, + unit_val=None, + value_type=int, + value_val=0) + + # VOLTAGE_SAG_L3_COUNT (1-0:72.32.0) + self.verify_telegram_item(telegram, + 'VOLTAGE_SAG_L3_COUNT', + object_type=CosemObject, + unit_val=None, + value_type=int, + value_val=0) + + # VOLTAGE_SWELL_L1_COUNT (1-0:32.36.0) + self.verify_telegram_item(telegram, + 'VOLTAGE_SWELL_L1_COUNT', + object_type=CosemObject, + unit_val=None, + value_type=int, + value_val=0) + + # VOLTAGE_SWELL_L2_COUNT (1-0:52.36.0) + self.verify_telegram_item(telegram, + 'VOLTAGE_SWELL_L2_COUNT', + object_type=CosemObject, + unit_val=None, + value_type=int, + value_val=0) + + # VOLTAGE_SWELL_L3_COUNT (1-0:72.36.0) + self.verify_telegram_item(telegram, + 'VOLTAGE_SWELL_L3_COUNT', + object_type=CosemObject, + unit_val=None, + value_type=int, + value_val=0) + + # TEXT_MESSAGE_CODE (0-0:96.13.1) + self.verify_telegram_item(telegram, + 'TEXT_MESSAGE_CODE', + object_type=CosemObject, + unit_val=None, + value_type=type(None), + value_val=None) + + # TEXT_MESSAGE (0-0:96.13.0) + self.verify_telegram_item(telegram, + 'TEXT_MESSAGE', + object_type=CosemObject, + unit_val=None, + value_type=type(None), + value_val=None) + + # INSTANTANEOUS_CURRENT_L1 (1-0:31.7.0) + self.verify_telegram_item(telegram, + 'INSTANTANEOUS_CURRENT_L1', + object_type=CosemObject, + unit_val='A', + value_type=Decimal, + value_val=Decimal('0')) + + # INSTANTANEOUS_CURRENT_L2 (1-0:51.7.0) + self.verify_telegram_item(telegram, + 'INSTANTANEOUS_CURRENT_L2', + object_type=CosemObject, + unit_val='A', + value_type=Decimal, + value_val=Decimal('6')) + + # INSTANTANEOUS_CURRENT_L3 (1-0:71.7.0) + self.verify_telegram_item(telegram, + 'INSTANTANEOUS_CURRENT_L3', + object_type=CosemObject, + unit_val='A', + value_type=Decimal, + value_val=Decimal('2')) + + # DEVICE_TYPE (0-x:24.1.0) + self.verify_telegram_item(telegram, + 'DEVICE_TYPE', + object_type=CosemObject, + unit_val=None, + value_type=int, + value_val=3) + + # INSTANTANEOUS_ACTIVE_POWER_L1_POSITIVE (1-0:21.7.0) + self.verify_telegram_item(telegram, + 'INSTANTANEOUS_ACTIVE_POWER_L1_POSITIVE', + object_type=CosemObject, + unit_val='kW', + value_type=Decimal, + value_val=Decimal('0.170')) + + # INSTANTANEOUS_ACTIVE_POWER_L2_POSITIVE (1-0:41.7.0) + self.verify_telegram_item(telegram, + 'INSTANTANEOUS_ACTIVE_POWER_L2_POSITIVE', + object_type=CosemObject, + unit_val='kW', + value_type=Decimal, + value_val=Decimal('1.247')) + + # INSTANTANEOUS_ACTIVE_POWER_L3_POSITIVE (1-0:61.7.0) + self.verify_telegram_item(telegram, + 'INSTANTANEOUS_ACTIVE_POWER_L3_POSITIVE', + object_type=CosemObject, + unit_val='kW', + value_type=Decimal, + value_val=Decimal('0.209')) + + # INSTANTANEOUS_ACTIVE_POWER_L1_NEGATIVE (1-0:22.7.0) + self.verify_telegram_item(telegram, + 'INSTANTANEOUS_ACTIVE_POWER_L1_NEGATIVE', + object_type=CosemObject, + unit_val='kW', + value_type=Decimal, + value_val=Decimal('0')) + + # INSTANTANEOUS_ACTIVE_POWER_L2_NEGATIVE (1-0:42.7.0) + self.verify_telegram_item(telegram, + 'INSTANTANEOUS_ACTIVE_POWER_L2_NEGATIVE', + object_type=CosemObject, + unit_val='kW', + value_type=Decimal, + value_val=Decimal('0')) + + # INSTANTANEOUS_ACTIVE_POWER_L3_NEGATIVE (1-0:62.7.0) + self.verify_telegram_item(telegram, + 'INSTANTANEOUS_ACTIVE_POWER_L3_NEGATIVE', + object_type=CosemObject, + unit_val='kW', + value_type=Decimal, + value_val=Decimal('0')) + + # EQUIPMENT_IDENTIFIER_GAS (0-x:96.1.0) + self.verify_telegram_item(telegram, + 'EQUIPMENT_IDENTIFIER_GAS', + object_type=CosemObject, + unit_val=None, + value_type=str, + value_val='4819243993373755377509728609491464') + + # HOURLY_GAS_METER_READING (0-1:24.2.1) + self.verify_telegram_item(telegram, + 'HOURLY_GAS_METER_READING', + object_type=MBusObject, + unit_val='m3', + value_type=Decimal, + value_val=Decimal('981.443')) + + # POWER_EVENT_FAILURE_LOG (1-0:99.97.0) + testitem_name = 'POWER_EVENT_FAILURE_LOG' + object_type = ProfileGenericObject + testitem = eval("telegram.{}".format(testitem_name)) + assert isinstance(testitem, object_type) + assert testitem.buffer_length == 3 + assert testitem.buffer_type == '0-0:96.7.19' + buffer = testitem.buffer + assert isinstance(testitem.buffer, list) + assert len(buffer) == 3 + assert all([isinstance(item, MBusObject) for item in buffer]) + date0 = datetime.datetime(2000, 1, 4, 17, 3, 20, tzinfo=datetime.timezone.utc) + date1 = datetime.datetime(1999, 12, 31, 23, 0, 1, tzinfo=datetime.timezone.utc) + date2 = datetime.datetime(2000, 1, 1, 23, 0, 3, tzinfo=datetime.timezone.utc) + assert buffer[0].datetime == date0 + assert buffer[1].datetime == date1 + assert buffer[2].datetime == date2 + assert buffer[0].value == 237126 + assert buffer[1].value == 2147583646 + assert buffer[2].value == 2317482647 + assert all([isinstance(item.value, int) for item in buffer]) + assert all([isinstance(item.unit, str) for item in buffer]) + assert all([(item.unit == 's') for item in buffer]) + self.item_names_tested.append(testitem_name) + + # check if all items in telegram V4 specification are covered + V4_name_list = [obis_name_mapping.EN[signature] for signature, parser in + telegram_specifications.V4['objects'].items()] + V4_name_set = set(V4_name_list) + item_names_tested_set = set(self.item_names_tested) + + assert item_names_tested_set == V4_name_set