Apply stricter typing to this project

.github/workflows/typing.yaml

@@ -0,0 +1,52 @@
+---
+name: Typing
+
+# yamllint disable-line rule:truthy
+on:
+  push:
+  pull_request:
+  workflow_dispatch:
+
+jobs:
+  mypy:
+    name: mypy on Python ${{ matrix.python }}
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        python: ["3.9", "3.10"]
+    steps:
+      - name: ⤵️ Check out code from GitHub
+        uses: actions/checkout@v3
+      - name: 🏗 Set up Python ${{ matrix.python }}
+        id: python
+        uses: actions/setup-python@v3
+        with:
+          python-version: ${{ matrix.python }}
+      - name: 🏗 Get pip cache dir
+        id: pip-cache
+        run: |
+          echo "::set-output name=dir::$(pip cache dir)"
+      - name: ⤵️ Restore cached Python PIP packages
+        uses: actions/cache@v2.1.7
+        with:
+          path: ${{ steps.pip-cache.outputs.dir }}
+          key: pip-${{ runner.os }}-v1-${{ steps.python.outputs.python-version }}-${{ hashFiles('.github/workflows/requirements.txt') }}
+          restore-keys: |
+            pip-${{ runner.os }}-v1-${{ steps.python.outputs.python-version }}-
+      - name: 🏗 Install workflow dependencies
+        run: |
+          pip install -r .github/workflows/requirements.txt
+          poetry config virtualenvs.create true
+          poetry config virtualenvs.in-project true
+      - name: ⤵️ Restore cached Python virtual environment
+        id: cached-poetry-dependencies
+        uses: actions/cache@v2.1.7
+        with:
+          path: .venv
+          key: venv-${{ runner.os }}-v1-${{ steps.python.outputs.python-version }}-${{ hashFiles('poetry.lock') }}
+          restore-keys: |
+            venv-${{ runner.os }}-v1-${{ steps.python.outputs.python-version }}-
+      - name: 🏗 Install dependencies
+        run: poetry install --no-interaction
+      - name: 🚀 Run mypy
+        run: poetry run mypy omnikinverter tests

omnikinverter/models.py

@@ -1,10 +1,9 @@
 """Models for Omnik Inverter."""
 from __future__ import annotations
 
-import json
 import re
 from dataclasses import dataclass
-from typing import Any
+from typing import Any, cast
 
 from .exceptions import OmnikInverterWrongSourceError, OmnikInverterWrongValuesError
 
@@ -36,14 +35,13 @@ def from_json(data: dict[str, Any]) -> Inverter:
             OmnikInverterWrongValuesError: Inverter pass on
                 incorrect data (day and total are equal).
         """
-        data = json.loads(data)
 
-        def get_value(search):
+        def get_value(search: str) -> Any:
             if data[search] != "":
                 return data[search]
             return None
 
-        def validation(data_list):
+        def validation(data_list: list[Any]) -> bool:
             """Check if the values are not equal to each other.
 
             Args:
@@ -72,7 +70,7 @@ def validation(data_list):
         )
 
     @staticmethod
-    def from_html(data: dict[str, Any]) -> Inverter:
+    def from_html(data: str) -> Inverter:
         """Return Inverter object from the Omnik Inverter response.
 
         Args:
@@ -82,16 +80,18 @@ def from_html(data: dict[str, Any]) -> Inverter:
             An Inverter object.
         """
 
-        def get_value(search_key):
-            match = re.search(f'(?<={search_key}=").*?(?=";)', data.replace(" ", ""))
+        def get_value(search_key: str) -> Any:
             try:
-                value = match.group(0)
-                if value != "":
+                match = cast(
+                    re.Match[str],
+                    re.search(f'(?<={search_key}=").*?(?=";)', data.replace(" ", "")),
+                ).group(0)
+                if match != "":
                     if search_key in ["webdata_now_p", "webdata_rate_p"]:
-                        return int(value)
+                        return int(match)
                     if search_key in ["webdata_today_e", "webdata_total_e"]:
-                        return float(value)
-                    return value
+                        return float(match)
+                    return match
                 return None
             except AttributeError as exception:
                 raise OmnikInverterWrongSourceError(
@@ -110,7 +110,7 @@ def get_value(search_key):
         )
 
     @staticmethod
-    def from_js(data: dict[str, Any]) -> Inverter:
+    def from_js(data: str) -> Inverter:
         """Return Inverter object from the Omnik Inverter response.
 
         Args:
@@ -120,25 +120,35 @@ def from_js(data: dict[str, Any]) -> Inverter:
             An Inverter object.
         """
 
-        def get_value(position):
-            if data.find("webData") != -1:
-                matches = re.search(r'(?<=webData=").*?(?=";)', data)
-            else:
-                matches = re.search(r'(?<=myDeviceArray\[0\]=").*?(?=";)', data)
-
+        def get_value(position: int) -> Any:
             try:
