ProfileGeneric parser working, TODO complete ProfileGenericObject + Test

Browse Source
This commit is contained in:
Hans Erik van Elburg 2020-05-16 16:31:26 +02:00
parent a0ce89054a
commit b6278a8991
5 changed files with 70 additions and 40 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
dsmr_parser/objects.py
View File

@ -74,7 +74,7 @@ class MBusObject(DSMRObject):
# TODO object, but let the parse set them differently? So don't use
# TODO hardcoded indexes here.
if len(self.values) != 2: # v2
return self.values[5]['value']
return self.values[6]['value']
else:
return self.values[1]['value']
@ -84,7 +84,7 @@ class MBusObject(DSMRObject):
# TODO object, but let the parse set them differently? So don't use
# TODO hardcoded indexes here.
if len(self.values) != 2: # v2
return self.values[4]['value']
return self.values[5]['value']
else:
return self.values[1]['unit']
@ -111,5 +111,7 @@ class CosemObject(DSMRObject):
return output
class ProfileGeneric(DSMRObject):
pass # TODO implement
class ProfileGenericObject(DSMRObject):
def __str__(self):
output = "{}".format(self.values)
return output

60
dsmr_parser/parsers.py
View File

@ -3,7 +3,7 @@ import re
from ctypes import c_ushort
from dsmr_parser.objects import MBusObject, CosemObject
from dsmr_parser.objects import MBusObject, CosemObject, ProfileGenericObject
from dsmr_parser.exceptions import ParseError, InvalidChecksumError
logger = logging.getLogger(__name__)
@ -123,19 +123,28 @@ class DSMRObjectParser(object):
def __init__(self, *value_formats):
self.value_formats = value_formats
def _is_line_wellformed(self, line, values):
# allows overriding by child class
return (values and (len(values) == len(self.value_formats)))
def _parse_values(self, values):
# allows overriding by child class
return [self.value_formats[i].parse(value)
for i, value in enumerate(values)]
def _parse(self, line):
# Match value groups, but exclude the parentheses
pattern = re.compile(r'((?<=\()[0-9a-zA-Z\.\*]{0,}(?=\)))+')
pattern = re.compile(r'((?<=\()[0-9a-zA-Z\.\*\-\:]{0,}(?=\)))')
values = re.findall(pattern, line)
if not self._is_line_wellformed(line, values):
raise ParseError("Invalid '%s' line for '%s'", line, self)
# Convert empty value groups to None for clarity.
values = [None if value == '' else value for value in values]
if not values or len(values) != len(self.value_formats):
raise ParseError("Invalid '%s' line for '%s'", line, self)
return [self.value_formats[i].parse(value)
for i, value in enumerate(values)]
return self._parse_values(values)
class MBusParser(DSMRObjectParser):
@ -183,7 +192,7 @@ class CosemParser(DSMRObjectParser):
return CosemObject(self._parse(line))
class ProfileGenericParser(object):
class ProfileGenericParser(DSMRObjectParser):
"""
Power failure log parser.
@ -204,25 +213,34 @@ class ProfileGenericParser(object):
8) Buffer value 2 (oldest entry of buffer attribute without unit)
9) Unit of buffer values (Unit of capture objects attribute)
"""
def __init__(self, buffer_types, head_parsers, parsers_for_unidentified):
self.value_formats = head_parsers
self.buffer_types = buffer_types
self.parsers_for_unidentified = parsers_for_unidentified
def _parse(self, line):
# Match value groups, but exclude the parentheses. Adapted to also match OBIS code in 3rd position.
pattern = re.compile(r'((?<=\()[0-9a-zA-Z\.\*\-\:]{0,}(?=\)))')
values = re.findall(pattern, line)
# Convert empty value groups to None for clarity.
values = [None if value == '' else value for value in values]
def _is_line_wellformed(self, line, values):
if values and (len(values) >= 2) and (values[0].isdigit()):
buffer_length = int(values[0])
return (buffer_length <= 10) and (len(values) == (buffer_length * 2 + 2))
else:
return False
if (not values) or (len(values) != (buffer_length * 2 + 2)):
raise ParseError("Invalid '%s' line for '%s'", line, self)
def _parse_values(self, values):
buffer_length = int(values[0])
buffer_value_obis_ID = values[1]
if (buffer_length > 0):
if buffer_value_obis_ID in self.buffer_types:
bufferValueParsers = self.buffer_types[buffer_value_obis_ID]
else:
bufferValueParsers = self.parsers_for_unidentified
# add the parsers for the encountered value type z times
for _ in range(buffer_length):
self.value_formats.extend(bufferValueParsers)
return [self.value_formats[i].parse(value)
for i, value in enumerate(values)]
return [self.value_formats[i].parse(value) for i, value in enumerate(values)]
def parse(self, line):
raise NotImplementedError()
return ProfileGenericObject(self._parse(line))
class ValueParser(object):

