updated version number

Pass lines to parser including line endings

CHANGELOG.rst

@@ -1,6 +1,11 @@
 Change Log
 ----------
 
+**0.6** (2017-01-04)
+
+- Fixed bug in CRC checksum verification for the asyncio client (`pull request #15 <https://github.com/ndokter/dsmr_parser/pull/15>`_)
+- Support added for TCP connections using the asyncio client (`pull request #12 <https://github.com/ndokter/dsmr_parser/pull/12/>`_)
+
 **0.5** (2016-12-29)
 
 - CRC checksum verification for DSMR v4 telegrams (`issue #10 <https://github.com/ndokter/dsmr_parser/issues/10>`_)

dsmr_parser/__main__.py

@@ -1,8 +1,9 @@
 import argparse
 import asyncio
 import logging
+from functools import partial
 
-from .protocol import create_dsmr_reader
+from .protocol import create_dsmr_reader, create_tcp_dsmr_reader
 
 
 def console():
@@ -11,6 +12,10 @@ def console():
     parser = argparse.ArgumentParser(description=console.__doc__)
     parser.add_argument('--device', default='/dev/ttyUSB0',
                         help='port to read DSMR data from')
+    parser.add_argument('--host', default=None,
+                        help='alternatively connect using TCP host.')
+    parser.add_argument('--port', default=None,
+                        help='TCP port to use for connection')
     parser.add_argument('--version', default='2.2', choices=['2.2', '4'],
                         help='DSMR version (2.2, 4)')
     parser.add_argument('--verbose', '-v', action='count')
@@ -32,7 +37,29 @@ def console():
                 print(obj.value, obj.unit)
         print()
 
-    conn = create_dsmr_reader(args.device, args.version, print_callback, loop=loop)
+    # create tcp or serial connection depending on args
+    if args.host and args.port:
+        create_connection = partial(create_tcp_dsmr_reader,
+                                    args.host, args.port, args.version,
+                                    print_callback, loop=loop)
+    else:
+        create_connection = partial(create_dsmr_reader,
+                                    args.device, args.version,
+                                    print_callback, loop=loop)
 
-    loop.create_task(conn)
-    loop.run_forever()
+    try:
+        # connect and keep connected until interrupted by ctrl-c
+        while True:
+            # create serial or tcp connection
+            conn = create_connection()
+            transport, protocol = loop.run_until_complete(conn)
+            # wait until connection it closed
+            loop.run_until_complete(protocol.wait_closed())
+            # wait 5 seconds before attempting reconnect
+            loop.run_until_complete(asyncio.sleep(5))
+    except KeyboardInterrupt:
+        # cleanup connection after user initiated shutdown
+        transport.close()
+        loop.run_until_complete(asyncio.sleep(0))
+    finally:
+        loop.close()

dsmr_parser/protocol.py

@@ -8,19 +8,13 @@ from serial_asyncio import create_serial_connection
 
 from . import telegram_specifications
 from .exceptions import ParseError
