Preparing release 0.28

Optional keep alive monitoring for TCP/IP connections

CHANGELOG.rst

@@ -1,6 +1,10 @@
 Change Log
 ----------
 
+**0.28** (2021-02-21)
+- Optional keep alive monitoring for TCP/IP connections (`pull request #73 <https://github.com/ndokter/dsmr_parser/pull/73>`_).
+- Catch parse errors in TelegramParser, ignore lines that can not be parsed (`pull request #74 <https://github.com/ndokter/dsmr_parser/pull/74>`_).
+
 **0.27** (2020-12-24)
 - fix for empty parentheses in ProfileGenericParser (redone) (`pull request #69 <https://github.com/ndokter/dsmr_parser/pull/69>`_).
 

dsmr_parser/__main__.py

@@ -16,8 +16,8 @@ def console():
                         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('--version', default='2.2', choices=['2.2', '4', '5', '5B', '5L'],
+                        help='DSMR version (2.2, 4, 5, 5B, 5L)')
     parser.add_argument('--verbose', '-v', action='count')
 
     args = parser.parse_args()

dsmr_parser/clients/protocol.py

@@ -14,7 +14,7 @@ from dsmr_parser.clients.settings import SERIAL_SETTINGS_V2_2, \
     SERIAL_SETTINGS_V4, SERIAL_SETTINGS_V5
 
 
-def create_dsmr_protocol(dsmr_version, telegram_callback, loop=None):
+def create_dsmr_protocol(dsmr_version, telegram_callback, loop=None, **kwargs):
     """Creates a DSMR asyncio protocol."""
 
     if dsmr_version == '2.2':
@@ -37,7 +37,7 @@ def create_dsmr_protocol(dsmr_version, telegram_callback, loop=None):
                                   dsmr_version)
 
     protocol = partial(DSMRProtocol, loop, TelegramParser(specification),
-                       telegram_callback=telegram_callback)
+                       telegram_callback=telegram_callback, **kwargs)
 
     return protocol, serial_settings
 
@@ -53,12 +53,14 @@ def create_dsmr_reader(port, dsmr_version, telegram_callback, loop=None):
 
 
 def create_tcp_dsmr_reader(host, port, dsmr_version,
-                           telegram_callback, loop=None):
+                           telegram_callback, loop=None,
+                           keep_alive_interval=None):
     """Creates a DSMR asyncio protocol coroutine using TCP connection."""
     if not loop:
         loop = asyncio.get_event_loop()
     protocol, _ = create_dsmr_protocol(
-        dsmr_version, telegram_callback, loop=loop)
+        dsmr_version, telegram_callback, loop=loop,
+        keep_alive_interval=keep_alive_interval)
     conn = loop.create_connection(protocol, host, port)
     return conn
 
@@ -69,7 +71,8 @@ class DSMRProtocol(asyncio.Protocol):
     transport = None
     telegram_callback = None
 
-    def __init__(self, loop, telegram_parser, telegram_callback=None):
+    def __init__(self, loop, telegram_parser,
+                 telegram_callback=None, keep_alive_interval=None):
         """Initialize class."""
         self.loop = loop
         self.log = logging.getLogger(__name__)
@@ -80,21 +83,38 @@ class DSMRProtocol(asyncio.Protocol):
         self.telegram_buffer = TelegramBuffer()
         # keep a lock until the connection is closed
         self._closed = asyncio.Event()
+        self._keep_alive_interval = keep_alive_interval
+        self._active = True
 
     def connection_made(self, transport):
         """Just logging for now."""
         self.transport = transport
         self.log.debug('connected')
+        self._active = False
+        if self.loop and self._keep_alive_interval:
+            self.loop.call_later(self._keep_alive_interval, self.keep_alive)
 
     def data_received(self, data):
         """Add incoming data to buffer."""
         data = data.decode('ascii')
+        self._active = True
         self.log.debug('received data: %s', data)
         self.telegram_buffer.append(data)
 
         for telegram in self.telegram_buffer.get_all():
             self.handle_telegram(telegram)
 
+    def keep_alive(self):
+        if self._active:
+            self.log.debug('keep-alive checked')
+            self._active = False
+            if self.loop:
+                self.loop.call_later(self._keep_alive_interval, self.keep_alive)
+        else:
+            self.log.warning('keep-alive check failed')
+            if self.transport:
+                self.transport.close()
+
     def connection_lost(self, exc):
         """Stop when connection is lost."""
         if exc:

dsmr_parser/parsers.py

@@ -10,7 +10,6 @@ logger = logging.getLogger(__name__)
 
 
 class TelegramParser(object):
-
     crc16_tab = []
 
     def __init__(self, telegram_specification, apply_checksum_validation=True):
@@ -56,7 +55,11 @@ class TelegramParser(object):
             # Some signatures are optional and may not be present,
             # so only parse lines that match
             if match:
-                telegram[signature] = parser.parse(match.group(0))
+                try:
+                    telegram[signature] = parser.parse(match.group(0))
+                except Exception:
+                    logger.error("ignore line with signature {}, because parsing failed.".format(signature),
+                                 exc_info=True)
 
         return telegram
 
@@ -219,6 +222,7 @@ class ProfileGenericParser(DSMRObjectParser):
     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
@@ -271,7 +275,6 @@ class ValueParser(object):
         self.coerce_type = coerce_type
 
     def parse(self, value):
-
         unit_of_measurement = None
 
         if value and '*' in value:

setup.py

@@ -3,10 +3,10 @@ from setuptools import setup, find_packages
 setup(
     name='dsmr-parser',
     description='Library to parse Dutch Smart Meter Requirements (DSMR)',
-    author='Nigel Dokter',
+    author='Nigel Dokter and many others',
     author_email='nigel@nldr.net',
     url='https://github.com/ndokter/dsmr_parser',
-    version='0.27',
+    version='0.28',
     packages=find_packages(exclude=('test', 'test.*')),
     install_requires=[
         'pyserial>=3,<4',

test/test_protocol.py

@@ -5,7 +5,7 @@ import unittest
 from dsmr_parser import obis_references as obis
 from dsmr_parser import telegram_specifications
 from dsmr_parser.parsers import TelegramParser
-from dsmr_parser.clients.protocol import DSMRProtocol
+from dsmr_parser.clients.protocol import create_dsmr_protocol
 
 
 TELEGRAM_V2_2 = (
@@ -35,9 +35,10 @@ TELEGRAM_V2_2 = (
 class ProtocolTest(unittest.TestCase):
 
     def setUp(self):
-        telegram_parser = TelegramParser(telegram_specifications.V2_2)
-        self.protocol = DSMRProtocol(None, telegram_parser,
-                                     telegram_callback=Mock())
+        new_protocol, _ = create_dsmr_protocol('2.2',
+                                               telegram_callback=Mock(),
+                                               keep_alive_interval=1)
+        self.protocol = new_protocol()
 
     def test_complete_packet(self):
         """Protocol should assemble incoming lines into complete packet."""
@@ -52,3 +53,23 @@ class ProtocolTest(unittest.TestCase):
 
         assert float(telegram[obis.GAS_METER_READING].value) == 1.001
         assert telegram[obis.GAS_METER_READING].unit == 'm3'
+
+    def test_receive_packet(self):
+        """Protocol packet reception."""
+
+        mock_transport = Mock()
+        self.protocol.connection_made(mock_transport)
+        assert not self.protocol._active
+
+        self.protocol.data_received(TELEGRAM_V2_2.encode('ascii'))
+        assert self.protocol._active
+
+        # 1st call of keep_alive resets 'active' flag
+        self.protocol.keep_alive()
+        assert not self.protocol._active
+
+        # 2nd call of keep_alive should close the transport
+        self.protocol.keep_alive()
+        assert mock_transport.close.called_once()
+
+        self.protocol.connection_lost(None)