cleaned up based on pylama complaints / pinpointed to coverage version 4.5.4 as next version is incompatible

Browse Source
This commit is contained in:
Hans Erik van Elburg 2020-02-02 22:12:25 +01:00
parent fee3f696c4
commit b6537678a7
8 changed files with 67 additions and 75 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
dsmr_parser/clients/filereader.py
View File

@ -9,6 +9,7 @@ from dsmr_parser.parsers import TelegramParser
logger = logging.getLogger(__name__)
class FileReader(object):
"""
Filereader to read and parse raw telegram strings from a file and instantiate Telegram objects
@ -60,7 +61,7 @@ class FileReader(object):
Read complete DSMR telegram's from a file and return a Telegram object.
:rtype: generator
"""
with open(self._file,"rb") as file_handle:
with open(self._file, "rb") as file_handle:
while True:
data = file_handle.readline()
str = data.decode()
@ -74,6 +75,7 @@ class FileReader(object):
except ParseError as e:
logger.error('Failed to parse telegram: %s', e)
class FileInputReader(object):
"""
Filereader to read and parse raw telegram strings from stdin or files specified at the commandline
@ -155,7 +157,7 @@ class FileTailReader(object):
Read complete DSMR telegram's from a files tail and return a Telegram object.
:rtype: generator
"""
with open(self._file,"rb") as file_handle:
with open(self._file, "rb") as file_handle:
for data in tailer.follow(file_handle):
str = data.decode()
self.telegram_buffer.append(str)
@ -163,8 +165,7 @@ class FileTailReader(object):
for telegram in self.telegram_buffer.get_all():
try:
yield Telegram(telegram, self.telegram_parser, self.telegram_specification)
except dsmr_parser.exceptions.InvalidChecksumError as e:
except InvalidChecksumError as e:
logger.warning(str(e))
except dsmr_parser.exceptions.ParseError as e:
except ParseError as e:
logger.error('Failed to parse telegram: %s', e)

1
dsmr_parser/clients/protocol.py
View File

@ -59,6 +59,7 @@ def create_tcp_dsmr_reader(host, port, dsmr_version,
conn = loop.create_connection(protocol, host, port)
return conn
class DSMRProtocol(asyncio.Protocol):
"""Assemble and handle incoming data into complete DSM telegrams."""

80
dsmr_parser/obis_name_mapping.py
View File

@ -10,45 +10,45 @@ 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_USED_TARIFF_1 : 'ELECTRICITY_USED_TARIFF_1',
obis.ELECTRICITY_USED_TARIFF_2 : 'ELECTRICITY_USED_TARIFF_2',
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.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.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.ELECTRICITY_IMPORTED_TOTAL: 'ELECTRICITY_IMPORTED_TOTAL',
obis.ELECTRICITY_USED_TARIFF_1: 'ELECTRICITY_USED_TARIFF_1',
obis.ELECTRICITY_USED_TARIFF_2: 'ELECTRICITY_USED_TARIFF_2',
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.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.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'
}
REVERSE_EN = dict([ (v,k) for k,v in EN.items()])
REVERSE_EN = dict([(v, k) for k, v in EN.items()])

3
dsmr_parser/objects.py
View File

@ -1,5 +1,6 @@
import dsmr_parser.obis_name_mapping
class Telegram(object):
"""
Container for raw and parsed telegram data.
@ -47,7 +48,7 @@ class Telegram(object):
def __str__(self):
output = ""
for attr, value in self:
output += "{}: \t {} \t[{}]\n".format(attr,str(value.value),str(value.unit))
output += "{}: \t {} \t[{}]\n".format(attr, str(value.value), str(value.unit))
return output

2
dsmr_parser/parsers.py
View File

@ -3,7 +3,7 @@ import re
from ctypes import c_ushort
from dsmr_parser.objects import MBusObject, CosemObject, Telegram
from dsmr_parser.objects import MBusObject, CosemObject
from dsmr_parser.exceptions import ParseError, InvalidChecksumError
logger = logging.getLogger(__name__)

10
test/experiment_telegram.py
View File

@ -1,14 +1,8 @@
from decimal import Decimal
import datetime
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, Telegram
from dsmr_parser.objects import Telegram
from dsmr_parser.parsers import TelegramParser
from example_telegrams import TELEGRAM_V4_2
parser = TelegramParser(telegram_specifications.V4)
telegram = Telegram(TELEGRAM_V4_2, parser, telegram_specifications.V4)
print(telegram)
print(telegram)

23
test/test_telegram.py
View File

@ -1,30 +1,21 @@
from decimal import Decimal
import datetime
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, Telegram
from dsmr_parser.objects import CosemObject
from dsmr_parser.objects import Telegram
from dsmr_parser.parsers import TelegramParser
from test.example_telegrams import TELEGRAM_V4_2
class TelegramTest(unittest.TestCase):
""" Test instantiation of Telegram object """
def test_instantiate(self):
parser = TelegramParser(telegram_specifications.V4)
#result = parser.parse(TELEGRAM_V4_2)
telegram = Telegram(TELEGRAM_V4_2, parser, telegram_specifications.V4)
# 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'
testitem = telegram.P1_MESSAGE_HEADER
assert isinstance(testitem, CosemObject)
assert testitem.unit is None
assert testitem.value == '42'

12
tox.ini
View File

@ -1,5 +1,6 @@
[tox]
envlist = py34,py35,p36
envlist = py34,py35,py36,py37
requires = coverage<=4.5.4
[testenv]
deps=
@ -14,10 +15,13 @@ commands=
pylama dsmr_parser test
[pylama:dsmr_parser/clients/__init__.py]
ignore = W0611,W0605
ignore = W0611
[pylama:dsmr_parser/parsers.py]
ignore = W605
[pylama:pylint]
max_line_length = 100
max_line_length = 120
[pylama:pycodestyle]
max_line_length = 100
max_line_length = 120