Async protocol

dsmr_parser/__main__.py

@@ -1,6 +1,8 @@
 import argparse
-from dsmr_parser.serial import SERIAL_SETTINGS_V2_2, SERIAL_SETTINGS_V4, SerialReader
-from dsmr_parser import telegram_specifications
+import asyncio
+import logging
+
+from .protocol import create_dsmr_reader
 
 
 def console():
@@ -11,22 +13,26 @@ def console():
                         help='port to read DSMR data from')
     parser.add_argument('--version', default='2.2', choices=['2.2', '4'],
                         help='DSMR version (2.2, 4)')
+    parser.add_argument('--verbose', '-v', action='count')
 
     args = parser.parse_args()
 
-    settings = {
-        '2.2': (SERIAL_SETTINGS_V2_2, telegram_specifications.V2_2),
-        '4': (SERIAL_SETTINGS_V4, telegram_specifications.V4),
-    }
+    if args.verbose:
+        level = logging.DEBUG
+    else:
+        level = logging.ERROR
+    logging.basicConfig(level=level)
 
-    serial_reader = SerialReader(
-        device=args.device,
-        serial_settings=settings[args.version][0],
-        telegram_specification=settings[args.version][1],
-    )
+    loop = asyncio.get_event_loop()
 
-    for telegram in serial_reader.read():
+    def print_callback(telegram):
+        """Callback that prints telegram values."""
         for obiref, obj in telegram.items():
             if obj:
                 print(obj.value, obj.unit)
         print()
+
+    conn = create_dsmr_reader(args.device, args.version, print_callback, loop=loop)
+
+    loop.create_task(conn)
+    loop.run_forever()

dsmr_parser/protocol.py

@@ -0,0 +1,97 @@
+"""Asyncio protocol implementation for handling telegrams."""
+
+import asyncio
+import logging
+from functools import partial
+
+from serial_asyncio import create_serial_connection
+
+from . import telegram_specifications
+from .exceptions import ParseError
+from .parsers import TelegramParser, TelegramParserV2_2
+from .serial import (SERIAL_SETTINGS_V2_2, SERIAL_SETTINGS_V4,
+                     is_end_of_telegram, is_start_of_telegram)
+
+
+def create_dsmr_reader(port, dsmr_version, telegram_callback, loop=None):
+    """Creates a DSMR asyncio protocol coroutine."""
+
+    if dsmr_version == '2.2':
+        specifications = telegram_specifications.V2_2
+        telegram_parser = TelegramParserV2_2
+        serial_settings = SERIAL_SETTINGS_V2_2
+    elif dsmr_version == '4':
+        specifications = telegram_specifications.V4
+        telegram_parser = TelegramParser
+        serial_settings = SERIAL_SETTINGS_V4
+
+    serial_settings['url'] = port
+
+    protocol = partial(DSMRProtocol, loop, telegram_parser(specifications),
+                       telegram_callback=telegram_callback)
+
+    conn = create_serial_connection(loop, protocol, **serial_settings)
+
+    return conn
+
+
+class DSMRProtocol(asyncio.Protocol):
+    """Assemble and handle incoming data into complete DSM telegrams."""
+
+    transport = None
+    telegram_callback = None
+
+    def __init__(self, loop, telegram_parser, telegram_callback=None):
+        """Initialize class."""
+        self.loop = loop
+        self.log = logging.getLogger(__name__)
+        self.telegram_parser = telegram_parser
+        # callback to call on complete telegram
+        self.telegram_callback = telegram_callback
+        # buffer to keep incoming telegram lines
+        self.telegram = []
+        # buffer to keep incomplete incoming data
+        self.buffer = ''
+
+    def connection_made(self, transport):
+        """Just logging for now."""
+        self.transport = transport
+        self.log.debug('connected')
+
+    def data_received(self, data):
+        """Add incoming data to buffer."""
+        data = data.decode()
+        self.log.debug('received data: %s', data.strip())
+        self.buffer += data
+        self.handle_lines()
+
+    def handle_lines(self):
+        """Assemble incoming data into single lines."""
+        while "\r\n" in self.buffer:
+            line, self.buffer = self.buffer.split("\r\n", 1)
+            self.log.debug('got line: %s', line)
+
+            # Telegrams need to be complete because the values belong to a
+            # particular reading and can also be related to eachother.
+            if not self.telegram and not is_start_of_telegram(line):
+                continue
+
+            self.telegram.append(line)
+            if is_end_of_telegram(line):
+                try:
+                    parsed_telegram = self.telegram_parser.parse(self.telegram)
+                    self.handle_telegram(parsed_telegram)
+                except ParseError:
+                    self.log.exception("failed to parse telegram")
+                self.telegram = []
+
+    def connection_lost(self, exc):
+        """Stop when connection is lost."""
+        self.log.error('disconnected')
+
+    def handle_telegram(self, telegram):
+        """Send off parsed telegram to handling callback."""
+        self.log.debug('got telegram: %s', telegram)
+
+        if self.telegram_callback:
+            self.telegram_callback(telegram)

