Adds typing support (#338)

* Initial basic type checking

* Fix untyped calls and defs

* Add mypy to CI

* Add py.typed marker for PEP 561 compat

* Fix flake8 spacing in setup.py

* pylint didn't like some line lengths

* Ignore TYPE_CHECKING blocks from code coverage

.coveragerc

@@ -5,3 +5,5 @@ omit =
 
 
 [report]
+exclude_lines = 
+    if TYPE_CHECKING:

.github/workflows/ci.yml

@@ -48,6 +48,8 @@ jobs:
       run: flake8
     - name: Linter Pylint
       run: PYTHONPATH=. pylint --rcfile=.pylintrc pyvlx test/*.py *.py examples/*.py
+    - name: Mypy
+      run: mypy pyvlx
     - name: Tests
       run: PYTHONPATH=. pytest --cov pyvlx --cov-report xml
     - name: Upload coverage artifact

mypy.ini

@@ -0,0 +1,6 @@
+[mypy]
+show_error_codes = True
+warn_unused_ignores = True
+disallow_untyped_defs = True
+disallow_untyped_calls = True
+check_untyped_defs = True

old_api/pyvlx/blind.py

@@ -24,7 +24,7 @@ def from_config(cls, pyvlx, item):
         typeid = item['typeId']
         return cls(pyvlx, ident, name, subtype, typeid)
 
-    def __str__(self):
+    def __str__(self) -> str:
         """Return object as readable string."""
         return '<Blind name="{0}" ' \
             'id="{1}" ' \

old_api/pyvlx/exception.py

@@ -9,7 +9,7 @@ def __init__(self, description):
         super(PyVLXException, self).__init__()
         self.description = description
 
-    def __str__(self):
+    def __str__(self) -> str:
         """Return object as readable string."""
         return '<PyVLXException description="{0}" />' \
             .format(self.description)

old_api/pyvlx/rollershutter.py

@@ -24,7 +24,7 @@ def from_config(cls, pyvlx, item):
         typeid = item['typeId']
         return cls(pyvlx, ident, name, subtype, typeid)
 
-    def __str__(self):
+    def __str__(self) -> str:
         """Return object as readable string."""
         return '<RollerShutter name="{0}" ' \
             'id="{1}" ' \

old_api/pyvlx/scene.py

@@ -25,7 +25,7 @@ def get_name(self):
         """Return name of object."""
         return self.name
 
-    def __str__(self):
+    def __str__(self) -> str:
         """Return object as readable string."""
         return '<Scene name="{0}" ' \
             'id="{1}" />' \

old_api/pyvlx/window.py

@@ -24,7 +24,7 @@ def from_config(cls, pyvlx, item):
         typeid = item['typeId']
         return cls(pyvlx, ident, name, subtype, typeid)
 
-    def __str__(self):
+    def __str__(self) -> str:
         """Return object as readable string."""
         return '<Window name="{0}" ' \
             'id="{1}" ' \

pyvlx/api/activate_scene.py

@@ -1,26 +1,32 @@
 """Module for retrieving scene list from API."""
+from typing import TYPE_CHECKING, Optional
+
 from .api_event import ApiEvent
 from .frames import (
     ActivateSceneConfirmationStatus, FrameActivateSceneConfirmation,
-    FrameActivateSceneRequest, FrameCommandRemainingTimeNotification,
-    FrameCommandRunStatusNotification, FrameSessionFinishedNotification)
+    FrameActivateSceneRequest, FrameBase,
+    FrameCommandRemainingTimeNotification, FrameCommandRunStatusNotification,
+    FrameSessionFinishedNotification)
 from .session_id import get_new_session_id
 
+if TYPE_CHECKING:
+    from pyvlx import PyVLX
+
 
 class ActivateScene(ApiEvent):
     """Class for activating scene via API."""
 
     def __init__(
-            self, pyvlx, scene_id, wait_for_completion=True, timeout_in_seconds=60
+            self, pyvlx: "PyVLX", scene_id: int, wait_for_completion: bool = True, timeout_in_seconds: int = 60
     ):
         """Initialize SceneList class."""
         super().__init__(pyvlx=pyvlx, timeout_in_seconds=timeout_in_seconds)
         self.success = False
         self.scene_id = scene_id
         self.wait_for_completion = wait_for_completion
-        self.session_id = None
+        self.session_id: Optional[int] = None
 
-    async def handle_frame(self, frame):
+    async def handle_frame(self, frame: FrameBase) -> bool:
         """Handle incoming API frame, return True if this was the expected frame."""
         if (
                 isinstance(frame, FrameActivateSceneConfirmation)
@@ -49,7 +55,7 @@ async def handle_frame(self, frame):
             return True
         return False
 
-    def request_frame(self):
+    def request_frame(self) -> FrameActivateSceneRequest:
         """Construct initiating frame."""
         self.session_id = get_new_session_id()
         return FrameActivateSceneRequest(

pyvlx/api/api_event.py

@@ -1,21 +1,27 @@
 """Base class for waiting for a specific answer frame from velux ap.."""
 import asyncio
+from typing import TYPE_CHECKING, Optional
+
+from .frames import FrameBase
+
+if TYPE_CHECKING:
+    from pyvlx import PyVLX
 
 
 class ApiEvent:
     """Base class for waiting a specific frame from API connection."""
 
-    def __init__(self, pyvlx, timeout_in_seconds=10):
+    def __init__(self, pyvlx: "PyVLX", timeout_in_seconds: int = 10):
         """Initialize ApiEvent."""
         self.pyvlx = pyvlx
         self.response_received_or_timeout = asyncio.Event()
 
         self.success = False
         self.timeout_in_seconds = timeout_in_seconds
         self.timeout_callback = None
-        self.timeout_handle = None
+        self.timeout_handle: Optional[asyncio.TimerHandle] = None
 
-    async def do_api_call(self):
+    async def do_api_call(self) -> None:
         """Start. Sending and waiting for answer."""
         self.pyvlx.connection.register_frame_received_cb(self.response_rec_callback)
         await self.send_frame()
@@ -24,33 +30,34 @@ async def do_api_call(self):
         await self.stop_timeout()
         self.pyvlx.connection.unregister_frame_received_cb(self.response_rec_callback)
 
-    async def handle_frame(self, frame):
+    async def handle_frame(self, frame: FrameBase) -> bool:
         """Handle incoming API frame, return True if this was the expected frame."""
         raise NotImplementedError("handle_frame has to be implemented")
 
-    async def send_frame(self):
+    async def send_frame(self) -> None:
         """Send frame to API connection."""
         await self.pyvlx.send_frame(self.request_frame())
 
-    def request_frame(self):
-        """Construct initiating framw."""
+    def request_frame(self) -> FrameBase:
+        """Construct initiating frame."""
         raise NotImplementedError("send_frame has to be implemented")
 
-    async def response_rec_callback(self, frame):
+    async def response_rec_callback(self, frame: FrameBase) -> None:
         """Handle frame. Callback from internal api connection."""
         if await self.handle_frame(frame):
             self.response_received_or_timeout.set()
 
-    def timeout(self):
+    def timeout(self) -> None:
         """Handle timeout for not having received expected frame."""
         self.response_received_or_timeout.set()
 
-    async def start_timeout(self):
+    async def start_timeout(self) -> None:
         """Start timeout."""
         self.timeout_handle = self.pyvlx.connection.loop.call_later(
             self.timeout_in_seconds, self.timeout
         )
 
-    async def stop_timeout(self):
+    async def stop_timeout(self) -> None:
         """Stop timeout."""
-        self.timeout_handle.cancel()
+        if self.timeout_handle is not None:
+            self.timeout_handle.cancel()

pyvlx/api/command_send.py

@@ -1,24 +1,31 @@
 """Module for retrieving scene list from API."""
+from typing import TYPE_CHECKING, Any, Optional
+
+from ..parameter import Parameter
 from .api_event import ApiEvent
 from .frames import (
-    CommandSendConfirmationStatus, FrameCommandRemainingTimeNotification,
-    FrameCommandRunStatusNotification, FrameCommandSendConfirmation,
-    FrameCommandSendRequest, FrameSessionFinishedNotification)
+    CommandSendConfirmationStatus, FrameBase,
+    FrameCommandRemainingTimeNotification, FrameCommandRunStatusNotification,
+    FrameCommandSendConfirmation, FrameCommandSendRequest,
+    FrameSessionFinishedNotification)
 from .session_id import get_new_session_id
 
+if TYPE_CHECKING:
+    from pyvlx import PyVLX
+
 
 class CommandSend(ApiEvent):
     """Class for sending command to API."""
 
     def __init__(
             self,
-            pyvlx,
-            node_id,
-            parameter,
-            active_parameter=0,
-            wait_for_completion=True,
-            timeout_in_seconds=60,
-            **functional_parameter
+            pyvlx: "PyVLX",
+            node_id: int,
+            parameter: Parameter,
+            active_parameter: int = 0,
+            wait_for_completion: bool = True,
+            timeout_in_seconds: int = 60,
+            **functional_parameter: Any
     ):
         """Initialize SceneList class."""
         super().__init__(pyvlx=pyvlx, timeout_in_seconds=timeout_in_seconds)
@@ -28,9 +35,9 @@ def __init__(
         self.active_parameter = active_parameter
         self.functional_parameter = functional_parameter
         self.wait_for_completion = wait_for_completion
-        self.session_id = None
+        self.session_id: Optional[int] = None
 
-    async def handle_frame(self, frame):
+    async def handle_frame(self, frame: FrameBase) -> bool:
         """Handle incoming API frame, return True if this was the expected frame."""
         if (
                 isinstance(frame, FrameCommandSendConfirmation)
@@ -59,7 +66,7 @@ async def handle_frame(self, frame):
             return True
         return False
 
-    def request_frame(self):
+    def request_frame(self) -> FrameCommandSendRequest:
         """Construct initiating frame."""
         self.session_id = get_new_session_id()
         return FrameCommandSendRequest(

pyvlx/api/factory_default.py

@@ -1,28 +1,34 @@
 """Module for handling the FactoryDefault to API."""
+from typing import TYPE_CHECKING
+
 from pyvlx.log import PYVLXLOG
 
 from .api_event import ApiEvent
 from .frames import (
-    FrameGatewayFactoryDefaultConfirmation, FrameGatewayFactoryDefaultRequest)
+    FrameBase, FrameGatewayFactoryDefaultConfirmation,
+    FrameGatewayFactoryDefaultRequest)
+
+if TYPE_CHECKING:
+    from pyvlx import PyVLX
 
 
 class FactoryDefault(ApiEvent):
     """Class for handling Factory reset API."""
 
-    def __init__(self, pyvlx):
+    def __init__(self, pyvlx: "PyVLX"):
         """Initialize facotry default class."""
         super().__init__(pyvlx=pyvlx)
         self.pyvlx = pyvlx
         self.success = False
 
-    async def handle_frame(self, frame):
+    async def handle_frame(self, frame: FrameBase) -> bool:
         """Handle incoming API frame, return True if this was the expected frame."""
         if isinstance(frame, FrameGatewayFactoryDefaultConfirmation):
             PYVLXLOG.warning("KLF200 is factory resetting")
             self.success = True
             return True
         return False
 
-    def request_frame(self):
+    def request_frame(self) -> FrameGatewayFactoryDefaultRequest:
         """Construct initiating frame."""
         return FrameGatewayFactoryDefaultRequest()

pyvlx/api/frame_creation.py

@@ -1,10 +1,12 @@
 """Helper module for creating a frame out of raw data."""
+from typing import Optional
+
 from pyvlx.const import Command
 from pyvlx.log import PYVLXLOG
 
 from .frames import (
     FrameActivateSceneConfirmation, FrameActivateSceneRequest,
-    FrameActivationLogUpdatedNotification,
+    FrameActivationLogUpdatedNotification, FrameBase,
     FrameCommandRemainingTimeNotification, FrameCommandRunStatusNotification,
     FrameCommandSendConfirmation, FrameCommandSendRequest,
     FrameDiscoverNodesConfirmation, FrameDiscoverNodesNotification,
@@ -38,7 +40,7 @@
     FrameStatusRequestRequest, extract_from_frame)
 
 
-def frame_from_raw(raw):
+def frame_from_raw(raw: bytes) -> Optional[FrameBase]:
     """Create and return frame from raw bytes."""
     command, payload = extract_from_frame(raw)
     frame = create_frame(command)
@@ -54,7 +56,7 @@ def frame_from_raw(raw):
     return frame
 
 
-def create_frame(command):
+def create_frame(command: Command) -> Optional[FrameBase]:
     """Create and return empty Frame from Command."""
     # pylint: disable=too-many-branches,too-many-return-statements,too-many-statements
     if command == Command.GW_ERROR_NTF:

pyvlx/api/frames/alias_array.py

@@ -1,32 +1,34 @@
 """Module for storing alias array."""
+from typing import List, Optional, Tuple
+
 from pyvlx.exception import PyVLXException
 
 
 class AliasArray:
     """Object for storing alias array."""
 
-    def __init__(self, raw=None):
+    def __init__(self, raw: Optional[bytes] = None):
         """Initialize alias array."""
-        self.alias_array_ = []
+        self.alias_array_: List[Tuple[bytes, bytes]] = []
         if raw is not None:
             self.parse_raw(raw)
 
-    def __str__(self):
+    def __str__(self) -> str:
         """Return human readable string."""
         return ", ".join(
             "{:02x}{:02x}={:02x}{:02x}".format(c[0][0], c[0][1], c[1][0], c[1][1])
             for c in self.alias_array_
         )
 
-    def __bytes__(self):
+    def __bytes__(self) -> bytes:
         """Get raw bytes of alias array."""
         ret = bytes([len(self.alias_array_)])
         for alias in self.alias_array_:
             ret += alias[0] + alias[1]
         ret += bytes((5 - len(self.alias_array_)) * 4)
         return ret
 
-    def parse_raw(self, raw):
+    def parse_raw(self, raw: bytes) -> None:
         """Parse alias array from raw bytes."""
         if not isinstance(raw, bytes):
             raise PyVLXException("AliasArray::invalid_type_if_raw", type_raw=type(raw))