Refactor of name mapping, added E.ON Hungary (#137)

Browse Source 
* Added EON HUNGARY specification

* refactoring obis name mapping
This commit is contained in:
balazs92117 2023-08-01 09:23:17 +02:00 committed by GitHub
parent af2e6558de
commit 2f1e080df3
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
11 changed files with 1843 additions and 410 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
dsmr_parser/clients/protocol.py
View File

@ -21,7 +21,9 @@ def create_dsmr_protocol(dsmr_version, telegram_callback, loop=None, **kwargs):
return protocol
def _create_dsmr_protocol(dsmr_version, telegram_callback, protocol, loop=None, **kwargs):
# pylama noqa - because of "complex" (too long) if-elif-else.
# Match - case might be a solution but it is not available in <3.10
def _create_dsmr_protocol(dsmr_version, telegram_callback, protocol, loop=None, **kwargs): #noqa
"""Creates a DSMR asyncio protocol."""
if dsmr_version == '2.2':
@ -51,6 +53,9 @@ def _create_dsmr_protocol(dsmr_version, telegram_callback, protocol, loop=None,
elif dsmr_version == 'ISKRA_IE':
specification = telegram_specifications.ISKRA_IE
serial_settings = SERIAL_SETTINGS_V5
elif dsmr_version == '5EONHU':
specification = telegram_specifications.EON_HUNGARY
serial_settings = SERIAL_SETTINGS_V5
else:
raise NotImplementedError("No telegram parser found for version: %s",
dsmr_version)

5
dsmr_parser/clients/socket_.py
View File

@ -43,7 +43,12 @@ class SocketReader(object):
continue
for data in lines:
try:
self.telegram_buffer.append(data.decode('ascii'))
except UnicodeDecodeError:
# Some garbage came through the channel
# E.g.: Happens at EON_HUNGARY, but only once at the start of the socket.
logger.error('Failed to parse telegram due to unicode decode error')
for telegram in self.telegram_buffer.get_all():
try:

98
dsmr_parser/obis_name_mapping.py
View File

@ -1,98 +0,0 @@
from dsmr_parser import obis_references as obis
"""
dsmr_parser.obis_name_mapping
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
This module contains a mapping of obis references to names.
"""
EN = {
obis.P1_MESSAGE_HEADER: 'P1_MESSAGE_HEADER',
obis.P1_MESSAGE_TIMESTAMP: 'P1_MESSAGE_TIMESTAMP',
obis.ELECTRICITY_IMPORTED_TOTAL: 'ELECTRICITY_IMPORTED_TOTAL',
obis.ELECTRICITY_REACTIVE_IMPORTED_TOTAL: 'ELECTRICITY_REACTIVE_IMPORTED_TOTAL',
obis.ELECTRICITY_USED_TARIFF_1: 'ELECTRICITY_USED_TARIFF_1',
obis.ELECTRICITY_USED_TARIFF_2: 'ELECTRICITY_USED_TARIFF_2',
obis.ELECTRICITY_EXPORTED_TOTAL: 'ELECTRICITY_EXPORTED_TOTAL',
obis.ELECTRICITY_REACTIVE_EXPORTED_TOTAL: 'ELECTRICITY_REACTIVE_EXPORTED_TOTAL',
obis.ELECTRICITY_DELIVERED_TARIFF_1: 'ELECTRICITY_DELIVERED_TARIFF_1',
obis.ELECTRICITY_DELIVERED_TARIFF_2: 'ELECTRICITY_DELIVERED_TARIFF_2',
obis.ELECTRICITY_ACTIVE_TARIFF: 'ELECTRICITY_ACTIVE_TARIFF',
obis.CURRENT_REACTIVE_EXPORTED: 'CURRENT_REACTIVE_EXPORTED',
obis.ELECTRICITY_REACTIVE_IMPORTED_TARIFF_1: 'ELECTRICITY_REACTIVE_IMPORTED_TARIFF_1',
obis.ELECTRICITY_REACTIVE_IMPORTED_TARIFF_2: 'ELECTRICITY_REACTIVE_IMPORTED_TARIFF_2',
obis.ELECTRICITY_REACTIVE_EXPORTED_TARIFF_1: 'ELECTRICITY_REACTIVE_EXPORTED_TARIFF_1',
obis.ELECTRICITY_REACTIVE_EXPORTED_TARIFF_2: 'ELECTRICITY_REACTIVE_EXPORTED_TARIFF_2',
obis.CURRENT_REACTIVE_IMPORTED: 'CURRENT_REACTIVE_IMPORTED',
obis.EQUIPMENT_IDENTIFIER: 'EQUIPMENT_IDENTIFIER',
obis.CURRENT_ELECTRICITY_USAGE: 'CURRENT_ELECTRICITY_USAGE',
obis.CURRENT_ELECTRICITY_DELIVERY: 'CURRENT_ELECTRICITY_DELIVERY',
obis.LONG_POWER_FAILURE_COUNT: 'LONG_POWER_FAILURE_COUNT',
obis.SHORT_POWER_FAILURE_COUNT: 'SHORT_POWER_FAILURE_COUNT',
obis.POWER_EVENT_FAILURE_LOG: 'POWER_EVENT_FAILURE_LOG',
obis.VOLTAGE_SAG_L1_COUNT: 'VOLTAGE_SAG_L1_COUNT',
obis.VOLTAGE_SAG_L2_COUNT: 'VOLTAGE_SAG_L2_COUNT',
obis.VOLTAGE_SAG_L3_COUNT: 'VOLTAGE_SAG_L3_COUNT',
obis.VOLTAGE_SWELL_L1_COUNT: 'VOLTAGE_SWELL_L1_COUNT',
obis.VOLTAGE_SWELL_L2_COUNT: 'VOLTAGE_SWELL_L2_COUNT',
obis.VOLTAGE_SWELL_L3_COUNT: 'VOLTAGE_SWELL_L3_COUNT',
obis.INSTANTANEOUS_VOLTAGE_L1: 'INSTANTANEOUS_VOLTAGE_L1',
obis.INSTANTANEOUS_VOLTAGE_L2: 'INSTANTANEOUS_VOLTAGE_L2',
obis.INSTANTANEOUS_VOLTAGE_L3: 'INSTANTANEOUS_VOLTAGE_L3',
obis.INSTANTANEOUS_CURRENT_L1: 'INSTANTANEOUS_CURRENT_L1',
obis.INSTANTANEOUS_CURRENT_L2: 'INSTANTANEOUS_CURRENT_L2',
obis.INSTANTANEOUS_CURRENT_L3: 'INSTANTANEOUS_CURRENT_L3',
obis.TEXT_MESSAGE_CODE: 'TEXT_MESSAGE_CODE',
obis.TEXT_MESSAGE: 'TEXT_MESSAGE',
obis.DEVICE_TYPE: 'DEVICE_TYPE',
obis.INSTANTANEOUS_ACTIVE_POWER_L1_POSITIVE: 'INSTANTANEOUS_ACTIVE_POWER_L1_POSITIVE',
obis.INSTANTANEOUS_ACTIVE_POWER_L2_POSITIVE: 'INSTANTANEOUS_ACTIVE_POWER_L2_POSITIVE',
obis.INSTANTANEOUS_ACTIVE_POWER_L3_POSITIVE: 'INSTANTANEOUS_ACTIVE_POWER_L3_POSITIVE',
obis.INSTANTANEOUS_ACTIVE_POWER_L1_NEGATIVE: 'INSTANTANEOUS_ACTIVE_POWER_L1_NEGATIVE',
obis.INSTANTANEOUS_ACTIVE_POWER_L2_NEGATIVE: 'INSTANTANEOUS_ACTIVE_POWER_L2_NEGATIVE',
obis.INSTANTANEOUS_ACTIVE_POWER_L3_NEGATIVE: 'INSTANTANEOUS_ACTIVE_POWER_L3_NEGATIVE',
obis.INSTANTANEOUS_REACTIVE_POWER_L1_POSITIVE: 'INSTANTANEOUS_REACTIVE_POWER_L1_POSITIVE',
obis.INSTANTANEOUS_REACTIVE_POWER_L1_NEGATIVE: 'INSTANTANEOUS_REACTIVE_POWER_L1_NEGATIVE',
obis.INSTANTANEOUS_REACTIVE_POWER_L2_POSITIVE: 'INSTANTANEOUS_REACTIVE_POWER_L2_POSITIVE',
obis.INSTANTANEOUS_REACTIVE_POWER_L2_NEGATIVE: 'INSTANTANEOUS_REACTIVE_POWER_L2_NEGATIVE',
obis.INSTANTANEOUS_REACTIVE_POWER_L3_POSITIVE: 'INSTANTANEOUS_REACTIVE_POWER_L3_POSITIVE',
obis.INSTANTANEOUS_REACTIVE_POWER_L3_NEGATIVE: 'INSTANTANEOUS_REACTIVE_POWER_L3_NEGATIVE',
obis.EQUIPMENT_IDENTIFIER_GAS: 'EQUIPMENT_IDENTIFIER_GAS',
obis.HOURLY_GAS_METER_READING: 'HOURLY_GAS_METER_READING',
obis.GAS_METER_READING: 'GAS_METER_READING',
obis.ACTUAL_TRESHOLD_ELECTRICITY: 'ACTUAL_TRESHOLD_ELECTRICITY',
obis.ACTUAL_SWITCH_POSITION: 'ACTUAL_SWITCH_POSITION',
obis.VALVE_POSITION_GAS: 'VALVE_POSITION_GAS',
obis.BELGIUM_VERSION_INFORMATION: 'BELGIUM_VERSION_INFORMATION',
obis.BELGIUM_EQUIPMENT_IDENTIFIER: 'BELGIUM_EQUIPMENT_IDENTIFIER',
obis.BELGIUM_CURRENT_AVERAGE_DEMAND: 'BELGIUM_CURRENT_AVERAGE_DEMAND',
obis.BELGIUM_MAXIMUM_DEMAND_MONTH: 'BELGIUM_MAXIMUM_DEMAND_MONTH',
obis.BELGIUM_MAXIMUM_DEMAND_13_MONTHS: 'BELGIUM_MAXIMUM_DEMAND_13_MONTHS',
obis.BELGIUM_MAX_POWER_PER_PHASE: 'BELGIUM_MAX_POWER_PER_PHASE',
obis.BELGIUM_MAX_CURRENT_PER_PHASE: 'BELGIUM_MAX_CURRENT_PER_PHASE',
obis.BELGIUM_MBUS1_DEVICE_TYPE: 'BELGIUM_MBUS1_DEVICE_TYPE',
obis.BELGIUM_MBUS1_EQUIPMENT_IDENTIFIER: 'BELGIUM_MBUS1_EQUIPMENT_IDENTIFIER',
obis.BELGIUM_MBUS1_VALVE_POSITION: 'BELGIUM_MBUS1_VALVE_POSITION',
obis.BELGIUM_MBUS1_METER_READING1: 'BELGIUM_MBUS1_METER_READING1',
obis.BELGIUM_MBUS1_METER_READING2: 'BELGIUM_MBUS1_METER_READING2',
obis.BELGIUM_MBUS2_EQUIPMENT_IDENTIFIER: 'BELGIUM_MBUS2_EQUIPMENT_IDENTIFIER',
obis.BELGIUM_MBUS2_VALVE_POSITION: 'BELGIUM_MBUS2_VALVE_POSITION',
obis.BELGIUM_MBUS2_METER_READING1: 'BELGIUM_MBUS2_METER_READING1',
obis.BELGIUM_MBUS2_METER_READING2: 'BELGIUM_MBUS2_METER_READING2',
obis.BELGIUM_MBUS3_EQUIPMENT_IDENTIFIER: 'BELGIUM_MBUS3_EQUIPMENT_IDENTIFIER',
obis.BELGIUM_MBUS3_VALVE_POSITION: 'BELGIUM_MBUS3_VALVE_POSITION',
obis.BELGIUM_MBUS3_METER_READING1: 'BELGIUM_MBUS3_METER_READING1',
obis.BELGIUM_MBUS3_METER_READING2: 'BELGIUM_MBUS3_METER_READING2',
obis.BELGIUM_MBUS4_EQUIPMENT_IDENTIFIER: 'BELGIUM_MBUS4_EQUIPMENT_IDENTIFIER',
obis.BELGIUM_MBUS4_VALVE_POSITION: 'BELGIUM_MBUS4_VALVE_POSITION',
obis.BELGIUM_MBUS4_METER_READING1: 'BELGIUM_MBUS4_METER_READING1',
obis.BELGIUM_MBUS4_METER_READING2: 'BELGIUM_MBUS4_METER_READING2',
obis.LUXEMBOURG_EQUIPMENT_IDENTIFIER: 'LUXEMBOURG_EQUIPMENT_IDENTIFIER',
obis.Q3D_EQUIPMENT_IDENTIFIER: 'Q3D_EQUIPMENT_IDENTIFIER',
obis.Q3D_EQUIPMENT_STATE: 'Q3D_EQUIPMENT_STATE',
obis.Q3D_EQUIPMENT_SERIALNUMBER: 'Q3D_EQUIPMENT_SERIALNUMBER',
obis.BELGIUM_MBUS2_DEVICE_TYPE: 'BELGIUM_MBUS2_DEVICE_TYPE'
}
REVERSE_EN = dict([(v, k) for k, v in EN.items()])

22
dsmr_parser/obis_references.py
View File

@ -10,8 +10,12 @@ P1_MESSAGE_HEADER = r'^\d-\d:0\.2\.8.+?\r\n'
P1_MESSAGE_TIMESTAMP = r'^\d-\d:1\.0\.0.+?\r\n'
ELECTRICITY_USED_TARIFF_1 = r'^\d-\d:1\.8\.1.+?\r\n'
ELECTRICITY_USED_TARIFF_2 = r'^\d-\d:1\.8\.2.+?\r\n'
ELECTRICITY_USED_TARIFF_3 = r'^\d-\d:1\.8\.3.+?\r\n'
ELECTRICITY_USED_TARIFF_4 = r'^\d-\d:1\.8\.4.+?\r\n'
ELECTRICITY_DELIVERED_TARIFF_1 = r'^\d-\d:2\.8\.1.+?\r\n'
ELECTRICITY_DELIVERED_TARIFF_2 = r'^\d-\d:2\.8\.2.+?\r\n'
ELECTRICITY_DELIVERED_TARIFF_3 = r'^\d-\d:2\.8\.3.+?\r\n'
ELECTRICITY_DELIVERED_TARIFF_4 = r'^\d-\d:2\.8\.4.+?\r\n'
CURRENT_REACTIVE_IMPORTED = r'^\d-\d:3\.7\.0.+?\r\n'
ELECTRICITY_REACTIVE_IMPORTED_TOTAL = r'^\d-\d:3\.8\.0.+?\r\n'
ELECTRICITY_REACTIVE_IMPORTED_TARIFF_1 = r'^\d-\d:3\.8\.1.+?\r\n'
@ -122,3 +126,21 @@ LUXEMBOURG_EQUIPMENT_IDENTIFIER = r'^\d-\d:42\.0\.0.+?\r\n' # Logical device na
Q3D_EQUIPMENT_IDENTIFIER = r'^\d-\d:0\.0\.0.+?\r\n' # Logical device name
Q3D_EQUIPMENT_STATE = r'^\d-\d:96\.5\.5.+?\r\n' # Device state (hexadecimal)
Q3D_EQUIPMENT_SERIALNUMBER = r'^\d-\d:96\.1\.255.+?\r\n' # Device Serialnumber
# EON Hungary
EON_HU_ELECTRICITY_REACTIVE_TOTAL_Q1 = r'^\d-\d:5\.8\.0.+?\r\n'
EON_HU_ELECTRICITY_REACTIVE_TOTAL_Q2 = r'^\d-\d:6\.8\.0.+?\r\n'
EON_HU_ELECTRICITY_REACTIVE_TOTAL_Q3 = r'^\d-\d:7\.8\.0.+?\r\n'
EON_HU_ELECTRICITY_REACTIVE_TOTAL_Q4 = r'^\d-\d:8\.8\.0.+?\r\n'
EON_HU_ELECTRICITY_COMBINED = r'^\d-\d:15\.8\.0.+?\r\n'
EON_HU_INSTANTANEOUS_POWER_FACTOR_TOTAL = r'^\d-\d:13\.7\.0.+?\r\n'
EON_HU_INSTANTANEOUS_POWER_FACTOR_L1 = r'^\d-\d:33\.7\.0.+?\r\n'
EON_HU_INSTANTANEOUS_POWER_FACTOR_L2 = r'^\d-\d:53\.7\.0.+?\r\n'
EON_HU_INSTANTANEOUS_POWER_FACTOR_L3 = r'^\d-\d:73\.7\.0.+?\r\n'
EON_HU_FREQUENCY = r'^\d-\d:14\.7\.0.+?\r\n'
EON_HU_INSTANTANEOUS_REACTIVE_POWER_Q1 = r'^\d-\d:5\.7\.0.+?\r\n'
EON_HU_INSTANTANEOUS_REACTIVE_POWER_Q2 = r'^\d-\d:6\.7\.0.+?\r\n'
EON_HU_INSTANTANEOUS_REACTIVE_POWER_Q3 = r'^\d-\d:7\.7\.0.+?\r\n'
EON_HU_INSTANTANEOUS_REACTIVE_POWER_Q4 = r'^\d-\d:8\.7\.0.+?\r\n'
EON_HU_MAX_POWER_ON_L2 = r'^\d-\d:51\.4\.0.+?\r\n'
EON_HU_MAX_POWER_ON_L3 = r'^\d-\d:71\.4\.0.+?\r\n'

14
dsmr_parser/objects.py
View File

@ -5,8 +5,6 @@ import json
import pytz
from dsmr_parser import obis_name_mapping
class Telegram(dict):
"""
@ -27,22 +25,21 @@ class Telegram(dict):
self._mbus_devices = []
super().__init__(*args, **kwargs)
def add(self, obis_reference, dsmr_object):
def add(self, obis_reference, dsmr_object, obis_name):
# Update name mapping used to get value by attribute. Example: telegram.P1_MESSAGE_HEADER
obis_name = obis_name_mapping.EN[obis_reference]
setattr(self, obis_name, dsmr_object)
if obis_name not in self._item_names: # TODO repeating obis references
self._item_names.append(obis_name)
# TODO isinstance check: MaxDemandParser (BELGIUM_MAXIMUM_DEMAND_13_MONTHS) returns a list
if isinstance(dsmr_object, DSMRObject) and dsmr_object.is_mbus_reading:
self._add_mbus(obis_reference, dsmr_object)
self._add_mbus(obis_reference, dsmr_object, obis_name)
# Fill dict which is only used for backwards compatibility
if obis_reference not in self:
self[obis_reference] = dsmr_object
def _add_mbus(self, obis_reference, dsmr_object):
def _add_mbus(self, obis_reference, dsmr_object, obis_name):
"""
The given DsmrObject is assumed to be Mbus related and will be grouped into a MbusDevice.
Grouping is done by the DsmrObject channel ID.
@ -55,7 +52,7 @@ class Telegram(dict):
mbus_device = MbusDevice(channel_id=channel_id)
self._mbus_devices.append(mbus_device)
mbus_device.add(obis_reference, dsmr_object)
mbus_device.add(obis_reference, dsmr_object, obis_name)
if not hasattr(self, 'MBUS_DEVICES'):
setattr(self, 'MBUS_DEVICES', self._mbus_devices)
@ -331,10 +328,9 @@ class MbusDevice:
self.channel_id = channel_id
self._item_names = []
def add(self, obis_reference, dsmr_object):
def add(self, obis_reference, dsmr_object, obis_name):
# Update name mapping used to get value by attribute. Example: telegram.P1_MESSAGE_HEADER
# Also keep track of the added names internally
obis_name = obis_name_mapping.EN[obis_reference]
setattr(self, obis_name, dsmr_object)
self._item_names.append(obis_name)

22
dsmr_parser/parsers.py
View File

@ -29,8 +29,8 @@ class TelegramParser(object):
self.telegram_specification = telegram_specification
# Regexes are compiled once to improve performance
self.telegram_specification_regexes = {
signature: re.compile(signature, re.DOTALL | re.MULTILINE)
for signature in self.telegram_specification['objects'].keys()
object["obis_reference"]: re.compile(object["obis_reference"], re.DOTALL | re.MULTILINE)
for object in self.telegram_specification['objects']
}
def parse(self, telegram_data, encryption_key="", authentication_key="", throw_ex=False): # noqa: C901
@ -84,25 +84,31 @@ class TelegramParser(object):
telegram = Telegram()
for signature, parser in self.telegram_specification['objects'].items():
pattern = self.telegram_specification_regexes[signature]
for object in self.telegram_specification['objects']:
pattern = self.telegram_specification_regexes[object["obis_reference"]]
matches = pattern.findall(telegram_data)
# Some signatures are optional and may not be present,
# so only parse lines that match
for match in matches:
try:
dsmr_object = parser.parse(match)
dsmr_object = object["value_parser"].parse(match)
except ParseError:
logger.error("ignore line with signature {}, because parsing failed.".format(signature),
exc_info=True)
logger.error(
"ignore line with signature {}, because parsing failed.".format(object["obis_reference"]),
exc_info=True
)
if throw_ex:
raise
except Exception as err:
logger.error("Unexpected {}: {}".format(type(err), err))
raise
else:
telegram.add(obis_reference=signature, dsmr_object=dsmr_object)
telegram.add(
obis_reference=object["obis_reference"],
dsmr_object=dsmr_object,
obis_name=object["value_name"]
)
return telegram

1603
dsmr_parser/telegram_specifications.py
View File

File diff suppressed because it is too large Load Diff

52
test/example_telegrams.py
View File

@ -311,3 +311,55 @@ TELEGRAM_ISKRA_IE = (
'0-1:96.1.1()\r\n'
'!AD3B\r\n'
)
# V5 telegram of EON in Hungary
TELEGRAM_V5_EON_HU = (
'/SAG5SAG-METER\r\n'
'\r\n'
'0-0:1.0.0(230724150730S)\r\n'
'0-0:42.0.0(53414733303832323030303032313630)\r\n'
'0-0:96.1.0(383930303832323030303032313630)\r\n'
'0-0:96.14.0(0001)\r\n'
'0-0:96.3.10(1)\r\n'
'0-0:17.0.0(90.000*kW)\r\n'
'1-0:1.8.0(000173.640*kWh)\r\n'
'1-0:1.8.1(000047.719*kWh)\r\n'
'1-0:1.8.2(000125.921*kWh)\r\n'
'1-0:1.8.3(000000.000*kWh)\r\n'
'1-0:1.8.4(000000.000*kWh)\r\n'
'1-0:2.8.0(000627.177*kWh)\r\n'
'1-0:2.8.1(000401.829*kWh)\r\n'
'1-0:2.8.2(000225.348*kWh)\r\n'
'1-0:2.8.3(000000.000*kWh)\r\n'
'1-0:2.8.4(000000.000*kWh)\r\n'
'1-0:3.8.0(000000.123*kvarh)\r\n'
'1-0:4.8.0(000303.131*kvarh)\r\n'
'1-0:5.8.0(000000.668*kvarh)\r\n'
'1-0:6.8.0(000000.071*kvarh)\r\n'
'1-0:7.8.0(000160.487*kvarh)\r\n'
'1-0:8.8.0(000143.346*kvarh)\r\n'
'1-0:15.8.0(000800.817*kWh)\r\n'
'1-0:32.7.0(240.4*V)\r\n'
'1-0:52.7.0(239.1*V)\r\n'
'1-0:72.7.0(241.2*V)\r\n'
'1-0:31.7.0(003*A)\r\n'
'1-0:51.7.0(004*A)\r\n'
'1-0:71.7.0(003*A)\r\n'
'1-0:13.7.0(4.556)\r\n'
'1-0:33.7.0(4.591)\r\n'
'1-0:53.7.0(4.542)\r\n'
'1-0:73.7.0(4.552)\r\n'
'1-0:14.7.0(50.00*Hz)\r\n'
'1-0:1.7.0(00.000*kW)\r\n'
'1-0:2.7.0(02.601*kW)\r\n'
'1-0:5.7.0(00.000*kvar)\r\n'
'1-0:6.7.0(00.000*kvar)\r\n'
'1-0:7.7.0(00.504*kvar)\r\n'
'1-0:8.7.0(00.000*kvar)\r\n'
'1-0:31.4.0(200.00*A)\r\n'
'1-0:51.4.0(200.00*A)\r\n'
'1-0:71.4.0(200.00*A)\r\n'
'0-0:98.1.0(230701000000S)(000040.777*kWh)(000008.950*kWh)(000031.827*kWh)(000142.250*kWh)(000111.164*kWh)(000031.086*kWh)(000000.030*kvarh)(000073.988*kvarh)(000000.205*kvarh)(000000.048*kvarh)(000039.199*kvarh)(000035.020*kvarh)(000183.027*kWh)(03.564*kW)(02.156*kW)(03.564*kW)(04.104*kW)(04.104*kW)(03.400*kW)\r\n'
'0-0:96.13.0()\r\n'
'!99DA\r\n'
)

24
test/objects/test_mbusdevice.py
View File

@ -12,19 +12,31 @@ class MbusDeviceTest(unittest.TestCase):
def setUp(self):
v5_objects = telegram_specifications.V5['objects']
device_type_parser = v5_objects[obis_references.DEVICE_TYPE]
device_type_parser = [
object["value_parser"]
for object in v5_objects
if object["obis_reference"] == obis_references.DEVICE_TYPE
][0]
device_type = device_type_parser.parse('0-2:24.1.0(003)\r\n')
equipment_parser = v5_objects[obis_references.EQUIPMENT_IDENTIFIER_GAS]
equipment_parser = [
object["value_parser"]
for object in v5_objects
if object["obis_reference"] == obis_references.EQUIPMENT_IDENTIFIER_GAS
][0]
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_parser = [
object["value_parser"]
for object in v5_objects
if object["obis_reference"] == obis_references.HOURLY_GAS_METER_READING
][0]
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)
mbus_device.add(obis_references.DEVICE_TYPE, device_type, "DEVICE_TYPE")
mbus_device.add(obis_references.EQUIPMENT_IDENTIFIER_GAS, equipment, "EQUIPMENT_IDENTIFIER_GAS")
mbus_device.add(obis_references.HOURLY_GAS_METER_READING, gas_reading, "HOURLY_GAS_METER_READING")
self.mbus_device = mbus_device

8
test/objects/test_telegram.py
View File

@ -4,7 +4,7 @@ import datetime
import pytz
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 ProfileGenericObject
@ -314,8 +314,8 @@ class TelegramTest(unittest.TestCase):
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_list = [object["value_name"] for object in
telegram_specifications.V4['objects']]
V4_name_set = set(V4_name_list)
item_names_tested_set = set(self.item_names_tested)
@ -329,7 +329,7 @@ class TelegramTest(unittest.TestCase):
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(obis_name, "P1_MESSAGE_HEADER")
self.assertEqual(dsmr_object.value, '50')
def test_mbus_devices(self):

308
test/test_parse_v5_eon_hungary.py Normal file
View File

@ -0,0 +1,308 @@
from decimal import Decimal
import datetime
import unittest
import pytz
from dsmr_parser import telegram_specifications
from dsmr_parser.exceptions import InvalidChecksumError, ParseError
from dsmr_parser.objects import CosemObject
from dsmr_parser.parsers import TelegramParser
from test.example_telegrams import TELEGRAM_V5_EON_HU
class TelegramParserV5EONHUTest(unittest.TestCase):
""" Test parsing of a DSMR v5 EON Hungary telegram. """
def test_parse(self):
parser = TelegramParser(telegram_specifications.EON_HUNGARY)
try:
telegram = parser.parse(TELEGRAM_V5_EON_HU, throw_ex=True)
except Exception as ex:
assert False, f"parse trigged an exception {ex}"
# P1_MESSAGE_TIMESTAMP (0-0:1.0.0)
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 == \
pytz.timezone("Europe/Budapest").localize(datetime.datetime(2023, 7, 24, 15, 7, 30))
# EON_HU_COSEM_LOGICAL_DEVICE_NAME (0-0:42.0.0)
assert isinstance(telegram.COSEM_LOGICAL_DEVICE_NAME, CosemObject)
assert telegram.COSEM_LOGICAL_DEVICE_NAME.unit is None
assert isinstance(telegram.COSEM_LOGICAL_DEVICE_NAME.value, str)
assert telegram.COSEM_LOGICAL_DEVICE_NAME.value == '53414733303832323030303032313630'
# EON_HU_EQUIPMENT_SERIAL_NUMBER (0-0:96.1.0)
assert isinstance(telegram.EQUIPMENT_SERIAL_NUMBER, CosemObject)
assert telegram.EQUIPMENT_SERIAL_NUMBER.unit is None
assert isinstance(telegram.EQUIPMENT_SERIAL_NUMBER.value, str)
assert telegram.EQUIPMENT_SERIAL_NUMBER.value == '383930303832323030303032313630'
# ELECTRICITY_ACTIVE_TARIFF (0-0:96.14.0)
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 == '0001'
# ACTUAL_SWITCH_POSITION (0-0:96.3.10)
assert isinstance(telegram.ACTUAL_SWITCH_POSITION, CosemObject)
assert telegram.ACTUAL_SWITCH_POSITION.unit is None
assert isinstance(telegram.ACTUAL_SWITCH_POSITION.value, str)
assert telegram.ACTUAL_SWITCH_POSITION.value == '1'
# ACTUAL_TRESHOLD_ELECTRICITY (0-0:17.0.0)
assert isinstance(telegram.ACTUAL_TRESHOLD_ELECTRICITY, CosemObject)
assert telegram.ACTUAL_TRESHOLD_ELECTRICITY.unit == 'kW'
assert isinstance(telegram.ACTUAL_TRESHOLD_ELECTRICITY.value, Decimal)
assert telegram.ACTUAL_TRESHOLD_ELECTRICITY.value == Decimal('90.000')
# ELECTRICITY_IMPORTED_TOTAL (1-0:1.8.0)
assert isinstance(telegram.ELECTRICITY_IMPORTED_TOTAL, CosemObject)
assert telegram.ELECTRICITY_IMPORTED_TOTAL.unit == 'kWh'
assert isinstance(telegram.ELECTRICITY_IMPORTED_TOTAL.value, Decimal)
assert telegram.ELECTRICITY_IMPORTED_TOTAL.value == Decimal('000173.640')
# ELECTRICITY_USED_TARIFF_1 (1-0:1.8.1)
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('000047.719')
# ELECTRICITY_USED_TARIFF_2 (1-0:1.8.2)
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('000125.921')
# EON_HU_ELECTRICITY_USED_TARIFF_3 (1-0:1.8.3)
assert isinstance(telegram.ELECTRICITY_USED_TARIFF_3, CosemObject)
assert telegram.ELECTRICITY_USED_TARIFF_3.unit == 'kWh'
assert isinstance(telegram.ELECTRICITY_USED_TARIFF_3.value, Decimal)
assert telegram.ELECTRICITY_USED_TARIFF_3.value == Decimal('000000.000')
# EON_HU_ELECTRICITY_USED_TARIFF_4 (1-0:1.8.4)
assert isinstance(telegram.ELECTRICITY_USED_TARIFF_4, CosemObject)
assert telegram.ELECTRICITY_USED_TARIFF_4.unit == 'kWh'
assert isinstance(telegram.ELECTRICITY_USED_TARIFF_4.value, Decimal)
assert telegram.ELECTRICITY_USED_TARIFF_4.value == Decimal('000000.000')
# ELECTRICITY_EXPORTED_TOTAL (1-0:2.8.0)
assert isinstance(telegram.ELECTRICITY_EXPORTED_TOTAL, CosemObject)
assert telegram.ELECTRICITY_EXPORTED_TOTAL.unit == 'kWh'
assert isinstance(telegram.ELECTRICITY_EXPORTED_TOTAL.value, Decimal)
assert telegram.ELECTRICITY_EXPORTED_TOTAL.value == Decimal('000627.177')
# ELECTRICITY_DELIVERED_TARIFF_1 (1-0:2.8.1)
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('000401.829')
# ELECTRICITY_DELIVERED_TARIFF_2 (1-0:2.8.2)
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('000225.348')
# EON_HU_ELECTRICITY_DELIVERED_TARIFF_3 (1-0:2.8.3)
assert isinstance(telegram.ELECTRICITY_DELIVERED_TARIFF_3, CosemObject)
assert telegram.ELECTRICITY_DELIVERED_TARIFF_3.unit == 'kWh'
assert isinstance(telegram.ELECTRICITY_DELIVERED_TARIFF_3.value, Decimal)
assert telegram.ELECTRICITY_DELIVERED_TARIFF_3.value == Decimal('000000.000')
# EON_HU_ELECTRICITY_DELIVERED_TARIFF_4 (1-0:2.8.4)
assert isinstance(telegram.ELECTRICITY_DELIVERED_TARIFF_4, CosemObject)
assert telegram.ELECTRICITY_DELIVERED_TARIFF_4.unit == 'kWh'
assert isinstance(telegram.ELECTRICITY_DELIVERED_TARIFF_4.value, Decimal)
assert telegram.ELECTRICITY_DELIVERED_TARIFF_4.value == Decimal('000000.000')
# ELECTRICITY_REACTIVE_IMPORTED_TOTAL (1-0:3.8.0)
assert isinstance(telegram.ELECTRICITY_REACTIVE_IMPORTED_TOTAL, CosemObject)
assert telegram.ELECTRICITY_REACTIVE_IMPORTED_TOTAL.unit == 'kvarh'
assert isinstance(telegram.ELECTRICITY_REACTIVE_IMPORTED_TOTAL.value, Decimal)
assert telegram.ELECTRICITY_REACTIVE_IMPORTED_TOTAL.value == Decimal('000000.123')
# ELECTRICITY_REACTIVE_EXPORTED_TOTAL (1-0:4.8.0)
assert isinstance(telegram.ELECTRICITY_REACTIVE_EXPORTED_TOTAL, CosemObject)
assert telegram.ELECTRICITY_REACTIVE_EXPORTED_TOTAL.unit == 'kvarh'
assert isinstance(telegram.ELECTRICITY_REACTIVE_EXPORTED_TOTAL.value, Decimal)
assert telegram.ELECTRICITY_REACTIVE_EXPORTED_TOTAL.value == Decimal('000303.131')
# EON_HU_ELECTRICITY_REACTIVE_TOTAL_Q1 (1-0:5.8.0)
assert isinstance(telegram.ELECTRICITY_REACTIVE_TOTAL_Q1, CosemObject)
assert telegram.ELECTRICITY_REACTIVE_TOTAL_Q1.unit == 'kvarh'
assert isinstance(telegram.ELECTRICITY_REACTIVE_TOTAL_Q1.value, Decimal)
assert telegram.ELECTRICITY_REACTIVE_TOTAL_Q1.value == Decimal('000000.668')
# EON_HU_ELECTRICITY_REACTIVE_TOTAL_Q2 (1-0:6.8.0)
assert isinstance(telegram.ELECTRICITY_REACTIVE_TOTAL_Q2, CosemObject)
assert telegram.ELECTRICITY_REACTIVE_TOTAL_Q2.unit == 'kvarh'
assert isinstance(telegram.ELECTRICITY_REACTIVE_TOTAL_Q2.value, Decimal)
assert telegram.ELECTRICITY_REACTIVE_TOTAL_Q2.value == Decimal('000000.071')
# EON_HU_ELECTRICITY_REACTIVE_TOTAL_Q3 (1-0:7.8.0)
assert isinstance(telegram.ELECTRICITY_REACTIVE_TOTAL_Q3, CosemObject)
assert telegram.ELECTRICITY_REACTIVE_TOTAL_Q3.unit == 'kvarh'
assert isinstance(telegram.ELECTRICITY_REACTIVE_TOTAL_Q3.value, Decimal)
assert telegram.ELECTRICITY_REACTIVE_TOTAL_Q3.value == Decimal('000160.487')
# EON_HU_ELECTRICITY_REACTIVE_TOTAL_Q4 (1-0:8.8.0)
assert isinstance(telegram.ELECTRICITY_REACTIVE_TOTAL_Q4, CosemObject)
assert telegram.ELECTRICITY_REACTIVE_TOTAL_Q4.unit == 'kvarh'
assert isinstance(telegram.ELECTRICITY_REACTIVE_TOTAL_Q4.value, Decimal)
assert telegram.ELECTRICITY_REACTIVE_TOTAL_Q4.value == Decimal('000143.346')
# EON_HU_ELECTRICITY_COMBINED (1-0:15.8.0)
assert isinstance(telegram.ELECTRICITY_COMBINED, CosemObject)
assert telegram.ELECTRICITY_COMBINED.unit == 'kWh'
assert isinstance(telegram.ELECTRICITY_COMBINED.value, Decimal)
assert telegram.ELECTRICITY_COMBINED.value == Decimal('000800.817')
# INSTANTANEOUS_VOLTAGE_L2 (1-0:32.7.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('240.4')
# INSTANTANEOUS_VOLTAGE_L2 (1-0:52.7.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('239.1')
# INSTANTANEOUS_VOLTAGE_L3 (1-0:72.7.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('241.2')
# INSTANTANEOUS_CURRENT_L1 (1-0:31.7.0)
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('003')
# INSTANTANEOUS_CURRENT_L2 (1-0:51.7.0)
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('004')
# INSTANTANEOUS_CURRENT_L3 (1-0:71.7.0)
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('003')
# EON_HU_INSTANTANEOUS_POWER_FACTOR_TOTAL (1-0:13.7.0)
assert isinstance(telegram.INSTANTANEOUS_POWER_FACTOR_TOTAL, CosemObject)
assert telegram.INSTANTANEOUS_POWER_FACTOR_TOTAL.unit is None
assert isinstance(telegram.INSTANTANEOUS_POWER_FACTOR_TOTAL.value, Decimal)
assert telegram.INSTANTANEOUS_POWER_FACTOR_TOTAL.value == Decimal('4.556')
# EON_HU_INSTANTANEOUS_POWER_FACTOR_L1 (1-0:33.7.0)
assert isinstance(telegram.INSTANTANEOUS_POWER_FACTOR_L1, CosemObject)
assert telegram.INSTANTANEOUS_POWER_FACTOR_L1.unit is None
assert isinstance(telegram.INSTANTANEOUS_POWER_FACTOR_L1.value, Decimal)
assert telegram.INSTANTANEOUS_POWER_FACTOR_L1.value == Decimal('4.591')
# EON_HU_INSTANTANEOUS_POWER_FACTOR_L2 (1-0:53.7.0)
assert isinstance(telegram.INSTANTANEOUS_POWER_FACTOR_L2, CosemObject)
assert telegram.INSTANTANEOUS_POWER_FACTOR_L2.unit is None
assert isinstance(telegram.INSTANTANEOUS_POWER_FACTOR_L2.value, Decimal)
assert telegram.INSTANTANEOUS_POWER_FACTOR_L2.value == Decimal('4.542')
# EON_HU_INSTANTANEOUS_POWER_FACTOR_L3 (1-0:73.7.0)
assert isinstance(telegram.INSTANTANEOUS_POWER_FACTOR_L3, CosemObject)
assert telegram.INSTANTANEOUS_POWER_FACTOR_L3.unit is None
assert isinstance(telegram.INSTANTANEOUS_POWER_FACTOR_L3.value, Decimal)
assert telegram.INSTANTANEOUS_POWER_FACTOR_L3.value == Decimal('4.552')
# EON_HU_FREQUENCY (1-0:14.7.0)
assert isinstance(telegram.FREQUENCY, CosemObject)
assert telegram.FREQUENCY.unit == "Hz"
assert isinstance(telegram.FREQUENCY.value, Decimal)
assert telegram.FREQUENCY.value == Decimal('50.00')
# CURRENT_ELECTRICITY_USAGE (1-0:1.7.0)
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('00.000')
# CURRENT_ELECTRICITY_DELIVERY (1-0:2.7.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('02.601')
# EON_HU_INSTANTANEOUS_REACTIVE_POWER_Q1 (1-0:5.7.0)
assert isinstance(telegram.INSTANTANEOUS_REACTIVE_POWER_Q1, CosemObject)
assert telegram.INSTANTANEOUS_REACTIVE_POWER_Q1.unit == 'kvar'
assert isinstance(telegram.INSTANTANEOUS_REACTIVE_POWER_Q1.value, Decimal)
assert telegram.INSTANTANEOUS_REACTIVE_POWER_Q1.value == Decimal('00.000')
# EON_HU_INSTANTANEOUS_REACTIVE_POWER_Q2 (1-0:6.7.0)
assert isinstance(telegram.INSTANTANEOUS_REACTIVE_POWER_Q2, CosemObject)
assert telegram.INSTANTANEOUS_REACTIVE_POWER_Q2.unit == 'kvar'
assert isinstance(telegram.INSTANTANEOUS_REACTIVE_POWER_Q2.value, Decimal)
assert telegram.INSTANTANEOUS_REACTIVE_POWER_Q2.value == Decimal('00.000')
# EON_HU_INSTANTANEOUS_REACTIVE_POWER_Q3 (1-0:7.7.0)
assert isinstance(telegram.INSTANTANEOUS_REACTIVE_POWER_Q3, CosemObject)
assert telegram.INSTANTANEOUS_REACTIVE_POWER_Q3.unit == 'kvar'
assert isinstance(telegram.INSTANTANEOUS_REACTIVE_POWER_Q3.value, Decimal)
assert telegram.INSTANTANEOUS_REACTIVE_POWER_Q3.value == Decimal('00.504')
# EON_HU_INSTANTANEOUS_REACTIVE_POWER_Q4 (1-0:8.7.0)
assert isinstance(telegram.INSTANTANEOUS_REACTIVE_POWER_Q4, CosemObject)
assert telegram.INSTANTANEOUS_REACTIVE_POWER_Q4.unit == 'kvar'
assert isinstance(telegram.INSTANTANEOUS_REACTIVE_POWER_Q4.value, Decimal)
assert telegram.INSTANTANEOUS_REACTIVE_POWER_Q4.value == Decimal('00.000')
# EON_HU_MAX_POWER_ON_L1 (1-0:31.4.0)
assert isinstance(telegram.MAX_POWER_ON_L1, CosemObject)
assert telegram.MAX_POWER_ON_L1.unit == 'A'
assert isinstance(telegram.MAX_POWER_ON_L1.value, Decimal)
assert telegram.MAX_POWER_ON_L1.value == Decimal('200.00')
# EON_HU_MAX_POWER_ON_L2 (1-0:31.4.0)
assert isinstance(telegram.MAX_POWER_ON_L2, CosemObject)
assert telegram.MAX_POWER_ON_L2.unit == 'A'
assert isinstance(telegram.MAX_POWER_ON_L2.value, Decimal)
assert telegram.MAX_POWER_ON_L2.value == Decimal('200.00')
# EON_HU_MAX_POWER_ON_L3 (1-0:31.4.0)
assert isinstance(telegram.MAX_POWER_ON_L3, CosemObject)
assert telegram.MAX_POWER_ON_L3.unit == 'A'
assert isinstance(telegram.MAX_POWER_ON_L3.value, Decimal)
assert telegram.MAX_POWER_ON_L3.value == Decimal('200.00')
# TEXT_MESSAGE (0-0:96.13.0)
assert isinstance(telegram.TEXT_MESSAGE, CosemObject)
assert telegram.TEXT_MESSAGE.unit is None
assert telegram.TEXT_MESSAGE.value is None
def test_checksum_valid(self):
# No exception is raised.
TelegramParser.validate_checksum(TELEGRAM_V5_EON_HU)
def test_checksum_invalid(self):
# Remove the electricty used data value. This causes the checksum to
# not match anymore.
corrupted_telegram = TELEGRAM_V5_EON_HU.replace(
'1-0:1.8.1(000047.719*kWh)\r\n',
''
)
with self.assertRaises(InvalidChecksumError):
TelegramParser.validate_checksum(corrupted_telegram)
def test_checksum_missing(self):
# Remove the checksum value causing a ParseError.
corrupted_telegram = TELEGRAM_V5_EON_HU.replace('!99DA\r\n', '')
with self.assertRaises(ParseError):
TelegramParser.validate_checksum(corrupted_telegram)