16
dsmr_parser/profile_generic_specifications.py
View File

@ -1,14 +1,10 @@
from dsmr_parser.parsers import ValueParser, MBusParser
from dsmr_parser.parsers import ValueParser
from dsmr_parser.value_types import timestamp
FAILURE_EVENT = r'0-0\:96\.7\.19'
V4 = {
'objects': {
FAILURE_EVENT: MBusParser(
ValueParser(timestamp),
ValueParser(int)
)
}
PG_FAILURE_EVENT = r'0-0:96.7.19'
PG_HEAD_PARSERS = [ValueParser(int), ValueParser(str)]
PG_UNIDENTIFIED_BUFFERTYPE_PARSERS = [ValueParser(str), ValueParser(str)]
BUFFER_TYPES = {
PG_FAILURE_EVENT: [ValueParser(timestamp), ValueParser(int)]
}

14
dsmr_parser/telegram_specifications.py
View File

@ -2,9 +2,9 @@ from decimal import Decimal
from copy import deepcopy
from dsmr_parser import obis_references as obis
from dsmr_parser.parsers import CosemParser, ValueParser, MBusParser
from dsmr_parser.parsers import CosemParser, ValueParser, MBusParser, ProfileGenericParser
from dsmr_parser.value_types import timestamp
from dsmr_parser.profile_generic_specifications import BUFFER_TYPES, PG_HEAD_PARSERS, PG_UNIDENTIFIED_BUFFERTYPE_PARSERS
"""
dsmr_parser.telegram_specifications
@ -37,8 +37,9 @@ V2_2 = {
ValueParser(int),
ValueParser(int),
ValueParser(int),
ValueParser(str),
ValueParser(Decimal),
ValueParser(str), # obis ref
ValueParser(str), # unit, position 5
ValueParser(Decimal), # meter reading, position 6
),
}
}
@ -60,7 +61,10 @@ V4 = {
obis.CURRENT_ELECTRICITY_DELIVERY: CosemParser(ValueParser(Decimal)),
obis.SHORT_POWER_FAILURE_COUNT: CosemParser(ValueParser(int)),
obis.LONG_POWER_FAILURE_COUNT: CosemParser(ValueParser(int)),
# POWER_EVENT_FAILURE_LOG: ProfileGenericParser(), TODO
obis.POWER_EVENT_FAILURE_LOG:
ProfileGenericParser(BUFFER_TYPES,
PG_HEAD_PARSERS,
PG_UNIDENTIFIED_BUFFERTYPE_PARSERS),
obis.VOLTAGE_SAG_L1_COUNT: CosemParser(ValueParser(int)),
obis.VOLTAGE_SAG_L2_COUNT: CosemParser(ValueParser(int)),
obis.VOLTAGE_SAG_L3_COUNT: CosemParser(ValueParser(int)),

10
test/test_telegram.py
View File

@ -7,6 +7,7 @@ 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 decimal import Decimal
@ -286,6 +287,15 @@ class TelegramTest(unittest.TestCase):
unit_val='m3',
value_type=Decimal,
value_val=Decimal('981.443'))
# POWER_EVENT_FAILURE_LOG (1-0:99.97.0)
testitem_name = 'POWER_EVENT_FAILURE_LOG'
object_type = ProfileGenericObject
testitem = eval("telegram.{}".format(testitem_name))
assert isinstance(testitem, object_type)
# assert testitem.unit == unit_val
# assert isinstance(testitem.value, value_type)
# assert testitem.value == value_val
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