Add method to retrive data for HA sensors (#11)

engrbm87 · fabaff · web-flow · commit 13fe2225ddfb · 2022-08-19T19:11:48.000+02:00
* add gitignore

* Add strict typing
Initialize self.data,
Check for bad request

* revert changes to httpx variable

* update tests

* add helper method  `get_ha_sensor_data`

* add tests

* Update docstring

Co-authored-by: Fabian Affolter &lt;mail@fabian-affolter.ch&gt;
diff --git a/glances_api/__init__.py b/glances_api/__init__.py
@@ -1,44 +1,44 @@
 """Client to interact with the Glances API."""
-import asyncio
+from __future__ import annotations
+
 import logging
+from typing import Any
 
 import httpx
 
 from . import exceptions
 
 _LOGGER = logging.getLogger(__name__)
-_RESOURCE = "{schema}://{host}:{port}/api/{version}"
 
 
-class Glances(object):
+class Glances:
     """A class for handling the data retrieval."""
 
     def __init__(
         self,
-        host="localhost",
-        port=61208,
-        version=2,
-        ssl=False,
-        verify_ssl=True,
-        username=None,
-        password=None,
-        httpx_client=None,
+        host: str = "localhost",
+        port: int = 61208,
+        version: int = 2,
+        ssl: bool = False,
+        verify_ssl: bool = True,
+        username: str | None = None,
+        password: str | None = None,
+        httpx_client: httpx.AsyncClient | None = None,
     ):
         """Initialize the connection."""
         schema = "https" if ssl else "http"
-        self.url = _RESOURCE.format(
-            schema=schema, host=host, port=port, version=version
-        )
-        self.plugins = None
-        self.values = None
+        self.url = f"{schema}://{host}:{port}/api/{version}"
+        self.data: dict[str, Any] = {}
+        self.plugins: list[str] = []
+        self.values: Any | None = None
         self.username = username
         self.password = password
         self.verify_ssl = verify_ssl
         self.httpx_client = httpx_client
 
-    async def get_data(self, endpoint):
+    async def get_data(self, endpoint: str) -> None:
         """Retrieve the data."""
-        url = "{}/{}".format(self.url, endpoint)
+        url = f"{self.url}/{endpoint}"
 
         httpx_client = (
             self.httpx_client
@@ -49,10 +49,10 @@ async def get_data(self, endpoint):
         try:
             async with httpx_client as client:
                 if self.password is None:
-                    response = await client.get(str(url))
+                    response = await client.get(url)
                 else:
                     response = await client.get(
-                        str(url), auth=(self.username, self.password)
+                        url, auth=(self.username, self.password)
                     )
         except (httpx.ConnectError, httpx.TimeoutException):
             raise exceptions.GlancesApiConnectionError(f"Connection to {url} failed")
@@ -61,21 +61,24 @@ async def get_data(self, endpoint):
             raise exceptions.GlancesApiAuthorizationError(
                 "Please check your credentials"
             )
-
+        if response.status_code == httpx.codes.BAD_REQUEST:
+            raise exceptions.GlancesApiNoDataAvailable(
+                f"endpoint: '{endpoint}' is not valid"
+            )
         if response.status_code == httpx.codes.OK:
             try:
                 _LOGGER.debug(response.json())
                 if endpoint == "all":
                     self.data = response.json()
-                if endpoint == "pluginslist":
+                elif endpoint == "pluginslist":
                     self.plugins = response.json()
             except TypeError:
                 _LOGGER.error("Can not load data from Glances")
                 raise exceptions.GlancesApiConnectionError(
                     "Unable to get the data from Glances"
                 )
 
-    async def get_metrics(self, element):
+    async def get_metrics(self, element: str) -> None:
         """Get all the metrics for a monitored element."""
         await self.get_data("all")
         await self.get_data("pluginslist")
@@ -84,3 +87,67 @@ async def get_metrics(self, element):
             self.values = self.data[element]
         else:
             raise exceptions.GlancesApiError("Element data not available")
+
+    async def get_ha_sensor_data(self) -> dict[str, Any] | None:
+        """Create a dictionary with data for Home Assistant sensors."""
+        await self.get_data("all")
+
+        sensor_data: dict[str, Any] = {}
+
+        if disks := self.data.get("fs"):
+            sensor_data["fs"] = {}
+            for disk in disks:
+                disk_free = disk.get("free") or (disk["size"] - disk["used"])
+                sensor_data["fs"][disk["mnt_point"]] = {
+                    "disk_use": round(disk["used"] / 1024**3, 1),
+                    "disk_use_percent": disk["percent"],
+                    "disk_free": round(disk_free / 1024**3, 1),
+                }
+        if data := self.data.get("sensors"):
+            sensor_data["sensors"] = {}
+            for sensor in data:
+                sensor_data["sensors"][sensor["label"]] = {
+                    sensor["type"]: sensor["value"]
+                }
+        if data := self.data.get("mem"):
+            sensor_data["mem"] = {
+                "memory_use_percent": data["percent"],
+                "memory_use": round(data["used"] / 1024**2, 1),
+                "memory_free": round(data["free"] / 1024**2, 1),
+            }
+        if data := self.data.get("memswap"):
+            sensor_data["memswap"] = {
+                "swap_use_percent": data["percent"],
+                "swap_use": round(data["used"] / 1024**3, 1),
+                "swap_free": round(data["free"] / 1024**3, 1),
+            }
+        if data := self.data.get("load"):
+            sensor_data["load"] = {
+                "processor_load": data.get("min15")
+                or self.data["cpu"]["total"]  # to be checked
+            }
+        if data := self.data.get("processcount"):
+            sensor_data["processcount"] = {
+                "process_running": data["running"],
+                "process_total": data["total"],
+                "process_thread": data["thread"],
+                "process_sleeping": data["sleeping"],
+            }
+        if data := self.data.get("quicklook"):
+            sensor_data["cpu"] = {"cpu_use_percent": data["cpu"]}
+        if "docker" in self.data and (data := self.data["docker"].get("containers")):
+            active_containers = [
+                container for container in data if container["Status"] == "running"
+            ]
+            sensor_data["docker"] = {"docker_active": len(active_containers)}
+            cpu_use = 0.0
+            for container in active_containers:
+                cpu_use += container["cpu"]["total"]
+            sensor_data["docker"]["docker_cpu_use"] = round(cpu_use, 1)
+            mem_use = 0.0
+            for container in active_containers:
+                mem_use += container["memory"]["usage"]
+            sensor_data["docker"]["docker_memory_use"] = round(mem_use / 1024**2, 1)
+        if data := self.data.get("raid"):
+            sensor_data["raid"] = data
+        return sensor_data
diff --git a/tests/test_responses.py b/tests/test_responses.py
@@ -1,8 +1,8 @@
 """Test the interaction with the Glances API."""
 import pytest
-from pytest_httpx import HTTPXMock
-
 from glances_api import Glances
+from glances_api.exceptions import GlancesApiNoDataAvailable
+from pytest_httpx import HTTPXMock
 
 PLUGINS_LIST_RESPONSE = [
     "alert",
@@ -37,6 +37,85 @@
             "key": "disk_name",
         },
     ],
+    "docker": {
+        "containers": [
+            {
+                "key": "name",
+                "name": "container1",
+                "Status": "running",
+                "cpu": {"total": 50.94973493230174},
+                "cpu_percent": 50.94973493230174,
+                "memory": {
+                    "usage": 1120321536,
+                    "limit": 3976318976,
+                    "rss": 480641024,
+                    "cache": 580915200,
+                    "max_usage": 1309597696,
+                },
+                "memory_usage": 539406336,
+            },
+            {
+                "key": "name",
+                "name": "container2",
+                "Status": "running",
+                "cpu": {"total": 26.23567931034483},
+                "cpu_percent": 26.23567931034483,
+                "memory": {
+                    "usage": 85139456,
+                    "limit": 3976318976,
+                    "rss": 33677312,
+                    "cache": 35012608,
+                    "max_usage": 87650304,
+                },
+                "memory_usage": 50126848,
+            },
+        ]
+    },
+    "fs": [
+        {
+            "device_name": "/dev/sda8",
+            "fs_type": "ext4",
+            "mnt_point": "/ssl",
+            "size": 511320748032,
+            "used": 32910458880,
+            "free": 457917374464,
+            "percent": 6.7,
+            "key": "mnt_point",
+        },
+        {
+            "device_name": "/dev/sda8",
+            "fs_type": "ext4",
+            "mnt_point": "/media",
+            "size": 511320748032,
+            "used": 32910458880,
+            "free": 457917374464,
+            "percent": 6.7,
+            "key": "mnt_point",
+        },
+    ],
+    "mem": {
+        "total": 3976318976,
+        "available": 2878337024,
+        "percent": 27.6,
+        "used": 1097981952,
+        "free": 2878337024,
+        "active": 567971840,
+        "inactive": 1679704064,
+        "buffers": 149807104,
+        "cached": 1334816768,
+        "shared": 1499136,
+    },
+    "sensors": [
+        {
+            "label": "cpu_thermal 1",
+            "value": 59,
+            "warning": None,
+            "critical": None,
+            "unit": "C",
+            "type": "temperature_core",
+            "key": "label",
+        }
+    ],
     "system": {
         "os_name": "Linux",
         "hostname": "fedora-35",
@@ -48,16 +127,35 @@
     "uptime": "3 days, 10:25:20",
 }
 
+HA_SENSOR_DATA = {
+    "fs": {
+        "/ssl": {"disk_use": 30.7, "disk_use_percent": 6.7, "disk_free": 426.5},
+        "/media": {"disk_use": 30.7, "disk_use_percent": 6.7, "disk_free": 426.5},
+    },
+    "sensors": {"cpu_thermal 1": {"temperature_core": 59}},
+    "mem": {
+        "memory_use_percent": 27.6,
+        "memory_use": 1047.1,
+        "memory_free": 2745.0,
+    },
+    "docker": {
+        "docker_active": 2,
+        "docker_cpu_use": 77.2,
+        "docker_memory_use": 1149.6
+    }
+}
+
 
 @pytest.mark.asyncio
 async def test_non_existing_endpoint(httpx_mock: HTTPXMock):
     """Test a non-exisiting endpoint."""
-    httpx_mock.add_response(json={})
+    httpx_mock.add_response(status_code=400)
 
     client = Glances()
-    await client.get_data("some-data")
+    with pytest.raises(GlancesApiNoDataAvailable):
+        await client.get_data("some-data")
 
-    assert client.values == None
+    assert not client.data
 
 
 @pytest.mark.asyncio
@@ -81,3 +179,14 @@ async def test_exisiting_endpoint(httpx_mock: HTTPXMock):
 
     assert client.values["total"] == 10.6
     assert client.values["system"] == 2.1
+
+
+@pytest.mark.asyncio
+async def test_ha_sensor_data(httpx_mock: HTTPXMock):
+    """Test the return value for ha sensors."""
+    httpx_mock.add_response(json=RESPONSE)
+
+    client = Glances()
+    result = await client.get_ha_sensor_data()
+
+    assert result == HA_SENSOR_DATA
