Feature/usb reconnect (#99)

* Add connection_lost handler to uart.
* Reconnect serial port on disconnects.
* Remove logging.
* Reduce Watchdog TTL to 10min.
* Update logging.
* Catch watchdog exceptions.
* Throw an Exception if command is sent while disconnected.
* Use new zigpy exceptions.
* Update test requirements.

Author: Adminiuga

setup.py

@@ -13,6 +13,6 @@
     author_email="[email protected]",
     license="GPL-3.0",
     packages=find_packages(exclude=["*.tests"]),
-    install_requires=["pyserial-asyncio", "zigpy-homeassistant>=0.10.0"],
-    tests_require=["pytest"],
+    install_requires=["pyserial-asyncio", "zigpy-homeassistant>=0.17.0"],
+    tests_require=["pytest", "pytest-asyncio", "asynctest"],
 )

tests/test_api.py

@@ -1,6 +1,8 @@
 import asyncio
+import logging
 from unittest import mock
 
+import asynctest
 import pytest
 
 from zigpy_deconz import api as deconz_api, types as t, uart
@@ -95,6 +97,22 @@ def mock_api_frame(name, *args):
         api._uart.send.reset_mock()
 
 
+@pytest.mark.asyncio
+async def test_command_not_connected(api):
+    api._uart = None
+
+    def mock_api_frame(name, *args):
+        return mock.sentinel.api_frame_data, api._seq
+
+    api._api_frame = mock.MagicMock(side_effect=mock_api_frame)
+
+    for cmd, cmd_opts in deconz_api.TX_COMMANDS.items():
+        with pytest.raises(deconz_api.CommandError):
+            await api._command(cmd, mock.sentinel.cmd_data)
+        assert api._api_frame.call_count == 0
+        api._api_frame.reset_mock()
+
+
 def test_api_frame(api):
     addr = t.DeconzAddressEndpoint()
     addr.address_mode = t.ADDRESS_MODE.NWK
@@ -414,3 +432,54 @@ def test_device_state_network_state(data, network_state):
         assert rest == extra
         assert state.network_state == deconz_api.NetworkState[network_state]
         assert state.serialize() == new_data
+
+
+@pytest.mark.asyncio
+async def test_reconnect_multiple_disconnects(monkeypatch, caplog):
+    api = deconz_api.Deconz()
+    dev = mock.sentinel.uart
+    connect_mock = asynctest.CoroutineMock()
+    connect_mock.return_value = asyncio.Future()
+    connect_mock.return_value.set_result(True)
+    monkeypatch.setattr(uart, "connect", connect_mock)
+
+    await api.connect(dev, 115200)
+
+    caplog.set_level(logging.DEBUG)
+    connected = asyncio.Future()
+    connected.set_result(mock.sentinel.uart_reconnect)
+    connect_mock.reset_mock()
+    connect_mock.side_effect = [asyncio.Future(), connected]
+    api.connection_lost("connection lost")
+    await asyncio.sleep(0.3)
+    api.connection_lost("connection lost 2")
+    await asyncio.sleep(0.3)
+
+    assert "Cancelling reconnection attempt" in caplog.messages
+    assert api._uart is mock.sentinel.uart_reconnect
+    assert connect_mock.call_count == 2
+
+
+@pytest.mark.asyncio
+async def test_reconnect_multiple_attempts(monkeypatch, caplog):
+    api = deconz_api.Deconz()
+    dev = mock.sentinel.uart
+    connect_mock = asynctest.CoroutineMock()
+    connect_mock.return_value = asyncio.Future()
+    connect_mock.return_value.set_result(True)
+    monkeypatch.setattr(uart, "connect", connect_mock)
+
+    await api.connect(dev, 115200)
+
+    caplog.set_level(logging.DEBUG)
+    connected = asyncio.Future()
+    connected.set_result(mock.sentinel.uart_reconnect)
+    connect_mock.reset_mock()
+    connect_mock.side_effect = [asyncio.TimeoutError, OSError, connected]
+
+    with asynctest.mock.patch("asyncio.sleep"):
+        api.connection_lost("connection lost")
+        await api._conn_lost_task
+
+    assert api._uart is mock.sentinel.uart_reconnect
+    assert connect_mock.call_count == 3