dsmr_parser/serial.py

@@ -1,8 +1,15 @@
-import serial
 import asyncio
+import logging
+
+import serial
+
 import serial_asyncio
+from dsmr_parser.exceptions import ParseError
 from dsmr_parser.parsers import TelegramParser, TelegramParserV2_2
 
+logger = logging.getLogger(__name__)
+
+
 SERIAL_SETTINGS_V2_2 = {
     'baudrate': 9600,
     'bytesize': serial.SEVENBITS,
@@ -44,7 +51,6 @@
 }
 
 
-
 def is_start_of_telegram(line):
     return line.startswith('/')
 
@@ -129,6 +135,11 @@ def read(self, queue):
             telegram.append(line)
 
             if is_end_of_telegram(line):
-                # push new parsed telegram onto queue
-                queue.put_nowait(self.telegram_parser.parse(telegram))
+                try:
+                    parsed_telegram = self.telegram_parser.parse(telegram)
+                    # push new parsed telegram onto queue
+                    queue.put_nowait(parsed_telegram)
+                except ParseError:
+                    logger.exception("failed to parse telegram")
+
                 telegram = []

test/test_parse_v4_2.py

@@ -1,13 +1,13 @@
 """Test parsing of a DSMR v4.2 telegram."""
 import datetime
-
 from decimal import Decimal
+
 import pytz
 
+from dsmr_parser import obis_references as obis
+from dsmr_parser import telegram_specifications
 from dsmr_parser.objects import CosemObject, MBusObject
 from dsmr_parser.parsers import TelegramParser
-from dsmr_parser import telegram_specifications
-from dsmr_parser import obis_references as obis
 
 TELEGRAM_V4_2 = [
     '1-3:0.2.8(42)',
@@ -22,7 +22,8 @@
     '1-0:2.7.0(00.000*kW)',
     '0-0:96.7.21(00015)',
     '0-0:96.7.9(00007)',
-    '1-0:99.97.0(3)(0-0:96.7.19)(000103180420W)(0000237126*s)(000101000001W)(2147483647*s)(000101000001W)(2147483647*s)',
+    ('1-0:99.97.0(3)(0-0:96.7.19)(000103180420W)(0000237126*s)'
+     '(000101000001W)(2147483647*s)(000101000001W)(2147483647*s)'),
     '1-0:32.32.0(00000)',
     '1-0:52.32.0(00000)',
     '1-0:72.32.0(00000)',

test/test_protocol.py

@@ -0,0 +1,39 @@
+"""Test DSMR serial protocol."""
+
+from unittest.mock import Mock
+
+import pytest
+from dsmr_parser import obis_references as obis
+from dsmr_parser import telegram_specifications
+from dsmr_parser.parsers import TelegramParserV2_2
+from dsmr_parser.protocol import DSMRProtocol
+
+from .test_parse_v2_2 import TELEGRAM_V2_2
+
+
+@pytest.fixture
+def protocol():
+    """DSMRprotocol instance with mocked telegram_callback."""
+
+    parser = TelegramParserV2_2
+    specification = telegram_specifications.V2_2
+
+    telegram_parser = parser(specification)
+    return DSMRProtocol(None, telegram_parser,
+                        telegram_callback=Mock())
+
+
+def test_complete_packet(protocol):
+    """Protocol should assemble incoming lines into complete packet."""
+
+    for line in TELEGRAM_V2_2:
+        protocol.data_received(bytes(line + '\r\n', 'ascii'))
+
+    telegram = protocol.telegram_callback.call_args_list[0][0][0]
+    assert isinstance(telegram, dict)
+
+    assert float(telegram[obis.CURRENT_ELECTRICITY_USAGE].value) == 1.01
+    assert telegram[obis.CURRENT_ELECTRICITY_USAGE].unit == 'kW'
+
+    assert float(telegram[obis.GAS_METER_READING].value) == 1.001
+    assert telegram[obis.GAS_METER_READING].unit == 'm3'

tox.ini

@@ -5,6 +5,9 @@ envlist = py34,py35
 deps=
   pytest
   pylama
+  pytest-asyncio
+  pytest-catchlog
+  pytest-mock
 commands=
   py.test test {posargs}
   pylama dsmr_parser test