diff --git a/.github/workflows/lint.yaml b/.github/workflows/lint.yaml index 9e1259a..23bdcd8 100644 --- a/.github/workflows/lint.yaml +++ b/.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: diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 3797bfe..1b2c499 100644 --- a/.pre-commit-config.yaml +++ b/.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: diff --git a/example.py b/example.py index 9d35629..8f20647 100644 --- a/example.py +++ b/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 diff --git a/pyopenhardwaremonitor/api.py b/pyopenhardwaremonitor/api.py index 2afbf2f..b958b81 100644 --- a/pyopenhardwaremonitor/api.py +++ b/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.""" diff --git a/pyopenhardwaremonitor/types.py b/pyopenhardwaremonitor/types.py new file mode 100644 index 0000000..ffec11b --- /dev/null +++ b/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