Adding typing and get_sensor_nodes()

.github/workflows/lint.yaml

@@ -25,11 +25,14 @@ jobs:
         run: python3 -m pip install -r requirements.txt
 
       - name: "Install linters"
-        run: python3 -m pip install ruff yamllint
+        run: python3 -m pip install ruff yamllint mypy
 
       - name: "ruff check ."
         run: python3 -m ruff check .
 
+      - name: "mypy"
+        run: mypy pyopenhardwaremonitor
+
       # - name: yaml-lint
       #   uses: ibiqlik/action-yamllint@v3
       #   with:

.pre-commit-config.yaml

@@ -17,6 +17,11 @@ repos:
         types: [file, yaml]
         files: ^((pyopenhardwaremonitor|tests)/.+)?[^/]+\.(yaml|yml)$
 
+  - repo: https://github.com/pre-commit/mirrors-mypy
+    rev: 'v1.15.0'
+    hooks:
+      - id: mypy
+
   - repo: https://github.com/sourcery-ai/sourcery
     rev: v1.34.0
     hooks:

example.py

@@ -11,7 +11,8 @@ async def main(host=None, port=8085):
 
     # Example when using `async with` syntax
     async with OpenHardwareMonitorAPI(host, port) as api:
-        data = await api.get_data()
+        # data = await api.get_data()
+        data = await api.get_sensor_nodes()
         j = json.dumps(data)
         print(j)
         return j

pyopenhardwaremonitor/api.py

@@ -7,6 +7,7 @@
 from yarl import URL
 
 from .exceptions import NotFoundError, OpenHardwareMonitorError, UnauthorizedError
+from .types import DataNode, SensorNode, SensorType
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -17,7 +18,7 @@ class OpenHardwareMonitorAPI:
     def __init__(
         self,
         host,
-        port,
+        port: int = 8085,
         loop=None,
         session=None,
         timeout=DEFAULT_TIMEOUT,
@@ -92,11 +93,53 @@ async def raw_request(  # pylint: disable=too-many-arguments
                 content = await response.read()
             _LOGGER.debug("Response %s, status: %s", response.url, response.status)
             _LOGGER.debug("Response content: %s", content)
+            response.raise_for_status()
             return content
 
-    async def get_data(self):
-        json = await self.request("data.json")
-        return json
+    async def get_data(self) -> DataNode:
+        """Get data-tree from OHM remote server."""
+        return await self.request("data.json")
+
+    async def get_sensor_nodes(self) -> dict[str, list[SensorNode]]:
+        """Get the Sensor data grouped by Computer within the data-tree."""
+        root_node = await self.get_data()
+        return {
+            c["Text"]: [*OpenHardwareMonitorAPI._parse_sensor_nodes(c)]
+            for c in root_node["Children"]
+        }
+
+    @staticmethod
+    def _parse_sensor_nodes(
+        node: DataNode, parent_names: list[str] | None = None
+    ) -> list[SensorNode]:
+        """Recursively loop through child objects, finding the values."""
+        result: list[SensorNode] = []
+        if parent_names is None:
+            parent_names = []
+        parent_names = [*parent_names, node["Text"]]
+
+        if node.get("Children"):
+            for n in node["Children"]:
+                sub_nodes = OpenHardwareMonitorAPI._parse_sensor_nodes(n, parent_names)
+                result.extend(sub_nodes)
+        elif node.get("Value"):
+            return [
+                SensorNode(
+                    id=node.get("id"),
+                    Text=node.get("Text"),
+                    Min=node.get("Min"),
+                    Max=node.get("Max"),
+                    Value=node.get("Value"),
+                    ImageURL=node.get("ImageURL"),
+                    # Extra
+                    SensorId=node.get("SensorId"),
+                    Type=SensorType(node.get("Type")) if node.get("Type") else None,
+                    ParentNames=parent_names,
+                    FullName=" ".join(parent_names),
+                    ComputerName=parent_names[0],
+                )
+            ]
+        return result
 
     async def close(self):
         """Close the session."""

pyopenhardwaremonitor/types.py

@@ -0,0 +1,63 @@
+"""Types for the OpenHardwareMonitor integration."""
+
+from enum import Enum
+from typing import TypedDict
+
+
+class SensorType(str, Enum):
+    """What kind of sensor is this."""
+
+    # Copied from `SensorType` enum in: https://github.com/LibreHardwareMonitor/LibreHardwareMonitor/blob/master/LibreHardwareMonitorLib/Hardware/ISensor.cs
+
+    Voltage = "Voltage"  # V
+    Current = "Current"  # A
+    Power = "Power"  # W
+    Clock = "Clock"  # MHz
+    Temperature = "Temperature"  # °C
+    Load = "Load"  # %
+    Frequency = "Frequency"  # Hz
+    Fan = "Fan"  # RPM
+    Flow = "Flow"  # L/h
+    Control = "Control"  # %
+    Level = "Level"  # %
+    Factor = "Factor"  # 1
+    Data = "Data"  # GB = 2^30 Bytes
+    SmallData = "SmallData"  # MB = 2^20 Bytes
+    Throughput = "Throughput"  # B/s
+    TimeSpan = "TimeSpan"  # Seconds
+    Energy = "Energy"  # milliwatt-hour (mWh)
+    Noise = "Noise"  # dBA
+    Conductivity = "Conductivity"  # µS/cm
+    Humidity = "Humidity"  # %
+
+
+class SensorNode(TypedDict):
+    """Describes a data point node (smallest descendant, with info about their parents)."""
+
+    id: int | None
+    Text: str | None
+    Min: str | None
+    Value: str | None
+    Max: str | None
+    ImageURL: str | None
+
+    SensorId: str | None
+    Type: SensorType | None
+
+    ComputerName: str
+    ParentNames: list[str]
+    FullName: str
+
+
+class DataNode(TypedDict):
+    """Describes a node in the data tree."""
+
+    id: int
+    Text: str
+    Min: str
+    Value: str
+    Max: str
+    ImageURL: str
+    Children: list["DataNode"]
+    SensorId: str | None
+    Type: SensorType | str | None