Add basic typing (#169)

.github/workflows/python-package.yml

@@ -26,7 +26,10 @@ jobs:
             - name: Install dependencies
               run: |
                   python -m pip install --upgrade pip
-                  python -m pip install flake8 pytest .[test] -r requirements.txt
+                  python -m pip install flake8 pytest .[test] -r requirements.txt -r requirements_test.txt
+            - name: mypy
+              run: |
+                  mypy flux_led
             - name: Lint with flake8
               run: |
                   # stop the build if there are Python syntax errors or undefined names

flux_led/aiodevice.py

@@ -1,7 +1,7 @@
 import asyncio
 import contextlib
 import logging
-from typing import Callable, Coroutine, List, Optional
+from typing import Callable, List, Optional
 
 from .aioprotocol import AIOLEDENETProtocol
 from .base_device import LEDENETDevice
@@ -52,9 +52,9 @@ async def _async_send_state_query(self):
         await self._async_send_msg(self._protocol.construct_state_query())
 
     async def _async_execute_and_wait_for(
-        self, futures: List[asyncio.Future], coro: Coroutine
-    ) -> None:
-        future = asyncio.Future()
+        self, futures: List[asyncio.Future], coro: Callable
+    ) -> bool:
+        future: asyncio.Future = asyncio.Future()
         futures.append(future)
         await coro()
         _LOGGER.debug("%s: Waiting for power state response", self.ipaddr)
@@ -75,6 +75,7 @@ async def _async_execute_and_wait_for(
         return False
 
     async def _async_turn_on(self) -> None:
+        assert self._protocol is not None
         await self._async_send_msg(self._protocol.construct_state_change(True))
 
     async def _async_turn_off_on(self) -> None:
@@ -96,6 +97,7 @@ async def async_turn_on(self) -> bool:
         return False
 
     async def _async_turn_off(self) -> None:
+        assert self._protocol is not None
         await self._async_send_msg(self._protocol.construct_state_change(False))
 
     async def _async_turn_on_off(self) -> None:

flux_led/aioprotocol.py

@@ -1,5 +1,7 @@
 import asyncio
+from asyncio.transports import BaseTransport, WriteTransport
 import logging
+from typing import Optional, cast
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -10,21 +12,22 @@ class AIOLEDENETProtocol(asyncio.Protocol):
     def __init__(self, data_received, connection_lost) -> None:
         self._data_receive_callback = data_received
         self._connection_lost_callback = connection_lost
-        self.transport = None
+        self.transport: Optional[WriteTransport] = None
 
-    def connection_lost(self, exc: Exception) -> None:
+    def connection_lost(self, exc: Optional[Exception]) -> None:
         """Handle connection lost."""
         _LOGGER.debug("%s: Connection lost: %s", self.peername, exc)
         self.close()
         self._connection_lost_callback(exc)
 
-    def connection_made(self, transport: asyncio.Transport) -> None:
+    def connection_made(self, transport: BaseTransport) -> None:
         """Handle connection made."""
-        self.transport = transport
+        self.transport = cast(WriteTransport, transport)
         self.peername = transport.get_extra_info("peername")
 
     def write(self, data: bytes) -> None:
         """Write data to the client."""
+        assert self.transport is not None
         _LOGGER.debug(
             "%s => %s (%d)",
             self.peername,
@@ -35,6 +38,7 @@ def write(self, data: bytes) -> None:
 
     def close(self) -> None:
         """Remove the connection and close the transport."""
+        assert self.transport is not None
         self.transport.write_eof()
         self.transport.close()