tests/test_application.py

@@ -2,6 +2,7 @@
 import logging
 from unittest import mock
 
+import asynctest
 import pytest
 
 import zigpy.device
@@ -510,3 +511,21 @@ async def test_mrequest_send_fail(app):
 async def test_mrequest_send_aps_data_error(app):
     r = await _test_mrequest(app, False, aps_data_error=True)
     assert r[0] != 0
+
+
+@pytest.mark.asyncio
+@mock.patch.object(application, "WATCHDOG_TTL", new=1)
+async def test_reset_watchdog(app):
+    """Test watchdog."""
+    with asynctest.patch.object(app._api, "write_parameter") as mock_api:
+        dog = asyncio.ensure_future(app._reset_watchdog())
+        await asyncio.sleep(0.3)
+        dog.cancel()
+        assert mock_api.call_count == 1
+
+    with asynctest.patch.object(app._api, "write_parameter") as mock_api:
+        mock_api.side_effect = zigpy_deconz.exception.CommandError
+        dog = asyncio.ensure_future(app._reset_watchdog())
+        await asyncio.sleep(0.3)
+        dog.cancel()
+        assert mock_api.call_count == 1

tests/test_uart.py

@@ -107,3 +107,17 @@ def test_checksum(gw):
     checksum = b"\x44\xFF"
     r = gw._checksum(data)
     assert r == checksum
+
+
+def test_connection_lost_exc(gw):
+    gw.connection_lost(mock.sentinel.exception)
+
+    conn_lost = gw._api.connection_lost
+    assert conn_lost.call_count == 1
+    assert conn_lost.call_args[0][0] is mock.sentinel.exception
+
+
+def test_connection_closed(gw):
+    gw.connection_lost(None)
+
+    assert gw._api.connection_lost.call_count == 0

tox.ini

@@ -12,6 +12,7 @@ setenv = PYTHONPATH = {toxinidir}
 install_command = pip install {opts} {packages}
 commands = py.test --cov --cov-report=
 deps =
+    asynctest
     coveralls
     pytest
     pytest-cov

zigpy_deconz/api.py

@@ -186,10 +186,12 @@ class NetworkParameter(t.uint8_t, enum.Enum):
 class Deconz:
     def __init__(self):
         self._uart = None
+        self._uart_path = None
         self._seq = 1
         self._awaiting = {}
         self._app = None
         self._cmd_mode_future = None
+        self._conn_lost_task = None
         self._device_state = DeviceState(NetworkState.OFFLINE)
         self._data_indication = False
         self._data_confirm = False
@@ -209,15 +211,57 @@ def protocol_version(self):
     def set_application(self, app):
         self._app = app
 
-    async def connect(self, device, baudrate=DECONZ_BAUDRATE):
+    async def connect(self, device: str, baudrate: int = DECONZ_BAUDRATE) -> None:
         assert self._uart is None
+        self._uart_path = device
         self._uart = await uart.connect(device, DECONZ_BAUDRATE, self)
 
+    def connection_lost(self, exc: Exception) -> None:
+        """Lost serial connection."""
+        LOGGER.warning(
+            "Serial '%s' connection lost unexpectedly: %s", self._uart_path, exc
+        )
+        self._uart = None
+        if self._conn_lost_task and not self._conn_lost_task.done():
+            self._conn_lost_task.cancel()
+        self._conn_lost_task = asyncio.ensure_future(self._connection_lost())
+
+    async def _connection_lost(self) -> None:
+        """Reconnect serial port."""
+        try:
+            await self._reconnect_till_done()
+        except asyncio.CancelledError:
+            LOGGER.debug("Cancelling reconnection attempt")
+
+    async def _reconnect_till_done(self) -> None:
+        attempt = 1
+        while True:
+            try:
+                await asyncio.wait_for(self.reconnect(), timeout=10)
+                break
+            except (asyncio.TimeoutError, OSError) as exc:
+                wait = 2 ** min(attempt, 5)
+                attempt += 1
+                LOGGER.debug(
+                    "Couldn't re-open '%s' serial port, retrying in %ss: %s",
+                    self._uart_path,
+                    wait,
+                    str(exc),
+                )
+                await asyncio.sleep(wait)
+
+        LOGGER.debug(
+            "Reconnected '%s' serial port after %s attempts", self._uart_path, attempt
+        )
+
     def close(self):
         return self._uart.close()
 
     async def _command(self, cmd, *args):
         LOGGER.debug("Command %s %s", cmd, args)