-                data_list = matches.group(0).split(",")
-                if data_list[position] != "":
+                if data.find("webData") != -1:
+                    matches = (
+                        cast(re.Match[str], re.search(r'(?<=webData=").*?(?=";)', data))
+                        .group(0)
+                        .split(",")
+                    )
+                else:
+                    matches = (
+                        cast(
+                            re.Match[str],
+                            re.search(r'(?<=myDeviceArray\[0\]=").*?(?=";)', data),
+                        )
+                        .group(0)
+                        .split(",")
+                    )
+
+                if matches[position] != "":
                     if position in [4, 5, 6, 7]:
                         if position in [4, 5]:
-                            return int(data_list[position])
+                            return int(matches[position])
 
                         if position == 6:
-                            energy_value = float(data_list[position]) / 100
+                            energy_value = float(matches[position]) / 100
                         if position == 7:
-                            energy_value = float(data_list[position]) / 10
+                            energy_value = float(matches[position]) / 10
                         return energy_value
-                    return data_list[position].replace(" ", "")
+                    return matches[position].replace(" ", "")
                 return None
             except AttributeError as exception:
                 raise OmnikInverterWrongSourceError(
@@ -175,16 +185,14 @@ def from_json(data: dict[str, Any]) -> Device:
         Returns:
             An Device object.
         """
-
-        data = json.loads(data)
         return Device(
             signal_quality=None,
             firmware=data["g_ver"].replace("VER:", ""),
             ip_address=data["ip"],
         )
 
     @staticmethod
-    def from_html(data: dict[str, Any]) -> Device:
+    def from_html(data: str) -> Device:
         """Return Device object from the Omnik Inverter response.
 
         Args:
@@ -197,13 +205,15 @@ def from_html(data: dict[str, Any]) -> Device:
         for correction in [" ", "%"]:
             data = data.replace(correction, "")
 
-        def get_value(search_key):
-            match = re.search(f'(?<={search_key}=").*?(?=";)', data)
-            value = match.group(0)
-            if value != "":
+        def get_value(search_key: str) -> Any:
+            match = cast(
+                re.Match[str], re.search(f'(?<={search_key}=").*?(?=";)', data)
+            ).group(0)
+
+            if match != "":
                 if search_key in ["cover_sta_rssi"]:
-                    return int(value)
-                return value
+                    return int(match)
+                return match
             return None
 
         return Device(
@@ -213,7 +223,7 @@ def get_value(search_key):
         )
 
     @staticmethod
-    def from_js(data: dict[str, Any]) -> Device:
+    def from_js(data: str) -> Device:
         """Return Device object from the Omnik Inverter response.
 
         Args:
@@ -225,13 +235,15 @@ def from_js(data: dict[str, Any]) -> Device:
         for correction in [" ", "%"]:
             data = data.replace(correction, "")
 
-        def get_value(search_key):
-            match = re.search(f'(?<={search_key}=").*?(?=";)', data)
-            value = match.group(0)
-            if value != "":
+        def get_value(search_key: str) -> Any:
+            match = cast(
+                re.Match[str], re.search(f'(?<={search_key}=").*?(?=";)', data)
+            ).group(0)
+
+            if match != "":
                 if search_key == "m2mRssi":
-                    return int(value)
-                return value
+                    return int(match)
+                return match
             return None
 
         return Device(

omnikinverter/omnikinverter.py

@@ -2,6 +2,7 @@
 from __future__ import annotations
 
 import asyncio
+import json
 from collections.abc import Mapping
 from dataclasses import dataclass
 from typing import Any
@@ -28,7 +29,7 @@ class OmnikInverter:
     username: str | None = None
     password: str | None = None
     source_type: str = "javascript"
-    request_timeout: int = 10
+    request_timeout: float = 10.0
     session: ClientSession | None = None
 
     _close_session: bool = False
@@ -38,8 +39,8 @@ async def request(
         uri: str,
         *,
         method: str = METH_GET,
-        params: Mapping[str, str] | None = None,
-    ) -> dict[str, Any]:
+        params: Mapping[str, Any] | None = None,
+    ) -> str:
         """Handle a request to a Omnik Inverter device.
 
         Args:
@@ -124,7 +125,7 @@ async def inverter(self) -> Inverter:
         """
         if self.source_type == "json":
             data = await self.request("status.json", params={"CMD": "inv_query"})
-            return Inverter.from_json(data)
+            return Inverter.from_json(json.loads(data))
         if self.source_type == "html":
             data = await self.request("status.html")
             return Inverter.from_html(data)
@@ -139,7 +140,7 @@ async def device(self) -> Device:
         """
         if self.source_type == "json":
             data = await self.request("status.json", params={"CMD": "inv_query"})
-            return Device.from_json(data)
+            return Device.from_json(json.loads(data))
         if self.source_type == "html":
             data = await self.request("status.html")
             return Device.from_html(data)
@@ -159,7 +160,7 @@ async def __aenter__(self) -> OmnikInverter:
         """
         return self
 
-    async def __aexit__(self, *_exc_info) -> None:
+    async def __aexit__(self, *_exc_info: str) -> None:
         """Async exit.
 
         Args: