retoor/dsmr_parser
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Change DSMR 5 and 5B to use DSMR devices (#142)

Browse Source 
At least the BE and the NL DSMR devices use MBUS.
Therefore we make the MBUS obis reference generic, and convert the V5
telegram to use MBUS values.

Finally set self.maxDiff = None to have a full test output on some
tests.
This commit is contained in:
dupondje
2024-03-07 11:21:23 +01:00
committed by GitHub
parent fcbcbb30e0
commit bc35a051f2
7 changed files with 190 additions and 316 deletions
Download Patch File Download Diff File Expand all files Collapse all files
test/objects/test_mbusdevice.py
+21 -21
View File
@@ -15,59 +15,59 @@ class MbusDeviceTest(unittest.TestCase):
device_type_parser = [
object["value_parser"]
for object in v5_objects
if object["obis_reference"] == obis_references.DEVICE_TYPE
if object["obis_reference"] == obis_references.MBUS_DEVICE_TYPE
][0]
device_type = device_type_parser.parse('0-2:24.1.0(003)\r\n')
equipment_parser = [
object["value_parser"]
for object in v5_objects
if object["obis_reference"] == obis_references.EQUIPMENT_IDENTIFIER_GAS
if object["obis_reference"] == obis_references.MBUS_EQUIPMENT_IDENTIFIER
][0]
equipment = equipment_parser.parse('0-2:96.1.0(4730303339303031393336393930363139)\r\n')
gas_reading_parser = [
object["value_parser"]
for object in v5_objects
if object["obis_reference"] == obis_references.HOURLY_GAS_METER_READING
if object["obis_reference"] == obis_references.MBUS_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, "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")
mbus_device = MbusDevice(channel_id=2)
mbus_device.add(obis_references.MBUS_DEVICE_TYPE, device_type, "MBUS_DEVICE_TYPE")
mbus_device.add(obis_references.MBUS_EQUIPMENT_IDENTIFIER, equipment, "MBUS_EQUIPMENT_IDENTIFIER")
mbus_device.add(obis_references.MBUS_METER_READING, gas_reading, "MBUS_METER_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.MBUS_DEVICE_TYPE.value, 3)
self.assertEqual(self.mbus_device.MBUS_DEVICE_TYPE.unit, None)
self.assertEqual(self.mbus_device.EQUIPMENT_IDENTIFIER_GAS.value,
self.assertEqual(self.mbus_device.MBUS_EQUIPMENT_IDENTIFIER.value,
'4730303339303031393336393930363139')
self.assertEqual(self.mbus_device.EQUIPMENT_IDENTIFIER_GAS.unit, None)
self.assertEqual(self.mbus_device.MBUS_EQUIPMENT_IDENTIFIER.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')
self.assertEqual(self.mbus_device.MBUS_METER_READING.value, Decimal('246.138'))
self.assertEqual(self.mbus_device.MBUS_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'}}
'CHANNEL_ID': 2,
'MBUS_DEVICE_TYPE': {'value': 3, 'unit': None},
'MBUS_EQUIPMENT_IDENTIFIER': {'value': '4730303339303031393336393930363139', 'unit': None},
'MBUS_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'
'MBUS DEVICE (channel 2)\n'
'\tMBUS_DEVICE_TYPE: 3 [None]\n'
'\tMBUS_EQUIPMENT_IDENTIFIER: 4730303339303031393336393930363139 [None]\n'
'\tMBUS_METER_READING: 246.138 [m3] at 2020-04-26T20:30:01+00:00\n'
)
)
test/objects/test_telegram.py
+41 -44
View File
@@ -216,14 +216,6 @@ class TelegramTest(unittest.TestCase):
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',
@@ -272,7 +264,15 @@ class TelegramTest(unittest.TestCase):
value_type=Decimal,
value_val=Decimal('0'))
# EQUIPMENT_IDENTIFIER_GAS (0-x:96.1.0)
# DEVICE_TYPE (0-1:24.1.0)
self.verify_telegram_item(telegram,
'DEVICE_TYPE',
object_type=CosemObject,
unit_val=None,
value_type=int,
value_val=3)
# EQUIPMENT_IDENTIFIER_GAS (0-1:96.1.0)
self.verify_telegram_item(telegram,
'EQUIPMENT_IDENTIFIER_GAS',
object_type=CosemObject,
@@ -340,28 +340,28 @@ 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'))
self.assertEqual(mbus_device_1.MBUS_DEVICE_TYPE.value, 3)
self.assertEqual(mbus_device_1.MBUS_EQUIPMENT_IDENTIFIER.value, None)
self.assertEqual(mbus_device_1.MBUS_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'))
self.assertEqual(mbus_device_2.MBUS_DEVICE_TYPE.value, 3)
self.assertEqual(mbus_device_2.MBUS_EQUIPMENT_IDENTIFIER.value, '4730303339303031393336393930363139')
self.assertEqual(mbus_device_2.MBUS_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'))
self.assertEqual(mbus_device_1.MBUS_DEVICE_TYPE.value, 3)
self.assertEqual(mbus_device_1.MBUS_EQUIPMENT_IDENTIFIER.value, None)
self.assertEqual(mbus_device_1.MBUS_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'))
self.assertEqual(mbus_device_2.MBUS_DEVICE_TYPE.value, 3)
self.assertEqual(mbus_device_2.MBUS_EQUIPMENT_IDENTIFIER.value, '4730303339303031393336393930363139')
self.assertEqual(mbus_device_2.MBUS_METER_READING.value, Decimal('246.138'))
def test_without_mbus_devices(self):
parser = TelegramParser(telegram_specifications.V5, apply_checksum_validation=False)
@@ -375,11 +375,12 @@ class TelegramTest(unittest.TestCase):
telegram = parser.parse(TELEGRAM_V5)
json_data = json.loads(telegram.to_json())
self.maxDiff = None
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},
@@ -387,10 +388,6 @@ class TelegramTest(unittest.TestCase):
'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},
@@ -405,15 +402,16 @@ class TelegramTest(unittest.TestCase):
'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}},
'MBUS_DEVICE_TYPE': {'unit': None, 'value': 3},
'MBUS_EQUIPMENT_IDENTIFIER': {'unit': None,
'value': '3232323241424344313233343536373839'},
'MBUS_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}}],
'MBUS_DEVICE_TYPE': {'unit': None, 'value': 3},
'MBUS_EQUIPMENT_IDENTIFIER': {'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': [],
@@ -433,6 +431,8 @@ class TelegramTest(unittest.TestCase):
parser = TelegramParser(telegram_specifications.V5)
telegram = parser.parse(TELEGRAM_V5)
self.maxDiff = None
self.assertEqual(
str(telegram),
(
@@ -463,22 +463,19 @@ class TelegramTest(unittest.TestCase):
'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'
'MBUS DEVICE (channel 1)\n'
' MBUS_DEVICE_TYPE: 3 [None]\n'
' MBUS_EQUIPMENT_IDENTIFIER: 3232323241424344313233343536373839 [None]\n'
' MBUS_METER_READING: 0.107 [m3] at 2017-01-02T15:10:05+00:00\n'
'MBUS DEVICE (channel 2)\n'
' MBUS_DEVICE_TYPE: 3 [None]\n'
' MBUS_EQUIPMENT_IDENTIFIER: None [None]\n'
)
)
test/test_parse_fluvius.py
+60 -66
View File
@@ -228,47 +228,53 @@ class TelegramParserFluviusTest(unittest.TestCase):
assert result.TEXT_MESSAGE.unit is None
assert result.TEXT_MESSAGE.value is None
# BELGIUM_MBUS1_DEVICE_TYPE (0-1:24.1.0)
assert isinstance(result.BELGIUM_MBUS1_DEVICE_TYPE, CosemObject)
assert result.BELGIUM_MBUS1_DEVICE_TYPE.unit is None
assert isinstance(result.BELGIUM_MBUS1_DEVICE_TYPE.value, int)
assert result.BELGIUM_MBUS1_DEVICE_TYPE.value == 3
# MBUS DEVICE 1
mbus1 = result.get_mbus_device_by_channel(1)
# BELGIUM_MBUS1_EQUIPMENT_IDENTIFIER (0-1:96.1.1)
assert isinstance(result.BELGIUM_MBUS1_EQUIPMENT_IDENTIFIER, CosemObject)
assert result.BELGIUM_MBUS1_EQUIPMENT_IDENTIFIER.unit is None
assert isinstance(result.BELGIUM_MBUS1_EQUIPMENT_IDENTIFIER.value, str)
assert result.BELGIUM_MBUS1_EQUIPMENT_IDENTIFIER.value == '37464C4F32313139303333373333'
# MBUS_DEVICE_TYPE (0-1:24.1.0)
assert isinstance(mbus1.MBUS_DEVICE_TYPE, CosemObject)
assert mbus1.MBUS_DEVICE_TYPE.unit is None
assert isinstance(mbus1.MBUS_DEVICE_TYPE.value, int)
assert mbus1.MBUS_DEVICE_TYPE.value == 3
# BELGIUM_MBUS1_VALVE_POSITION (0-1:24.4.0)
assert isinstance(result.BELGIUM_MBUS1_VALVE_POSITION, CosemObject)
assert result.BELGIUM_MBUS1_VALVE_POSITION.unit is None
assert isinstance(result.BELGIUM_MBUS1_VALVE_POSITION.value, int)
assert result.BELGIUM_MBUS1_VALVE_POSITION.value == 1
# MBUS_EQUIPMENT_IDENTIFIER (0-1:96.1.1)
assert isinstance(mbus1.MBUS_EQUIPMENT_IDENTIFIER, CosemObject)
assert mbus1.MBUS_EQUIPMENT_IDENTIFIER.unit is None
assert isinstance(mbus1.MBUS_EQUIPMENT_IDENTIFIER.value, str)
assert mbus1.MBUS_EQUIPMENT_IDENTIFIER.value == '37464C4F32313139303333373333'
# BELGIUM_MBUS1_METER_READING2 (0-1:24.2.3)
assert isinstance(result.BELGIUM_MBUS1_METER_READING2, MBusObject)
assert result.BELGIUM_MBUS1_METER_READING2.unit == 'm3'
assert isinstance(result.BELGIUM_MBUS1_METER_READING2.value, Decimal)
assert result.BELGIUM_MBUS1_METER_READING2.value == Decimal('112.384')
# MBUS_VALVE_POSITION (0-1:24.4.0)
assert isinstance(result.MBUS_VALVE_POSITION, CosemObject)
assert result.MBUS_VALVE_POSITION.unit is None
assert isinstance(result.MBUS_VALVE_POSITION.value, int)
assert result.MBUS_VALVE_POSITION.value == 1
# BELGIUM_MBUS2_DEVICE_TYPE (0-2:24.1.0)
assert isinstance(result.BELGIUM_MBUS2_DEVICE_TYPE, CosemObject)
assert result.BELGIUM_MBUS2_DEVICE_TYPE.unit is None
assert isinstance(result.BELGIUM_MBUS2_DEVICE_TYPE.value, int)
assert result.BELGIUM_MBUS2_DEVICE_TYPE.value == 7
# MBUS_METER_READING (0-1:24.2.3)
assert isinstance(mbus1.MBUS_METER_READING, MBusObject)
assert mbus1.MBUS_METER_READING.unit == 'm3'
assert isinstance(mbus1.MBUS_METER_READING.value, Decimal)
assert mbus1.MBUS_METER_READING.value == Decimal('112.384')
# BELGIUM_MBUS2_EQUIPMENT_IDENTIFIER (0-2:96.1.1)
assert isinstance(result.BELGIUM_MBUS2_EQUIPMENT_IDENTIFIER, CosemObject)
assert result.BELGIUM_MBUS2_EQUIPMENT_IDENTIFIER.unit is None
assert isinstance(result.BELGIUM_MBUS2_EQUIPMENT_IDENTIFIER.value, str)
assert result.BELGIUM_MBUS2_EQUIPMENT_IDENTIFIER.value == '3853414731323334353637383930'
# MBUS DEVICE 2
mbus2 = result.get_mbus_device_by_channel(2)
# BELGIUM_MBUS2_METER_READING1 (0-1:24.2.1)
assert isinstance(result.BELGIUM_MBUS2_METER_READING1, MBusObject)
assert result.BELGIUM_MBUS2_METER_READING1.unit == 'm3'
assert isinstance(result.BELGIUM_MBUS2_METER_READING1.value, Decimal)
assert result.BELGIUM_MBUS2_METER_READING1.value == Decimal('872.234')
# MBUS_DEVICE_TYPE (0-2:24.1.0)
assert isinstance(mbus2.MBUS_DEVICE_TYPE, CosemObject)
assert mbus2.MBUS_DEVICE_TYPE.unit is None
assert isinstance(mbus2.MBUS_DEVICE_TYPE.value, int)
assert mbus2.MBUS_DEVICE_TYPE.value == 7
# MBUS_EQUIPMENT_IDENTIFIER (0-2:96.1.1)
assert isinstance(mbus2.MBUS_EQUIPMENT_IDENTIFIER, CosemObject)
assert mbus2.MBUS_EQUIPMENT_IDENTIFIER.unit is None
assert isinstance(mbus2.MBUS_EQUIPMENT_IDENTIFIER.value, str)
assert mbus2.MBUS_EQUIPMENT_IDENTIFIER.value == '3853414731323334353637383930'
# MBUS_METER_READING (0-1:24.2.1)
assert isinstance(mbus2.MBUS_METER_READING, MBusObject)
assert mbus2.MBUS_METER_READING.unit == 'm3'
assert isinstance(mbus2.MBUS_METER_READING.value, Decimal)
assert mbus2.MBUS_METER_READING.value == Decimal('872.234')
def test_checksum_valid(self):
# No exception is raised.
@@ -296,6 +302,8 @@ class TelegramParserFluviusTest(unittest.TestCase):
telegram = parser.parse(TELEGRAM_FLUVIUS_V171_ALT)
json_data = json.loads(telegram.to_json())
self.maxDiff = None
self.assertEqual(
json_data,
{'BELGIUM_VERSION_INFORMATION': {'value': '50217', 'unit': None},
@@ -338,26 +346,19 @@ class TelegramParserFluviusTest(unittest.TestCase):
'ACTUAL_TRESHOLD_ELECTRICITY': {'value': 999.9, 'unit': 'kW'},
'FUSE_THRESHOLD_L1': {'value': 999.0, 'unit': 'A'},
'TEXT_MESSAGE': {'value': None, 'unit': None},
'BELGIUM_MBUS1_DEVICE_TYPE': {'value': 3, 'unit': None},
'MBUS_DEVICES': [{'BELGIUM_MBUS1_DEVICE_TYPE': {'value': 3, 'unit': None},
'BELGIUM_MBUS1_EQUIPMENT_IDENTIFIER': {'value': '37464C4F32313233303838303237',
'MBUS_DEVICES': [{'MBUS_DEVICE_TYPE': {'value': 3, 'unit': None},
'MBUS_EQUIPMENT_IDENTIFIER': {'value': '37464C4F32313233303838303237',
'unit': None},
'BELGIUM_MBUS1_VALVE_POSITION': {'value': 1, 'unit': None},
'BELGIUM_MBUS1_METER_READING2': {'datetime': '2023-11-02T11:10:02+00:00',
'value': 92.287, 'unit': 'm3'},
'MBUS_VALVE_POSITION': {'value': 1, 'unit': None},
'MBUS_METER_READING': {'datetime': '2023-11-02T11:10:02+00:00',
'value': 92.287, 'unit': 'm3'},
'CHANNEL_ID': 1},
{'BELGIUM_MBUS2_DEVICE_TYPE': {'value': 7, 'unit': None},
'BELGIUM_MBUS2_EQUIPMENT_IDENTIFIER': {'value': '3853455430303030393631313733',
{'MBUS_DEVICE_TYPE': {'value': 7, 'unit': None},
'MBUS_EQUIPMENT_IDENTIFIER': {'value': '3853455430303030393631313733',
'unit': None},
'BELGIUM_MBUS2_METER_READING1': {'datetime': '2023-11-02T11:15:32+00:00',
'value': 8.579, 'unit': 'm3'},
'CHANNEL_ID': 2}],
'BELGIUM_MBUS1_EQUIPMENT_IDENTIFIER': {'value': '37464C4F32313233303838303237', 'unit': None},
'BELGIUM_MBUS1_VALVE_POSITION': {'value': 1, 'unit': None},
'BELGIUM_MBUS1_METER_READING2': {'datetime': '2023-11-02T11:10:02+00:00', 'value': 92.287, 'unit': 'm3'},
'BELGIUM_MBUS2_DEVICE_TYPE': {'value': 7, 'unit': None},
'BELGIUM_MBUS2_EQUIPMENT_IDENTIFIER': {'value': '3853455430303030393631313733', 'unit': None},
'BELGIUM_MBUS2_METER_READING1': {'datetime': '2023-11-02T11:15:32+00:00', 'value': 8.579, 'unit': 'm3'}}
'MBUS_METER_READING': {'datetime': '2023-11-02T11:15:32+00:00',
'value': 8.579, 'unit': 'm3'},
'CHANNEL_ID': 2}]}
)
def test_to_str(self):
@@ -399,21 +400,14 @@ class TelegramParserFluviusTest(unittest.TestCase):
'ACTUAL_TRESHOLD_ELECTRICITY: 999.9 [kW]\n'
'FUSE_THRESHOLD_L1: 999 [A]\n'
'TEXT_MESSAGE: None [None]\n'
'BELGIUM_MBUS1_DEVICE_TYPE: 3 [None]\n'
'MBUS DEVICE (channel 1)\n'
' BELGIUM_MBUS1_DEVICE_TYPE: 3 [None]\n'
' BELGIUM_MBUS1_EQUIPMENT_IDENTIFIER: 37464C4F32313233303838303237 [None]\n'
' BELGIUM_MBUS1_VALVE_POSITION: 1 [None]\n'
' BELGIUM_MBUS1_METER_READING2: 92.287 [m3] at 2023-11-02T11:10:02+00:00\n'
' MBUS_DEVICE_TYPE: 3 [None]\n'
' MBUS_EQUIPMENT_IDENTIFIER: 37464C4F32313233303838303237 [None]\n'
' MBUS_VALVE_POSITION: 1 [None]\n'
' MBUS_METER_READING: 92.287 [m3] at 2023-11-02T11:10:02+00:00\n'
'MBUS DEVICE (channel 2)\n'
' BELGIUM_MBUS2_DEVICE_TYPE: 7 [None]\n'
' BELGIUM_MBUS2_EQUIPMENT_IDENTIFIER: 3853455430303030393631313733 [None]\n'
' BELGIUM_MBUS2_METER_READING1: 8.579 [m3] at 2023-11-02T11:15:32+00:00\n'
'BELGIUM_MBUS1_EQUIPMENT_IDENTIFIER: 37464C4F32313233303838303237 [None]\n'
'BELGIUM_MBUS1_VALVE_POSITION: 1 [None]\n'
'BELGIUM_MBUS1_METER_READING2: 92.287 [m3] at 2023-11-02T11:10:02+00:00\n'
'BELGIUM_MBUS2_DEVICE_TYPE: 7 [None]\n'
'BELGIUM_MBUS2_EQUIPMENT_IDENTIFIER: 3853455430303030393631313733 [None]\n'
'BELGIUM_MBUS2_METER_READING1: 8.579 [m3] at 2023-11-02T11:15:32+00:00\n'
' MBUS_DEVICE_TYPE: 7 [None]\n'
' MBUS_EQUIPMENT_IDENTIFIER: 3853455430303030393631313733 [None]\n'
' MBUS_METER_READING: 8.579 [m3] at 2023-11-02T11:15:32+00:00\n'
)
)
test/test_parse_v5.py
+24 -27
View File
@@ -172,12 +172,6 @@ class TelegramParserV5Test(unittest.TestCase):
assert telegram.TEXT_MESSAGE.unit is None
assert telegram.TEXT_MESSAGE.value is None
# DEVICE_TYPE (0-x:24.1.0)
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(telegram.INSTANTANEOUS_ACTIVE_POWER_L1_POSITIVE, CosemObject)
assert telegram.INSTANTANEOUS_ACTIVE_POWER_L1_POSITIVE.unit == 'kW'
@@ -215,27 +209,27 @@ class TelegramParserV5Test(unittest.TestCase):
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
# use get_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(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
# MBUS_DEVICE_TYPE (0-1:96.1.0)
assert isinstance(gas_meter_device.MBUS_DEVICE_TYPE, CosemObject)
assert gas_meter_device.MBUS_DEVICE_TYPE.unit is None
assert isinstance(gas_meter_device.MBUS_DEVICE_TYPE.value, int)
assert gas_meter_device.MBUS_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'
# MBUS_EQUIPMENT_IDENTIFIER (0-1:96.1.0)
assert isinstance(gas_meter_device.MBUS_EQUIPMENT_IDENTIFIER, CosemObject)
assert gas_meter_device.MBUS_EQUIPMENT_IDENTIFIER.unit is None
assert isinstance(gas_meter_device.MBUS_EQUIPMENT_IDENTIFIER.value, str)
assert gas_meter_device.MBUS_EQUIPMENT_IDENTIFIER.value == '3232323241424344313233343536373839'
# HOURLY_GAS_METER_READING (0-1:24.2.1)
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')
# MBUS_METER_READING (0-1:24.2.1)
assert isinstance(gas_meter_device.MBUS_METER_READING, MBusObject)
assert gas_meter_device.MBUS_METER_READING.unit == 'm3'
assert isinstance(telegram.MBUS_METER_READING.value, Decimal)
assert gas_meter_device.MBUS_METER_READING.value == Decimal('0.107')
def test_checksum_valid(self):
# No exception is raised.
@@ -270,8 +264,11 @@ class TelegramParserV5Test(unittest.TestCase):
parser = TelegramParser(telegram_specifications.V5)
telegram = parser.parse(invalid_date_telegram)
# HOURLY_GAS_METER_READING (0-1:24.2.1)
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')
# MBUS DEVICE 1
mbus1 = telegram.get_mbus_device_by_channel(1)
# MBUS_METER_READING (0-1:24.2.1)
assert isinstance(mbus1.MBUS_METER_READING, MBusObject)
assert mbus1.MBUS_METER_READING.unit is None
assert isinstance(mbus1.MBUS_METER_READING.value, Decimal)
assert mbus1.MBUS_METER_READING.value == Decimal('0.000')