Rework connection logic for reset sequence

RFXtrx/__init__.py

@@ -24,13 +24,12 @@
 # pylint: disable= too-many-lines
 
 import functools
-import glob
 import socket
 import threading
 import logging
 from contextlib import suppress
 
-from time import sleep
+from time import sleep, time
 
 import serial
 
@@ -795,23 +794,23 @@ def __init__(self, *args, **kwargs):
     def write(self, *args, **kwargs):
         """ Dummy function for writing"""
 
-    # pylint: disable=invalid-name
-    def flushInput(self, *args, **kwargs):
-        """ Called by PySerialTransport"""
-
-    def read(self, data=None):
+    def read(self):
         """ Dummy function for reading"""
-        if data is not None or self._read_num >= len(self._data):
-            self._close_event.wait(0.1)
-            return []
+        if self._read_num >= len(self._data):
+            self._close_event.wait()
+            raise serial.PortNotOpenError()
         res = self._data[self._read_num]
         self._read_num = self._read_num + 1
-        return res
+        return bytes(res)
 
     def close(self):
         """ close connection to rfxtrx device """
         self._close_event.set()
 
+    def open(self):
+        """ open connection to rfxtrx device """
+        self._close_event.clear()
+
 
 ###############################################################################
 # RFXtrxTransportError class
@@ -821,6 +820,7 @@ def close(self):
 class RFXtrxTransportError(Exception):
     """ Connection error """
 
+
 ###############################################################################
 # RFXtrxTransport class
 ###############################################################################
@@ -829,6 +829,10 @@ class RFXtrxTransportError(Exception):
 class RFXtrxTransport:
     """ Abstract superclass for all transport mechanisms """
 
+    RESET_SLEEP_TIME = 0.3
+    """ Time delay efter reset to ensure it finished.
+        Reported to not be enough with documented 0.05s """
+
     # pylint: disable=attribute-defined-outside-init
     @staticmethod
     def parse(data):
@@ -890,22 +894,34 @@ def __errors(instance: RFXtrxTransport, *args, **kargs):
 
 class PySerialTransport(RFXtrxTransport):
     """ Implementation of a transport using PySerial """
+    CONNECTION_RETRY_INTERVAL = 0.2
+    CONNECTION_RESET_TIMEOUT = 0.6
 
-    def __init__(self, port):
-        self.port = port
-        self.serial = None
+    def __init__(self, port, serial_object=None):
+        if serial_object:
+            self.serial = serial_object
+        else:
+            self.serial = serial.Serial(baudrate=38400)
+        self.serial.port = port
 
     @transport_errors("connect")
     def connect(self, timeout=None):
         """ Open a serial connexion """
-        try:
-            self.serial = serial.Serial(self.port, 38400)
-        except serial.SerialException:
-            port = glob.glob('/dev/serial/by-id/usb-RFXCOM_*-port0')
-            if len(port) < 1:
-                raise
-            _LOGGER.debug("Attempting connection by name %s", port)
-            self.serial = serial.Serial(port[0], 38400)
+
+        if timeout:
+            timeout += time() - self.CONNECTION_RETRY_INTERVAL
+
+        while True:
+            try:
+                self.serial.open()
+                return
+
+            except serial.SerialException:
+                if timeout is None or time() >= timeout:
+                    raise
+                _LOGGER.debug("Retrying connection", exc_info=True)
+
+            sleep(self.CONNECTION_RETRY_INTERVAL)
 
     @transport_errors("receive")
     def receive_blocking(self):
@@ -914,8 +930,6 @@ def receive_blocking(self):
     def _receive_packet(self):
         """ Wait until a packet is received and return with an RFXtrxEvent """
         data = self.serial.read()
-        if data == '\x00':
-            return None
         pkt = bytearray(data)
         while len(pkt) < pkt[0]+1:
             data = self.serial.read(pkt[0]+1 - len(pkt))
@@ -946,8 +960,9 @@ def reset(self):
         """ Reset the RFXtrx """
         self.send(b'\x0D\x00\x00\x00\x00\x00\x00'
                   b'\x00\x00\x00\x00\x00\x00\x00')
-        sleep(0.3)  # Should work with 0.05, but not for me
-        self.serial.flushInput()
+        self.close()
+        sleep(self.RESET_SLEEP_TIME)
+        self.connect(timeout=self.CONNECTION_RESET_TIMEOUT)
 
     @transport_errors("close")
     def close(self):
@@ -985,8 +1000,6 @@ def _receive_packet(self):
         data = self.sock.recv(1)
         if data == b'':
             raise RFXtrxTransportError("Server was shutdown")
-        if data == '\x00':
-            return None
         pkt = bytearray(data)
         while len(pkt) < pkt[0]+1:
             data = self.sock.recv(pkt[0]+1 - len(pkt))
@@ -1012,19 +1025,26 @@ def send(self, data):
             "Send: %s",
             " ".join("0x{0:02x}".format(x) for x in pkt)
         )
-        self.sock.send(pkt)
+        self.sock.sendall(pkt)
+
+    def _flush_receive(self):
+        self.sock.settimeout(0.0)
+        while True:
+            try:
+                if self.sock.recv(100) == b"":
+                    raise RFXtrxTransportError("Server was shutdown")
+            except (socket.timeout, BlockingIOError):
+                break
+        self.sock.settimeout(None)
 
     @transport_errors("reset")
     def reset(self):
         """ Reset the RFXtrx """