+        if self._uart is None:
+            # connection was lost
+            raise CommandError(Status.ERROR, "API is not running")
         data, seq = self._api_frame(cmd, *args)
         self._uart.send(data)
         fut = asyncio.Future()
@@ -298,6 +342,11 @@ async def read_parameter(self, id_, *args):
         LOGGER.debug("Read parameter %s response: %s", param.name, data)
         return data
 
+    def reconnect(self):
+        """Reconnect using saved parameters."""
+        LOGGER.debug("Reconnecting '%s' serial port", self._uart_path)
+        return self.connect(self._uart_path)
+
     def _handle_read_parameter(self, data):
         pass
 

zigpy_deconz/exception.py

@@ -1,11 +1,7 @@
-from zigpy.exceptions import ZigbeeException
+from zigpy.exceptions import APIException
 
 
-class DeconzException(ZigbeeException):
-    pass
-
-
-class CommandError(DeconzException):
+class CommandError(APIException):
     def __init__(self, status, *args, **kwargs):
         self._status = status
         super().__init__(*args, **kwargs)

zigpy_deconz/uart.py

@@ -15,9 +15,24 @@ class Gateway(asyncio.Protocol):
     ESC_ESC = b"\xDD"
 
     def __init__(self, api, connected_future=None):
+        self._api = api
         self._buffer = b""
         self._connected_future = connected_future
-        self._api = api
+        self._transport = None
+
+    def connection_lost(self, exc) -> None:
+        """Port was closed expecteddly or unexpectedly."""
+        if self._connected_future and not self._connected_future.done():
+            if exc is None:
+                self._connected_future.set_result(True)
+            else:
+                self._connected_future.set_exception(exc)
+        if exc is None:
+            LOGGER.debug("Closed serial connection")
+            return
+
+        LOGGER.error("Lost serial connection: %s", exc)
+        self._api.connection_lost(exc)
 
     def connection_made(self, transport):
         """Callback when the uart is connected"""

zigpy_deconz/zigbee/application.py

@@ -8,15 +8,17 @@
 import zigpy.exceptions
 import zigpy.types
 import zigpy.util
-import zigpy_deconz.exception
+
 from zigpy_deconz import types as t
 from zigpy_deconz.api import NetworkParameter, NetworkState, Status
+import zigpy_deconz.exception
 
 LOGGER = logging.getLogger(__name__)
 
 CHANGE_NETWORK_WAIT = 1
 SEND_CONFIRM_TIMEOUT = 60
 PROTO_VER_WATCHDOG = 0x0108
+WATCHDOG_TTL = 600
 
 
 class ControllerApplication(zigpy.application.ControllerApplication):
@@ -33,8 +35,13 @@ def __init__(self, api, database_file=None):
 
     async def _reset_watchdog(self):
         while True:
-            await self._api.write_parameter(NetworkParameter.watchdog_ttl, 3600)
-            await asyncio.sleep(1200)
+            try:
+                await self._api.write_parameter(
+                    NetworkParameter.watchdog_ttl, WATCHDOG_TTL
+                )
+            except (asyncio.TimeoutError, zigpy.exceptions.ZigbeeException):
+                LOGGER.warning("No watchdog response")
+            await asyncio.sleep(WATCHDOG_TTL * 0.75)
 
     async def shutdown(self):
         """Shutdown application."""