From b527e991ef7ad9c3000863bdd8712500c39eb059 Mon Sep 17 00:00:00 2001 From: Nigel Dokter Date: Sun, 12 Feb 2023 11:14:47 +0100 Subject: [PATCH] issue-51-telegram updated documentation; refactored V5 unittest to use telegram attributes instead of keys; --- README.rst | 35 +++--- test/test_parse_v5.py | 284 ++++++++++++++++++++++-------------------- test/test_telegram.py | 4 + 3 files changed, 168 insertions(+), 155 deletions(-) diff --git a/README.rst b/README.rst index ec5d3a1..9490bdd 100644 --- a/README.rst +++ b/README.rst @@ -171,6 +171,8 @@ Parsing module usage The parsing module accepts complete unaltered telegram strings and parses these into a Telegram object. This previously was a dictionary, but the Telegram object is mostly compatible. +Getting full telegrams from a bytestream can be made easier by using the TelegramBuffer helper class. + .. code-block:: python from dsmr_parser import telegram_specifications @@ -210,42 +212,39 @@ into a Telegram object. This previously was a dictionary, but the Telegram objec Telegram object --------------------- +A Telegram has attributes for all the parsed values according to the given telegram specification. Each value is a DsmrObject which have a 'value' and 'unit' property. MBusObject's, which are DsmrObject's as well additionally have a 'datetime' property. The 'value' can contain any python type (int, str, Decimal) depending on the field. The 'unit' contains 'kW', 'A', 'kWh' or 'm3'. + +Below are some examples on how to get the meter data. Alternatively check out the following unit test for a complete example: TelegramParserV5Test.test_parse + .. code-block:: python - # DSMR v5 telegram example - from dsmr_parser import telegram_specifications - from dsmr_parser.parsers import TelegramParser - from test.example_telegrams import TELEGRAM_V5 - - parser = TelegramParser(telegram_specifications.V5) - telegram = parser.parse(TELEGRAM_V5) - # Print contents of all available values # See dsmr_parser.obis_name_mapping for all readable telegram values. # The available values differ per DSMR version and meter. print(telegram) - # P1_MESSAGE_HEADER: 42 [None] - # P1_MESSAGE_TIMESTAMP: 2016-11-13 19:57:57+00:00 [None] - # EQUIPMENT_IDENTIFIER: 3960221976967177082151037881335713 [None] - # ELECTRICITY_USED_TARIFF_1: 1581.123 [kWh] + # P1_MESSAGE_HEADER: 42 [None] + # P1_MESSAGE_TIMESTAMP: 2016-11-13 19:57:57+00:00 [None] + # EQUIPMENT_IDENTIFIER: 3960221976967177082151037881335713 [None] + # ELECTRICITY_USED_TARIFF_1: 1581.123 [kWh] # etc. # Example to get current electricity usage - print(telegram.CURRENT_ELECTRICITY_USAGE) - print(telegram.CURRENT_ELECTRICITY_USAGE.value) - print(telegram.CURRENT_ELECTRICITY_USAGE.unit) - # - # Decimal('2.027') - # 'kW' + print(telegram.CURRENT_ELECTRICITY_USAGE) # + print(telegram.CURRENT_ELECTRICITY_USAGE.value) # Decimal('2.027') + print(telegram.CURRENT_ELECTRICITY_USAGE.unit) # 'kW' # All Mbus device readings like gas meters and water meters can be retrieved as follows: mbus_devices = telegram.get_mbus_devices() # A specific device based on the channel the device is connected to can be retrieved as follows: mbus_device = telegram.get_mbus_device_by_channel(1) + print(mbus_device.DEVICE_TYPE.value) # 3 print(mbus_device.EQUIPMENT_IDENTIFIER_GAS.value) # '4730303339303031393336393930363139' print(mbus_device.HOURLY_GAS_METER_READING.value) # Decimal('246.138') + # A deprecated way of getting the values is by key + telegram[obis_references.CURRENT_ELECTRICITY_USAGE] + Installation ------------ diff --git a/test/test_parse_v5.py b/test/test_parse_v5.py index fe3ed84..0d4896c 100644 --- a/test/test_parse_v5.py +++ b/test/test_parse_v5.py @@ -5,7 +5,6 @@ import unittest import pytz -from dsmr_parser import obis_references as obis from dsmr_parser import telegram_specifications from dsmr_parser.exceptions import InvalidChecksumError, ParseError from dsmr_parser.objects import CosemObject, MBusObject @@ -18,211 +17,222 @@ class TelegramParserV5Test(unittest.TestCase): def test_parse(self): parser = TelegramParser(telegram_specifications.V5) - result = parser.parse(TELEGRAM_V5) + telegram = parser.parse(TELEGRAM_V5) # P1_MESSAGE_HEADER (1-3:0.2.8) - assert isinstance(result[obis.P1_MESSAGE_HEADER], CosemObject) - assert result[obis.P1_MESSAGE_HEADER].unit is None - assert isinstance(result[obis.P1_MESSAGE_HEADER].value, str) - assert result[obis.P1_MESSAGE_HEADER].value == '50' + assert isinstance(telegram.P1_MESSAGE_HEADER, CosemObject) + assert telegram.P1_MESSAGE_HEADER.unit is None + assert isinstance(telegram.P1_MESSAGE_HEADER.value, str) + assert telegram.P1_MESSAGE_HEADER.value == '50' # P1_MESSAGE_TIMESTAMP (0-0:1.0.0) - assert isinstance(result[obis.P1_MESSAGE_TIMESTAMP], CosemObject) - assert result[obis.P1_MESSAGE_TIMESTAMP].unit is None - assert isinstance(result[obis.P1_MESSAGE_TIMESTAMP].value, datetime.datetime) - assert result[obis.P1_MESSAGE_TIMESTAMP].value == \ + assert isinstance(telegram.P1_MESSAGE_TIMESTAMP, CosemObject) + assert telegram.P1_MESSAGE_TIMESTAMP.unit is None + assert isinstance(telegram.P1_MESSAGE_TIMESTAMP.value, datetime.datetime) + assert telegram.P1_MESSAGE_TIMESTAMP.value == \ datetime.datetime(2017, 1, 2, 18, 20, 2, tzinfo=pytz.UTC) # ELECTRICITY_USED_TARIFF_1 (1-0:1.8.1) - assert isinstance(result[obis.ELECTRICITY_USED_TARIFF_1], CosemObject) - assert result[obis.ELECTRICITY_USED_TARIFF_1].unit == 'kWh' - assert isinstance(result[obis.ELECTRICITY_USED_TARIFF_1].value, Decimal) - assert result[obis.ELECTRICITY_USED_TARIFF_1].value == Decimal('4.426') + assert isinstance(telegram.ELECTRICITY_USED_TARIFF_1, CosemObject) + assert telegram.ELECTRICITY_USED_TARIFF_1.unit == 'kWh' + assert isinstance(telegram.ELECTRICITY_USED_TARIFF_1.value, Decimal) + assert telegram.ELECTRICITY_USED_TARIFF_1.value == Decimal('4.426') # ELECTRICITY_USED_TARIFF_2 (1-0:1.8.2) - assert isinstance(result[obis.ELECTRICITY_USED_TARIFF_2], CosemObject) - assert result[obis.ELECTRICITY_USED_TARIFF_2].unit == 'kWh' - assert isinstance(result[obis.ELECTRICITY_USED_TARIFF_2].value, Decimal) - assert result[obis.ELECTRICITY_USED_TARIFF_2].value == Decimal('2.399') + assert isinstance(telegram.ELECTRICITY_USED_TARIFF_2, CosemObject) + assert telegram.ELECTRICITY_USED_TARIFF_2.unit == 'kWh' + assert isinstance(telegram.ELECTRICITY_USED_TARIFF_2.value, Decimal) + assert telegram.ELECTRICITY_USED_TARIFF_2.value == Decimal('2.399') # ELECTRICITY_DELIVERED_TARIFF_1 (1-0:2.8.1) - assert isinstance(result[obis.ELECTRICITY_DELIVERED_TARIFF_1], CosemObject) - assert result[obis.ELECTRICITY_DELIVERED_TARIFF_1].unit == 'kWh' - assert isinstance(result[obis.ELECTRICITY_DELIVERED_TARIFF_1].value, Decimal) - assert result[obis.ELECTRICITY_DELIVERED_TARIFF_1].value == Decimal('2.444') + assert isinstance(telegram.ELECTRICITY_DELIVERED_TARIFF_1, CosemObject) + assert telegram.ELECTRICITY_DELIVERED_TARIFF_1.unit == 'kWh' + assert isinstance(telegram.ELECTRICITY_DELIVERED_TARIFF_1.value, Decimal) + assert telegram.ELECTRICITY_DELIVERED_TARIFF_1.value == Decimal('2.444') # ELECTRICITY_DELIVERED_TARIFF_2 (1-0:2.8.2) - assert isinstance(result[obis.ELECTRICITY_DELIVERED_TARIFF_2], CosemObject) - assert result[obis.ELECTRICITY_DELIVERED_TARIFF_2].unit == 'kWh' - assert isinstance(result[obis.ELECTRICITY_DELIVERED_TARIFF_2].value, Decimal) - assert result[obis.ELECTRICITY_DELIVERED_TARIFF_2].value == Decimal('0') + assert isinstance(telegram.ELECTRICITY_DELIVERED_TARIFF_2, CosemObject) + assert telegram.ELECTRICITY_DELIVERED_TARIFF_2.unit == 'kWh' + assert isinstance(telegram.ELECTRICITY_DELIVERED_TARIFF_2.value, Decimal) + assert telegram.ELECTRICITY_DELIVERED_TARIFF_2.value == Decimal('0') # ELECTRICITY_ACTIVE_TARIFF (0-0:96.14.0) - assert isinstance(result[obis.ELECTRICITY_ACTIVE_TARIFF], CosemObject) - assert result[obis.ELECTRICITY_ACTIVE_TARIFF].unit is None - assert isinstance(result[obis.ELECTRICITY_ACTIVE_TARIFF].value, str) - assert result[obis.ELECTRICITY_ACTIVE_TARIFF].value == '0002' + assert isinstance(telegram.ELECTRICITY_ACTIVE_TARIFF, CosemObject) + assert telegram.ELECTRICITY_ACTIVE_TARIFF.unit is None + assert isinstance(telegram.ELECTRICITY_ACTIVE_TARIFF.value, str) + assert telegram.ELECTRICITY_ACTIVE_TARIFF.value == '0002' # EQUIPMENT_IDENTIFIER (0-0:96.1.1) - assert isinstance(result[obis.EQUIPMENT_IDENTIFIER], CosemObject) - assert result[obis.EQUIPMENT_IDENTIFIER].unit is None - assert isinstance(result[obis.EQUIPMENT_IDENTIFIER].value, str) - assert result[obis.EQUIPMENT_IDENTIFIER].value == '4B384547303034303436333935353037' + assert isinstance(telegram.EQUIPMENT_IDENTIFIER, CosemObject) + assert telegram.EQUIPMENT_IDENTIFIER.unit is None + assert isinstance(telegram.EQUIPMENT_IDENTIFIER.value, str) + assert telegram.EQUIPMENT_IDENTIFIER.value == '4B384547303034303436333935353037' # CURRENT_ELECTRICITY_USAGE (1-0:1.7.0) - assert isinstance(result[obis.CURRENT_ELECTRICITY_USAGE], CosemObject) - assert result[obis.CURRENT_ELECTRICITY_USAGE].unit == 'kW' - assert isinstance(result[obis.CURRENT_ELECTRICITY_USAGE].value, Decimal) - assert result[obis.CURRENT_ELECTRICITY_USAGE].value == Decimal('0.244') + assert isinstance(telegram.CURRENT_ELECTRICITY_USAGE, CosemObject) + assert telegram.CURRENT_ELECTRICITY_USAGE.unit == 'kW' + assert isinstance(telegram.CURRENT_ELECTRICITY_USAGE.value, Decimal) + assert telegram.CURRENT_ELECTRICITY_USAGE.value == Decimal('0.244') # CURRENT_ELECTRICITY_DELIVERY (1-0:2.7.0) - assert isinstance(result[obis.CURRENT_ELECTRICITY_DELIVERY], CosemObject) - assert result[obis.CURRENT_ELECTRICITY_DELIVERY].unit == 'kW' - assert isinstance(result[obis.CURRENT_ELECTRICITY_DELIVERY].value, Decimal) - assert result[obis.CURRENT_ELECTRICITY_DELIVERY].value == Decimal('0') + assert isinstance(telegram.CURRENT_ELECTRICITY_DELIVERY, CosemObject) + assert telegram.CURRENT_ELECTRICITY_DELIVERY.unit == 'kW' + assert isinstance(telegram.CURRENT_ELECTRICITY_DELIVERY.value, Decimal) + assert telegram.CURRENT_ELECTRICITY_DELIVERY.value == Decimal('0') # 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 - assert isinstance(result[obis.LONG_POWER_FAILURE_COUNT].value, int) - assert result[obis.LONG_POWER_FAILURE_COUNT].value == 0 + assert isinstance(telegram.LONG_POWER_FAILURE_COUNT, CosemObject) + assert telegram.LONG_POWER_FAILURE_COUNT.unit is None + assert isinstance(telegram.LONG_POWER_FAILURE_COUNT.value, int) + assert telegram.LONG_POWER_FAILURE_COUNT.value == 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 == 13 + assert isinstance(telegram.SHORT_POWER_FAILURE_COUNT, CosemObject) + assert telegram.SHORT_POWER_FAILURE_COUNT.unit is None + assert isinstance(telegram.SHORT_POWER_FAILURE_COUNT.value, int) + assert telegram.SHORT_POWER_FAILURE_COUNT.value == 13 # VOLTAGE_SAG_L1_COUNT (1-0:32.32.0) - assert isinstance(result[obis.VOLTAGE_SAG_L1_COUNT], CosemObject) - assert result[obis.VOLTAGE_SAG_L1_COUNT].unit is None - assert isinstance(result[obis.VOLTAGE_SAG_L1_COUNT].value, int) - assert result[obis.VOLTAGE_SAG_L1_COUNT].value == 0 + assert isinstance(telegram.VOLTAGE_SAG_L1_COUNT, CosemObject) + assert telegram.VOLTAGE_SAG_L1_COUNT.unit is None + assert isinstance(telegram.VOLTAGE_SAG_L1_COUNT.value, int) + assert telegram.VOLTAGE_SAG_L1_COUNT.value == 0 # VOLTAGE_SAG_L2_COUNT (1-0:52.32.0) - assert isinstance(result[obis.VOLTAGE_SAG_L2_COUNT], CosemObject) - assert result[obis.VOLTAGE_SAG_L2_COUNT].unit is None - assert isinstance(result[obis.VOLTAGE_SAG_L2_COUNT].value, int) - assert result[obis.VOLTAGE_SAG_L2_COUNT].value == 0 + assert isinstance(telegram.VOLTAGE_SAG_L2_COUNT, CosemObject) + assert telegram.VOLTAGE_SAG_L2_COUNT.unit is None + assert isinstance(telegram.VOLTAGE_SAG_L2_COUNT.value, int) + assert telegram.VOLTAGE_SAG_L2_COUNT.value == 0 # VOLTAGE_SAG_L3_COUNT (1-0:72.32.0) - assert isinstance(result[obis.VOLTAGE_SAG_L3_COUNT], CosemObject) - assert result[obis.VOLTAGE_SAG_L3_COUNT].unit is None - assert isinstance(result[obis.VOLTAGE_SAG_L3_COUNT].value, int) - assert result[obis.VOLTAGE_SAG_L3_COUNT].value == 0 + assert isinstance(telegram.VOLTAGE_SAG_L3_COUNT, CosemObject) + assert telegram.VOLTAGE_SAG_L3_COUNT.unit is None + assert isinstance(telegram.VOLTAGE_SAG_L3_COUNT.value, int) + assert telegram.VOLTAGE_SAG_L3_COUNT.value == 0 # VOLTAGE_SWELL_L1_COUNT (1-0:32.36.0) - assert isinstance(result[obis.VOLTAGE_SWELL_L1_COUNT], CosemObject) - assert result[obis.VOLTAGE_SWELL_L1_COUNT].unit is None - assert isinstance(result[obis.VOLTAGE_SWELL_L1_COUNT].value, int) - assert result[obis.VOLTAGE_SWELL_L1_COUNT].value == 0 + assert isinstance(telegram.VOLTAGE_SWELL_L1_COUNT, CosemObject) + assert telegram.VOLTAGE_SWELL_L1_COUNT.unit is None + assert isinstance(telegram.VOLTAGE_SWELL_L1_COUNT.value, int) + assert telegram.VOLTAGE_SWELL_L1_COUNT.value == 0 # VOLTAGE_SWELL_L2_COUNT (1-0:52.36.0) - assert isinstance(result[obis.VOLTAGE_SWELL_L2_COUNT], CosemObject) - assert result[obis.VOLTAGE_SWELL_L2_COUNT].unit is None - assert isinstance(result[obis.VOLTAGE_SWELL_L2_COUNT].value, int) - assert result[obis.VOLTAGE_SWELL_L2_COUNT].value == 0 + assert isinstance(telegram.VOLTAGE_SWELL_L2_COUNT, CosemObject) + assert telegram.VOLTAGE_SWELL_L2_COUNT.unit is None + assert isinstance(telegram.VOLTAGE_SWELL_L2_COUNT.value, int) + assert telegram.VOLTAGE_SWELL_L2_COUNT.value == 0 # VOLTAGE_SWELL_L3_COUNT (1-0:72.36.0) - assert isinstance(result[obis.VOLTAGE_SWELL_L3_COUNT], CosemObject) - assert result[obis.VOLTAGE_SWELL_L3_COUNT].unit is None - assert isinstance(result[obis.VOLTAGE_SWELL_L3_COUNT].value, int) - assert result[obis.VOLTAGE_SWELL_L3_COUNT].value == 0 + assert isinstance(telegram.VOLTAGE_SWELL_L3_COUNT, CosemObject) + assert telegram.VOLTAGE_SWELL_L3_COUNT.unit is None + assert isinstance(telegram.VOLTAGE_SWELL_L3_COUNT.value, int) + assert telegram.VOLTAGE_SWELL_L3_COUNT.value == 0 # INSTANTANEOUS_VOLTAGE_L1 (1-0:32.7.0) - assert isinstance(result[obis.INSTANTANEOUS_VOLTAGE_L1], CosemObject) - assert result[obis.INSTANTANEOUS_VOLTAGE_L1].unit == 'V' - assert isinstance(result[obis.INSTANTANEOUS_VOLTAGE_L1].value, Decimal) - assert result[obis.INSTANTANEOUS_VOLTAGE_L1].value == Decimal('230.0') + assert isinstance(telegram.INSTANTANEOUS_VOLTAGE_L1, CosemObject) + assert telegram.INSTANTANEOUS_VOLTAGE_L1.unit == 'V' + assert isinstance(telegram.INSTANTANEOUS_VOLTAGE_L1.value, Decimal) + assert telegram.INSTANTANEOUS_VOLTAGE_L1.value == Decimal('230.0') # INSTANTANEOUS_VOLTAGE_L2 (1-0:52.7.0) - assert isinstance(result[obis.INSTANTANEOUS_VOLTAGE_L2], CosemObject) - assert result[obis.INSTANTANEOUS_VOLTAGE_L2].unit == 'V' - assert isinstance(result[obis.INSTANTANEOUS_VOLTAGE_L2].value, Decimal) - assert result[obis.INSTANTANEOUS_VOLTAGE_L2].value == Decimal('230.0') + assert isinstance(telegram.INSTANTANEOUS_VOLTAGE_L2, CosemObject) + assert telegram.INSTANTANEOUS_VOLTAGE_L2.unit == 'V' + assert isinstance(telegram.INSTANTANEOUS_VOLTAGE_L2.value, Decimal) + assert telegram.INSTANTANEOUS_VOLTAGE_L2.value == Decimal('230.0') # INSTANTANEOUS_VOLTAGE_L3 (1-0:72.7.0) - assert isinstance(result[obis.INSTANTANEOUS_VOLTAGE_L3], CosemObject) - assert result[obis.INSTANTANEOUS_VOLTAGE_L3].unit == 'V' - assert isinstance(result[obis.INSTANTANEOUS_VOLTAGE_L3].value, Decimal) - assert result[obis.INSTANTANEOUS_VOLTAGE_L3].value == Decimal('229.0') + assert isinstance(telegram.INSTANTANEOUS_VOLTAGE_L3, CosemObject) + assert telegram.INSTANTANEOUS_VOLTAGE_L3.unit == 'V' + assert isinstance(telegram.INSTANTANEOUS_VOLTAGE_L3.value, Decimal) + assert telegram.INSTANTANEOUS_VOLTAGE_L3.value == Decimal('229.0') # 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.48') + assert isinstance(telegram.INSTANTANEOUS_CURRENT_L1, CosemObject) + assert telegram.INSTANTANEOUS_CURRENT_L1.unit == 'A' + assert isinstance(telegram.INSTANTANEOUS_CURRENT_L1.value, Decimal) + assert telegram.INSTANTANEOUS_CURRENT_L1.value == Decimal('0.48') # 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('0.44') + assert isinstance(telegram.INSTANTANEOUS_CURRENT_L2, CosemObject) + assert telegram.INSTANTANEOUS_CURRENT_L2.unit == 'A' + assert isinstance(telegram.INSTANTANEOUS_CURRENT_L2.value, Decimal) + assert telegram.INSTANTANEOUS_CURRENT_L2.value == Decimal('0.44') # 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('0.86') + assert isinstance(telegram.INSTANTANEOUS_CURRENT_L3, CosemObject) + assert telegram.INSTANTANEOUS_CURRENT_L3.unit == 'A' + assert isinstance(telegram.INSTANTANEOUS_CURRENT_L3.value, Decimal) + assert telegram.INSTANTANEOUS_CURRENT_L3.value == Decimal('0.86') # TEXT_MESSAGE (0-0:96.13.0) - assert isinstance(result[obis.TEXT_MESSAGE], CosemObject) - assert result[obis.TEXT_MESSAGE].unit is None - assert result[obis.TEXT_MESSAGE].value is None + assert isinstance(telegram.TEXT_MESSAGE, CosemObject) + assert telegram.TEXT_MESSAGE.unit is None + assert telegram.TEXT_MESSAGE.value is None # DEVICE_TYPE (0-x:24.1.0) - 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 + assert isinstance(telegram.DEVICE_TYPE, CosemObject) + assert telegram.DEVICE_TYPE.unit is None + assert isinstance(telegram.DEVICE_TYPE.value, int) + assert telegram.DEVICE_TYPE.value == 3 # INSTANTANEOUS_ACTIVE_POWER_L1_POSITIVE (1-0:21.7.0) - assert isinstance(result[obis.INSTANTANEOUS_ACTIVE_POWER_L1_POSITIVE], CosemObject) - assert result[obis.INSTANTANEOUS_ACTIVE_POWER_L1_POSITIVE].unit == 'kW' - assert isinstance(result[obis.INSTANTANEOUS_ACTIVE_POWER_L1_POSITIVE].value, Decimal) - assert result[obis.INSTANTANEOUS_ACTIVE_POWER_L1_POSITIVE].value == Decimal('0.070') + assert isinstance(telegram.INSTANTANEOUS_ACTIVE_POWER_L1_POSITIVE, CosemObject) + assert telegram.INSTANTANEOUS_ACTIVE_POWER_L1_POSITIVE.unit == 'kW' + assert isinstance(telegram.INSTANTANEOUS_ACTIVE_POWER_L1_POSITIVE.value, Decimal) + assert telegram.INSTANTANEOUS_ACTIVE_POWER_L1_POSITIVE.value == Decimal('0.070') # INSTANTANEOUS_ACTIVE_POWER_L2_POSITIVE (1-0:41.7.0) - assert isinstance(result[obis.INSTANTANEOUS_ACTIVE_POWER_L2_POSITIVE], CosemObject) - assert result[obis.INSTANTANEOUS_ACTIVE_POWER_L2_POSITIVE].unit == 'kW' - assert isinstance(result[obis.INSTANTANEOUS_ACTIVE_POWER_L2_POSITIVE].value, Decimal) - assert result[obis.INSTANTANEOUS_ACTIVE_POWER_L2_POSITIVE].value == Decimal('0.032') + assert isinstance(telegram.INSTANTANEOUS_ACTIVE_POWER_L2_POSITIVE, CosemObject) + assert telegram.INSTANTANEOUS_ACTIVE_POWER_L2_POSITIVE.unit == 'kW' + assert isinstance(telegram.INSTANTANEOUS_ACTIVE_POWER_L2_POSITIVE.value, Decimal) + assert telegram.INSTANTANEOUS_ACTIVE_POWER_L2_POSITIVE.value == Decimal('0.032') # INSTANTANEOUS_ACTIVE_POWER_L3_POSITIVE (1-0:61.7.0) - assert isinstance(result[obis.INSTANTANEOUS_ACTIVE_POWER_L3_POSITIVE], CosemObject) - assert result[obis.INSTANTANEOUS_ACTIVE_POWER_L3_POSITIVE].unit == 'kW' - assert isinstance(result[obis.INSTANTANEOUS_ACTIVE_POWER_L3_POSITIVE].value, Decimal) - assert result[obis.INSTANTANEOUS_ACTIVE_POWER_L3_POSITIVE].value == Decimal('0.142') + assert isinstance(telegram.INSTANTANEOUS_ACTIVE_POWER_L3_POSITIVE, CosemObject) + assert telegram.INSTANTANEOUS_ACTIVE_POWER_L3_POSITIVE.unit == 'kW' + assert isinstance(telegram.INSTANTANEOUS_ACTIVE_POWER_L3_POSITIVE.value, Decimal) + assert telegram.INSTANTANEOUS_ACTIVE_POWER_L3_POSITIVE.value == Decimal('0.142') # INSTANTANEOUS_ACTIVE_POWER_L1_NEGATIVE (1-0:22.7.0) - assert isinstance(result[obis.INSTANTANEOUS_ACTIVE_POWER_L1_NEGATIVE], CosemObject) - assert result[obis.INSTANTANEOUS_ACTIVE_POWER_L1_NEGATIVE].unit == 'kW' - assert isinstance(result[obis.INSTANTANEOUS_ACTIVE_POWER_L1_NEGATIVE].value, Decimal) - assert result[obis.INSTANTANEOUS_ACTIVE_POWER_L1_NEGATIVE].value == Decimal('0') + assert isinstance(telegram.INSTANTANEOUS_ACTIVE_POWER_L1_NEGATIVE, CosemObject) + assert telegram.INSTANTANEOUS_ACTIVE_POWER_L1_NEGATIVE.unit == 'kW' + assert isinstance(telegram.INSTANTANEOUS_ACTIVE_POWER_L1_NEGATIVE.value, Decimal) + assert telegram.INSTANTANEOUS_ACTIVE_POWER_L1_NEGATIVE.value == Decimal('0') # INSTANTANEOUS_ACTIVE_POWER_L2_NEGATIVE (1-0:42.7.0) - assert isinstance(result[obis.INSTANTANEOUS_ACTIVE_POWER_L2_NEGATIVE], CosemObject) - assert result[obis.INSTANTANEOUS_ACTIVE_POWER_L2_NEGATIVE].unit == 'kW' - assert isinstance(result[obis.INSTANTANEOUS_ACTIVE_POWER_L2_NEGATIVE].value, Decimal) - assert result[obis.INSTANTANEOUS_ACTIVE_POWER_L2_NEGATIVE].value == Decimal('0') + assert isinstance(telegram.INSTANTANEOUS_ACTIVE_POWER_L2_NEGATIVE, CosemObject) + assert telegram.INSTANTANEOUS_ACTIVE_POWER_L2_NEGATIVE.unit == 'kW' + assert isinstance(telegram.INSTANTANEOUS_ACTIVE_POWER_L2_NEGATIVE.value, Decimal) + assert telegram.INSTANTANEOUS_ACTIVE_POWER_L2_NEGATIVE.value == Decimal('0') # INSTANTANEOUS_ACTIVE_POWER_L3_NEGATIVE (1-0:62.7.0) - assert isinstance(result[obis.INSTANTANEOUS_ACTIVE_POWER_L3_NEGATIVE], CosemObject) - assert result[obis.INSTANTANEOUS_ACTIVE_POWER_L3_NEGATIVE].unit == 'kW' - assert isinstance(result[obis.INSTANTANEOUS_ACTIVE_POWER_L3_NEGATIVE].value, Decimal) - assert result[obis.INSTANTANEOUS_ACTIVE_POWER_L3_NEGATIVE].value == Decimal('0') + assert isinstance(telegram.INSTANTANEOUS_ACTIVE_POWER_L3_NEGATIVE, CosemObject) + assert telegram.INSTANTANEOUS_ACTIVE_POWER_L3_NEGATIVE.unit == 'kW' + assert isinstance(telegram.INSTANTANEOUS_ACTIVE_POWER_L3_NEGATIVE.value, Decimal) + assert telegram.INSTANTANEOUS_ACTIVE_POWER_L3_NEGATIVE.value == Decimal('0') + + # There's only one Mbus device (gas meter) in this case. Alternatively + # use get_mbget_mbus_device_by_channel + gas_meter_devices = telegram.get_mbus_devices() + gas_meter_device = gas_meter_devices[0] # EQUIPMENT_IDENTIFIER_GAS (0-x:96.1.0) - assert isinstance(result[obis.EQUIPMENT_IDENTIFIER_GAS], CosemObject) - assert result[obis.EQUIPMENT_IDENTIFIER_GAS].unit is None - assert isinstance(result[obis.EQUIPMENT_IDENTIFIER_GAS].value, str) - assert result[obis.EQUIPMENT_IDENTIFIER_GAS].value == '3232323241424344313233343536373839' + assert isinstance(gas_meter_device.DEVICE_TYPE, CosemObject) + assert gas_meter_device.DEVICE_TYPE.unit is None + assert isinstance(gas_meter_device.DEVICE_TYPE.value, int) + assert gas_meter_device.DEVICE_TYPE.value == 3 + + # EQUIPMENT_IDENTIFIER_GAS (0-x:96.1.0) + assert isinstance(gas_meter_device.EQUIPMENT_IDENTIFIER_GAS, CosemObject) + assert gas_meter_device.EQUIPMENT_IDENTIFIER_GAS.unit is None + assert isinstance(gas_meter_device.EQUIPMENT_IDENTIFIER_GAS.value, str) + assert gas_meter_device.EQUIPMENT_IDENTIFIER_GAS.value == '3232323241424344313233343536373839' # HOURLY_GAS_METER_READING (0-1:24.2.1) - assert isinstance(result[obis.HOURLY_GAS_METER_READING], MBusObject) - assert result[obis.HOURLY_GAS_METER_READING].unit == 'm3' - assert isinstance(result[obis.HOURLY_GAS_METER_READING].value, Decimal) - assert result[obis.HOURLY_GAS_METER_READING].value == Decimal('0.107') + assert isinstance(gas_meter_device.HOURLY_GAS_METER_READING, MBusObject) + assert gas_meter_device.HOURLY_GAS_METER_READING.unit == 'm3' + assert isinstance(telegram.HOURLY_GAS_METER_READING.value, Decimal) + assert gas_meter_device.HOURLY_GAS_METER_READING.value == Decimal('0.107') def test_checksum_valid(self): # No exception is raised. diff --git a/test/test_telegram.py b/test/test_telegram.py index 2f00396..bb60b8a 100644 --- a/test/test_telegram.py +++ b/test/test_telegram.py @@ -334,10 +334,12 @@ class TelegramTest(unittest.TestCase): self.assertEqual(len(mbus_devices), 2) mbus_device_1 = mbus_devices[0] + self.assertEqual(mbus_device_1.DEVICE_TYPE.value, 3) 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 = mbus_devices[1] + 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')) @@ -346,9 +348,11 @@ class TelegramTest(unittest.TestCase): telegram = parser.parse(TELEGRAM_V5_TWO_MBUS) mbus_device_1 = telegram.get_mbus_device_by_channel(1) + self.assertEqual(mbus_device_1.DEVICE_TYPE.value, 3) 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.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'))