issue-51-telegram refactored TelegramParser.parse to return Telegram … (#121)
* issue-51-telegram improved mbus device parsing; refactored TelegramParser.parse to return Telegram object which is backwards compatible with the dict result
This commit is contained in:
@@ -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'
|
||||
|
||||
@@ -1,8 +1,7 @@
|
||||
from dsmr_parser import telegram_specifications
|
||||
from dsmr_parser.objects import Telegram
|
||||
from dsmr_parser.parsers import TelegramParser
|
||||
from test.example_telegrams import TELEGRAM_V4_2
|
||||
parser = TelegramParser(telegram_specifications.V4)
|
||||
telegram = Telegram(TELEGRAM_V4_2, parser, telegram_specifications.V4)
|
||||
telegram = parser.parse(TELEGRAM_V4_2)
|
||||
|
||||
print(telegram)
|
||||
|
||||
@@ -0,0 +1,61 @@
|
||||
from decimal import Decimal
|
||||
|
||||
import json
|
||||
import unittest
|
||||
|
||||
from dsmr_parser import telegram_specifications, obis_references
|
||||
from dsmr_parser.objects import MbusDevice
|
||||
|
||||
|
||||
class MbusDeviceTest(unittest.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
v5_objects = telegram_specifications.V5['objects']
|
||||
|
||||
device_type_parser = v5_objects[obis_references.DEVICE_TYPE]
|
||||
device_type = device_type_parser.parse('0-2:24.1.0(003)\r\n')
|
||||
|
||||
equipment_parser = v5_objects[obis_references.EQUIPMENT_IDENTIFIER_GAS]
|
||||
equipment = equipment_parser.parse('0-2:96.1.0(4730303339303031393336393930363139)\r\n')
|
||||
|
||||
gas_reading_parser = v5_objects[obis_references.HOURLY_GAS_METER_READING]
|
||||
gas_reading = gas_reading_parser.parse('0-2:24.2.1(200426223001S)(00246.138*m3)\r\n')
|
||||
|
||||
mbus_device = MbusDevice(channel_id=1)
|
||||
mbus_device.add(obis_references.DEVICE_TYPE, device_type)
|
||||
mbus_device.add(obis_references.EQUIPMENT_IDENTIFIER_GAS, equipment)
|
||||
mbus_device.add(obis_references.HOURLY_GAS_METER_READING, gas_reading)
|
||||
|
||||
self.mbus_device = mbus_device
|
||||
|
||||
def test_attributes(self):
|
||||
self.assertEqual(self.mbus_device.DEVICE_TYPE.value, 3)
|
||||
self.assertEqual(self.mbus_device.DEVICE_TYPE.unit, None)
|
||||
|
||||
self.assertEqual(self.mbus_device.EQUIPMENT_IDENTIFIER_GAS.value,
|
||||
'4730303339303031393336393930363139')
|
||||
self.assertEqual(self.mbus_device.EQUIPMENT_IDENTIFIER_GAS.unit, None)
|
||||
|
||||
self.assertEqual(self.mbus_device.HOURLY_GAS_METER_READING.value, Decimal('246.138'))
|
||||
self.assertEqual(self.mbus_device.HOURLY_GAS_METER_READING.unit, 'm3')
|
||||
|
||||
def test_to_json(self):
|
||||
self.assertEqual(
|
||||
json.loads(self.mbus_device.to_json()),
|
||||
{
|
||||
'CHANNEL_ID': 1,
|
||||
'DEVICE_TYPE': {'value': 3, 'unit': None},
|
||||
'EQUIPMENT_IDENTIFIER_GAS': {'value': '4730303339303031393336393930363139', 'unit': None},
|
||||
'HOURLY_GAS_METER_READING': {'datetime': '2020-04-26T20:30:01+00:00', 'value': 246.138, 'unit': 'm3'}}
|
||||
)
|
||||
|
||||
def test_str(self):
|
||||
self.assertEqual(
|
||||
str(self.mbus_device),
|
||||
(
|
||||
'MBUS DEVICE (channel 1)\n'
|
||||
'\tDEVICE_TYPE: 3 [None]\n'
|
||||
'\tEQUIPMENT_IDENTIFIER_GAS: 4730303339303031393336393930363139 [None]\n'
|
||||
'\tHOURLY_GAS_METER_READING: 246.138 [m3] at 2020-04-26T20:30:01+00:00\n'
|
||||
)
|
||||
)
|
||||
@@ -2,7 +2,6 @@ import unittest
|
||||
|
||||
from dsmr_parser import telegram_specifications
|
||||
|
||||
from dsmr_parser.objects import Telegram
|
||||
from dsmr_parser.objects import ProfileGenericObject
|
||||
from dsmr_parser.parsers import TelegramParser
|
||||
from dsmr_parser.parsers import ProfileGenericParser
|
||||
@@ -18,7 +17,7 @@ class TestParserCornerCases(unittest.TestCase):
|
||||
def test_power_event_log_empty_1(self):
|
||||
# POWER_EVENT_FAILURE_LOG (1-0:99.97.0)
|
||||
parser = TelegramParser(telegram_specifications.V5)
|
||||
telegram = Telegram(TELEGRAM_V5, parser, telegram_specifications.V5)
|
||||
telegram = parser.parse(TELEGRAM_V5)
|
||||
|
||||
object_type = ProfileGenericObject
|
||||
testitem = telegram.POWER_EVENT_FAILURE_LOG
|
||||
@@ -1,15 +1,15 @@
|
||||
import json
|
||||
import unittest
|
||||
import datetime
|
||||
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.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 test.example_telegrams import TELEGRAM_V4_2, TELEGRAM_V5_TWO_MBUS, TELEGRAM_V5
|
||||
from decimal import Decimal
|
||||
|
||||
|
||||
@@ -30,7 +30,7 @@ class TelegramTest(unittest.TestCase):
|
||||
|
||||
def test_instantiate(self):
|
||||
parser = TelegramParser(telegram_specifications.V4)
|
||||
telegram = Telegram(TELEGRAM_V4_2, parser, telegram_specifications.V4)
|
||||
telegram = parser.parse(TELEGRAM_V4_2)
|
||||
|
||||
# P1_MESSAGE_HEADER (1-3:0.2.8)
|
||||
self.verify_telegram_item(telegram,
|
||||
@@ -320,3 +320,170 @@ class TelegramTest(unittest.TestCase):
|
||||
item_names_tested_set = set(self.item_names_tested)
|
||||
|
||||
assert item_names_tested_set == V4_name_set
|
||||
|
||||
def test_iter(self):
|
||||
parser = TelegramParser(telegram_specifications.V5)
|
||||
telegram = parser.parse(TELEGRAM_V5)
|
||||
|
||||
for obis_name, dsmr_object in telegram:
|
||||
break
|
||||
|
||||
# Verify that the iterator works for at least one value
|
||||
self.assertEqual(obis_name, obis_name_mapping.EN[obis_references.P1_MESSAGE_HEADER])
|
||||
self.assertEqual(dsmr_object.value, '50')
|
||||
|
||||
def test_mbus_devices(self):
|
||||
parser = TelegramParser(telegram_specifications.V5)
|
||||
telegram = parser.parse(TELEGRAM_V5_TWO_MBUS)
|
||||
mbus_devices = telegram.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.assertFalse(hasattr(telegram, 'MBUS_DEVICES'))
|
||||
self.assertIsNone(telegram.get_mbus_device_by_channel(1))
|
||||
|
||||
def test_to_json(self):
|
||||
parser = TelegramParser(telegram_specifications.V5)
|
||||
telegram = parser.parse(TELEGRAM_V5)
|
||||
json_data = json.loads(telegram.to_json())
|
||||
|
||||
self.assertEqual(
|
||||
json_data,
|
||||
{'CURRENT_ELECTRICITY_DELIVERY': {'unit': 'kW', 'value': 0.0},
|
||||
'CURRENT_ELECTRICITY_USAGE': {'unit': 'kW', 'value': 0.244},
|
||||
'DEVICE_TYPE': {'unit': None, 'value': 3},
|
||||
'ELECTRICITY_ACTIVE_TARIFF': {'unit': None, 'value': '0002'},
|
||||
'ELECTRICITY_DELIVERED_TARIFF_1': {'unit': 'kWh', 'value': 2.444},
|
||||
'ELECTRICITY_DELIVERED_TARIFF_2': {'unit': 'kWh', 'value': 0.0},
|
||||
'ELECTRICITY_USED_TARIFF_1': {'unit': 'kWh', 'value': 4.426},
|
||||
'ELECTRICITY_USED_TARIFF_2': {'unit': 'kWh', 'value': 2.399},
|
||||
'EQUIPMENT_IDENTIFIER': {'unit': None,
|
||||
'value': '4B384547303034303436333935353037'},
|
||||
'EQUIPMENT_IDENTIFIER_GAS': {'unit': None, 'value': None},
|
||||
'HOURLY_GAS_METER_READING': {'datetime': '2017-01-02T15:10:05+00:00',
|
||||
'unit': 'm3',
|
||||
'value': 0.107},
|
||||
'INSTANTANEOUS_ACTIVE_POWER_L1_NEGATIVE': {'unit': 'kW', 'value': 0.0},
|
||||
'INSTANTANEOUS_ACTIVE_POWER_L1_POSITIVE': {'unit': 'kW', 'value': 0.07},
|
||||
'INSTANTANEOUS_ACTIVE_POWER_L2_NEGATIVE': {'unit': 'kW', 'value': 0.0},
|
||||
'INSTANTANEOUS_ACTIVE_POWER_L2_POSITIVE': {'unit': 'kW', 'value': 0.032},
|
||||
'INSTANTANEOUS_ACTIVE_POWER_L3_NEGATIVE': {'unit': 'kW', 'value': 0.0},
|
||||
'INSTANTANEOUS_ACTIVE_POWER_L3_POSITIVE': {'unit': 'kW', 'value': 0.142},
|
||||
'INSTANTANEOUS_CURRENT_L1': {'unit': 'A', 'value': 0.48},
|
||||
'INSTANTANEOUS_CURRENT_L2': {'unit': 'A', 'value': 0.44},
|
||||
'INSTANTANEOUS_CURRENT_L3': {'unit': 'A', 'value': 0.86},
|
||||
'INSTANTANEOUS_VOLTAGE_L1': {'unit': 'V', 'value': 230.0},
|
||||
'INSTANTANEOUS_VOLTAGE_L2': {'unit': 'V', 'value': 230.0},
|
||||
'INSTANTANEOUS_VOLTAGE_L3': {'unit': 'V', 'value': 229.0},
|
||||
'LONG_POWER_FAILURE_COUNT': {'unit': None, 'value': 0},
|
||||
'MBUS_DEVICES': [{'CHANNEL_ID': 1,
|
||||
'DEVICE_TYPE': {'unit': None, 'value': 3},
|
||||
'EQUIPMENT_IDENTIFIER_GAS': {'unit': None,
|
||||
'value': '3232323241424344313233343536373839'},
|
||||
'HOURLY_GAS_METER_READING': {'datetime': '2017-01-02T15:10:05+00:00',
|
||||
'unit': 'm3',
|
||||
'value': 0.107}},
|
||||
{'CHANNEL_ID': 2,
|
||||
'DEVICE_TYPE': {'unit': None, 'value': 3},
|
||||
'EQUIPMENT_IDENTIFIER_GAS': {'unit': None, 'value': None}}],
|
||||
'P1_MESSAGE_HEADER': {'unit': None, 'value': '50'},
|
||||
'P1_MESSAGE_TIMESTAMP': {'unit': None, 'value': '2017-01-02T18:20:02+00:00'},
|
||||
'POWER_EVENT_FAILURE_LOG': {'buffer': [],
|
||||
'buffer_length': 0,
|
||||
'buffer_type': '0-0:96.7.19'},
|
||||
'SHORT_POWER_FAILURE_COUNT': {'unit': None, 'value': 13},
|
||||
'TEXT_MESSAGE': {'unit': None, 'value': None},
|
||||
'VOLTAGE_SAG_L1_COUNT': {'unit': None, 'value': 0},
|
||||
'VOLTAGE_SAG_L2_COUNT': {'unit': None, 'value': 0},
|
||||
'VOLTAGE_SAG_L3_COUNT': {'unit': None, 'value': 0},
|
||||
'VOLTAGE_SWELL_L1_COUNT': {'unit': None, 'value': 0},
|
||||
'VOLTAGE_SWELL_L2_COUNT': {'unit': None, 'value': 0},
|
||||
'VOLTAGE_SWELL_L3_COUNT': {'unit': None, 'value': 0}}
|
||||
)
|
||||
|
||||
def test_to_str(self):
|
||||
parser = TelegramParser(telegram_specifications.V5)
|
||||
telegram = parser.parse(TELEGRAM_V5)
|
||||
|
||||
self.assertEqual(
|
||||
str(telegram),
|
||||
(
|
||||
'P1_MESSAGE_HEADER: 50 [None]\n'
|
||||
'P1_MESSAGE_TIMESTAMP: 2017-01-02T18:20:02+00:00 [None]\n'
|
||||
'EQUIPMENT_IDENTIFIER: 4B384547303034303436333935353037 [None]\n'
|
||||
'ELECTRICITY_USED_TARIFF_1: 4.426 [kWh]\n'
|
||||
'ELECTRICITY_USED_TARIFF_2: 2.399 [kWh]\n'
|
||||
'ELECTRICITY_DELIVERED_TARIFF_1: 2.444 [kWh]\n'
|
||||
'ELECTRICITY_DELIVERED_TARIFF_2: 0.000 [kWh]\n'
|
||||
'ELECTRICITY_ACTIVE_TARIFF: 0002 [None]\n'
|
||||
'CURRENT_ELECTRICITY_USAGE: 0.244 [kW]\n'
|
||||
'CURRENT_ELECTRICITY_DELIVERY: 0.000 [kW]\n'
|
||||
'LONG_POWER_FAILURE_COUNT: 0 [None]\n'
|
||||
'SHORT_POWER_FAILURE_COUNT: 13 [None]\n'
|
||||
'POWER_EVENT_FAILURE_LOG: buffer length: 0\n'
|
||||
' buffer type: 0-0:96.7.19\n'
|
||||
'VOLTAGE_SAG_L1_COUNT: 0 [None]\n'
|
||||
'VOLTAGE_SAG_L2_COUNT: 0 [None]\n'
|
||||
'VOLTAGE_SAG_L3_COUNT: 0 [None]\n'
|
||||
'VOLTAGE_SWELL_L1_COUNT: 0 [None]\n'
|
||||
'VOLTAGE_SWELL_L2_COUNT: 0 [None]\n'
|
||||
'VOLTAGE_SWELL_L3_COUNT: 0 [None]\n'
|
||||
'INSTANTANEOUS_VOLTAGE_L1: 230.0 [V]\n'
|
||||
'INSTANTANEOUS_VOLTAGE_L2: 230.0 [V]\n'
|
||||
'INSTANTANEOUS_VOLTAGE_L3: 229.0 [V]\n'
|
||||
'INSTANTANEOUS_CURRENT_L1: 0.48 [A]\n'
|
||||
'INSTANTANEOUS_CURRENT_L2: 0.44 [A]\n'
|
||||
'INSTANTANEOUS_CURRENT_L3: 0.86 [A]\n'
|
||||
'TEXT_MESSAGE: None [None]\n'
|
||||
'DEVICE_TYPE: 3 [None]\n'
|
||||
'MBUS DEVICE (channel 1)\n'
|
||||
' DEVICE_TYPE: 3 [None]\n'
|
||||
' EQUIPMENT_IDENTIFIER_GAS: 3232323241424344313233343536373839 [None]\n'
|
||||
' HOURLY_GAS_METER_READING: 0.107 [m3] at 2017-01-02T15:10:05+00:00\n'
|
||||
'MBUS DEVICE (channel 2)\n'
|
||||
' DEVICE_TYPE: 3 [None]\n'
|
||||
' EQUIPMENT_IDENTIFIER_GAS: None [None]\n'
|
||||
'INSTANTANEOUS_ACTIVE_POWER_L1_POSITIVE: 0.070 [kW]\n'
|
||||
'INSTANTANEOUS_ACTIVE_POWER_L2_POSITIVE: 0.032 [kW]\n'
|
||||
'INSTANTANEOUS_ACTIVE_POWER_L3_POSITIVE: 0.142 [kW]\n'
|
||||
'INSTANTANEOUS_ACTIVE_POWER_L1_NEGATIVE: 0.000 [kW]\n'
|
||||
'INSTANTANEOUS_ACTIVE_POWER_L2_NEGATIVE: 0.000 [kW]\n'
|
||||
'INSTANTANEOUS_ACTIVE_POWER_L3_NEGATIVE: 0.000 [kW]\n'
|
||||
'EQUIPMENT_IDENTIFIER_GAS: None [None]\n'
|
||||
'HOURLY_GAS_METER_READING: 0.107 [m3] at 2017-01-02T15:10:05+00:00\n'
|
||||
)
|
||||
)
|
||||
|
||||
def test_getitem(self):
|
||||
parser = TelegramParser(telegram_specifications.V5)
|
||||
telegram = parser.parse(TELEGRAM_V5)
|
||||
|
||||
self.assertEqual(telegram[obis_references.P1_MESSAGE_HEADER].value, '50')
|
||||
+152
-142
@@ -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.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.
|
||||
@@ -255,10 +265,10 @@ class TelegramParserV5Test(unittest.TestCase):
|
||||
)
|
||||
invalid_date_telegram = invalid_date_telegram.replace('!6EEE\r\n', '!90C2\r\n')
|
||||
parser = TelegramParser(telegram_specifications.V5)
|
||||
result = parser.parse(invalid_date_telegram)
|
||||
telegram = parser.parse(invalid_date_telegram)
|
||||
|
||||
# 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 is None
|
||||
assert isinstance(result[obis.HOURLY_GAS_METER_READING].value, Decimal)
|
||||
assert result[obis.HOURLY_GAS_METER_READING].value == Decimal('0.000')
|
||||
assert isinstance(telegram.HOURLY_GAS_METER_READING, MBusObject)
|
||||
assert telegram.HOURLY_GAS_METER_READING.unit is None
|
||||
assert isinstance(telegram.HOURLY_GAS_METER_READING.value, Decimal)
|
||||
assert telegram.HOURLY_GAS_METER_READING.value == Decimal('0.000')
|
||||
|
||||
@@ -4,7 +4,7 @@ import unittest
|
||||
|
||||
from dsmr_parser import obis_references as obis
|
||||
from dsmr_parser.clients.protocol import create_dsmr_protocol
|
||||
|
||||
from dsmr_parser.objects import Telegram
|
||||
|
||||
TELEGRAM_V2_2 = (
|
||||
'/ISk5\2MT382-1004\r\n'
|
||||
@@ -44,7 +44,7 @@ class ProtocolTest(unittest.TestCase):
|
||||
self.protocol.data_received(TELEGRAM_V2_2.encode('ascii'))
|
||||
|
||||
telegram = self.protocol.telegram_callback.call_args_list[0][0][0]
|
||||
assert isinstance(telegram, dict)
|
||||
assert isinstance(telegram, Telegram)
|
||||
|
||||
assert float(telegram[obis.CURRENT_ELECTRICITY_USAGE].value) == 1.01
|
||||
assert telegram[obis.CURRENT_ELECTRICITY_USAGE].unit == 'kW'
|
||||
|
||||
@@ -4,7 +4,7 @@ import unittest
|
||||
|
||||
from dsmr_parser import obis_references as obis
|
||||
from dsmr_parser.clients.rfxtrx_protocol import create_rfxtrx_dsmr_protocol, PACKETTYPE_DSMR, SUBTYPE_P1
|
||||
|
||||
from dsmr_parser.objects import Telegram
|
||||
|
||||
TELEGRAM_V2_2 = (
|
||||
'/ISk5\2MT382-1004\r\n'
|
||||
@@ -68,7 +68,7 @@ class RFXtrxProtocolTest(unittest.TestCase):
|
||||
self.protocol.data_received(data[200:])
|
||||
|
||||
telegram = self.protocol.telegram_callback.call_args_list[0][0][0]
|
||||
assert isinstance(telegram, dict)
|
||||
assert isinstance(telegram, Telegram)
|
||||
|
||||
assert float(telegram[obis.CURRENT_ELECTRICITY_USAGE].value) == 1.01
|
||||
assert telegram[obis.CURRENT_ELECTRICITY_USAGE].unit == 'kW'
|
||||
|
||||
Reference in New Issue
Block a user