-        try:
-            self.send(b'\x0D\x00\x00\x00\x00\x00\x00'
-                      b'\x00\x00\x00\x00\x00\x00\x00')
-            sleep(0.3)
-            self.sock.sendall(b'')
-        except socket.error as exception:
-            raise RFXtrxTransportError(
-                "Reset failed: {0}".format(exception)) from exception
+        self.send(b'\x0D\x00\x00\x00\x00\x00\x00'
+                  b'\x00\x00\x00\x00\x00\x00\x00')
+
+        sleep(self.RESET_SLEEP_TIME)
+        self._flush_receive()
 
     @transport_errors("close")
     def close(self):
@@ -1077,12 +1097,8 @@ def close(self):
 class DummyTransport2(PySerialTransport):
     """ Dummy transport for testing purposes """
     #  pylint: disable=super-init-not-called
-    def __init__(self, device=""):
-        self.serial = _dummySerial(device, 38400, timeout=0.1)
-        self._run_event = threading.Event()
-
-    def connect(self, timeout=None):
-        self._run_event.set()
+    def __init__(self, port=""):
+        super().__init__(port, _dummySerial())
 
 
 class Connect:

requirements_test.txt

@@ -3,4 +3,5 @@ pylint>=1.5.3
 coveralls>=1.1
 pytest>=2.8.0
 pytest-cov>=2.2.0
-pytest-timeout>=1.0.0
+pytest-timeout>=1.0.0
+freezegun>=1.5.1

tests/conftest.py

@@ -0,0 +1,12 @@
+from datetime import timedelta
+import pytest
+from freezegun import freeze_time
+from unittest import mock
+
+
+@pytest.fixture(name="frozen_sleep", autouse=True)
+def fixture_frozen_sleep():
+    with mock.patch('RFXtrx.sleep') as mock_sleep:
+        with freeze_time() as frozen_time:
+            mock_sleep.side_effect = lambda seconds: frozen_time.tick(timedelta(seconds=seconds))
+            yield frozen_time

tests/test_transport_network.py

@@ -62,6 +62,20 @@ def connected_transport(server: Server):
     return transport, server.connections[-1]
 
 
+def test_transport_flush_reset(server: Server):
+    transport, connection = connected_transport(server)
+    connection.sendall(bytes([0x09, 0x00, 0x00, 0x00, 0x00,
+                              0x00, 0x00, 0x00, 0x00, 0x00]))
+
+    transport.reset()
+
+    connection.sendall(bytes([0x09, 0x03, 0x01, 0x04, 0x28,
+                              0x0a, 0xb7, 0x66, 0x04, 0x70]))
+
+    pkt = transport.receive_blocking()
+    assert isinstance(pkt, RFXtrx.SensorEvent)
+
+
 def test_transport_shutdown_between_packet(server: Server):
     transport, connection = connected_transport(server)
     connection.sendall(bytes([0x09, 0x03, 0x01, 0x04, 0x28,

tests/test_transport_serial.py

@@ -0,0 +1,53 @@
+
+import pytest
+from serial import Serial, SerialException
+from unittest import mock
+import RFXtrx
+
+
+def test_open():
+    serial_mock: mock.Mock | Serial = mock.Mock(spec=Serial)
+    transport =  RFXtrx.PySerialTransport(0, serial_mock)
+    transport.connect()
+    assert serial_mock.open.call_count == 1
+
+
+def test_reset(frozen_sleep):
+    serial_mock: mock.Mock | Serial = mock.Mock(spec=Serial)
+    transport =  RFXtrx.PySerialTransport(0, serial_mock)
+    transport.connect()
+    assert serial_mock.open.call_count == 1
+    transport.reset()
+    serial_mock.write.assert_called_with(b'\r\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00')
+    assert serial_mock.close.call_count == 1
+    assert serial_mock.open.call_count == 2
+
+
+def test_reset_retry(frozen_sleep):
+    """Verify that we retry """
+    serial_mock: mock.Mock | Serial = mock.Mock(spec=Serial)
+    serial_mock.open.side_effect = [None, SerialException(), None]
+    transport =  RFXtrx.PySerialTransport(0, serial_mock)
+    transport.CONNECTION_RESET_TIMEOUT  = 2
+    transport.CONNECTION_RETRY_INTERVAL = 1
+
+    transport.connect()
+    assert serial_mock.open.call_count == 1
+    transport.reset()
+    serial_mock.write.assert_called_with(b'\r\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00')
+    assert serial_mock.close.call_count == 1
+    assert serial_mock.open.call_count == 3
+
+
+def test_reset_retry_timeout(frozen_sleep):
+    """Verify that we retry """
+    serial_mock: mock.Mock | Serial = mock.Mock(spec=Serial)
+    serial_mock.open.side_effect = [None, SerialException(), SerialException()]
+    transport = RFXtrx.PySerialTransport(0, serial_mock)
+    transport.CONNECTION_RESET_TIMEOUT  = 2
+    transport.CONNECTION_RETRY_INTERVAL = 1
+
+    transport.connect()
+    with pytest.raises(RFXtrx.RFXtrxTransportError):
+        transport.reset()
+