dsmr_parser/test/test_telegram.py

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 ProfileGenericObject
from dsmr_parser.parsers import TelegramParser
from test.example_telegrams import TELEGRAM_V4_2, TELEGRAM_V5_TWO_MBUS
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 = parser.parse(TELEGRAM_V4_2)

        # P1_MESSAGE_HEADER (1-3:0.2.8)
        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

    def test_len(self):
        parser = TelegramParser(telegram_specifications.V5)
        telegram = parser.parse(TELEGRAM_V5_TWO_MBUS)

        self.assertEqual(len(telegram), 35)

    def test_get_mbus_devices(self):
        parser = TelegramParser(telegram_specifications.V5)
        telegram = parser.parse(TELEGRAM_V5_TWO_MBUS)
        mbus_devices = telegram.get_mbus_devices()

        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'))

    def test_get_mbus_device_by_channel(self):
        parser = TelegramParser(telegram_specifications.V5)
        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'))

    def test_without_mbus_devices(self):
        parser = TelegramParser(telegram_specifications.V5, apply_checksum_validation=False)
        telegram = parser.parse('')

        self.assertEqual(telegram.get_mbus_devices(), [])
        self.assertIsNone(telegram.get_mbus_device_by_channel(1))

        # Because of a bug related to incorrect use of defaultdict,
        # test again for unwanted side effects
        self.assertEqual(telegram.get_mbus_devices(), [])