-from .parsers import (
-    TelegramParserV2_2,
-    TelegramParserV4
-)
-from .serial import (
-    SERIAL_SETTINGS_V2_2, SERIAL_SETTINGS_V4,
-    is_end_of_telegram,
-    is_start_of_telegram
-)
+from .parsers import TelegramParserV2_2, TelegramParserV4
+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."""
+def create_dsmr_protocol(dsmr_version, telegram_callback, loop=None):
+    """Creates a DSMR asyncio protocol."""
 
     if dsmr_version == '2.2':
         specifications = telegram_specifications.V2_2
@@ -31,13 +25,28 @@ def create_dsmr_reader(port, dsmr_version, telegram_callback, loop=None):
         telegram_parser = TelegramParserV4
         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 protocol, serial_settings
 
+
+def create_dsmr_reader(port, dsmr_version, telegram_callback, loop=None):
+    """Creates a DSMR asyncio protocol coroutine using serial port."""
+    protocol, serial_settings = create_dsmr_protocol(
+        dsmr_version, telegram_callback, loop=None)
+    serial_settings['url'] = port
+
+    conn = create_serial_connection(loop, protocol, **serial_settings)
+    return conn
+
+
+def create_tcp_dsmr_reader(host, port, dsmr_version,
+                           telegram_callback, loop=None):
+    """Creates a DSMR asyncio protocol coroutine using TCP connection."""
+    protocol, _ = create_dsmr_protocol(
+        dsmr_version, telegram_callback, loop=None)
+    conn = loop.create_connection(protocol, host, port)
     return conn
 
 
@@ -58,6 +67,8 @@ class DSMRProtocol(asyncio.Protocol):
         self.telegram = []
         # buffer to keep incomplete incoming data
         self.buffer = ''
+        # keep a lock until the connection is closed
+        self._closed = asyncio.Event()
 
     def connection_made(self, transport):
         """Just logging for now."""
@@ -73,9 +84,11 @@ class DSMRProtocol(asyncio.Protocol):
 
     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)
+        crlf = "\r\n"
+        while crlf in self.buffer:
+            line, self.buffer = self.buffer.split(crlf, 1)
             self.log.debug('got line: %s', line)
+            line += crlf  # add the trailing crlf again
 
             # Telegrams need to be complete because the values belong to a
             # particular reading and can also be related to eachother.
@@ -93,7 +106,11 @@ class DSMRProtocol(asyncio.Protocol):
 
     def connection_lost(self, exc):
         """Stop when connection is lost."""
-        self.log.error('disconnected')
+        if exc:
+            self.log.exception('disconnected due to exception')
+        else:
+            self.log.info('disconnected because of close/abort.')
+        self._closed.set()
 
     def handle_telegram(self, telegram):
         """Send off parsed telegram to handling callback."""
@@ -101,3 +118,8 @@ class DSMRProtocol(asyncio.Protocol):
 
         if self.telegram_callback:
             self.telegram_callback(telegram)
+
+    @asyncio.coroutine
+    def wait_closed(self):
+        """Wait until connection is closed."""
+        yield from self._closed.wait()

setup.py

@@ -6,7 +6,7 @@ setup(
     author='Nigel Dokter',
     author_email='nigeldokter@gmail.com',
     url='https://github.com/ndokter/dsmr_parser',
-    version='0.5',
+    version='0.6',
     packages=find_packages(),
     install_requires=[
         'pyserial>=3,<4',

test/test_protocol.py

@@ -1,9 +1,7 @@
 """Test DSMR serial protocol."""
-
+import unittest
 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
@@ -34,29 +32,27 @@ TELEGRAM_V2_2 = [
 ]
 
 
-@pytest.fixture
-def protocol():
-    """DSMRprotocol instance with mocked telegram_callback."""
+class ProtocolTest(unittest.TestCase):
 
-    parser = TelegramParserV2_2
-    specification = telegram_specifications.V2_2
+    def setUp(self):
+        parser = TelegramParserV2_2
+        specification = telegram_specifications.V2_2
 
-    telegram_parser = parser(specification)
-    return DSMRProtocol(None, telegram_parser,
-                        telegram_callback=Mock())
+        telegram_parser = parser(specification)
+        self.protocol = DSMRProtocol(None, telegram_parser,
+                                     telegram_callback=Mock())
 
+    def test_complete_packet(self):
+        """Protocol should assemble incoming lines into complete packet."""
 
-def test_complete_packet(protocol):
-    """Protocol should assemble incoming lines into complete packet."""
+        for line in TELEGRAM_V2_2:
+            self.protocol.data_received(bytes(line + '\r\n', 'ascii'))
 
-    for line in TELEGRAM_V2_2:
-        protocol.data_received(bytes(line + '\r\n', 'ascii'))
+        telegram = self.protocol.telegram_callback.call_args_list[0][0][0]
+        assert isinstance(telegram, dict)
 
-    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.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'
+        assert float(telegram[obis.GAS_METER_READING].value) == 1.001
+        assert telegram[obis.GAS_METER_READING].unit == 'm3'