Merge pull request #265 from AbigailAsselin-Hilo/add_graphql_subscription

Add graphql subscription

Author: ic-dev21

pyhilo/api.py

@@ -6,14 +6,12 @@
 import random
 import string
 import sys
-from typing import Any, Callable, Dict, Union, cast
+from typing import Any, Callable, Union, cast
 from urllib import parse
 
 from aiohttp import ClientSession
 from aiohttp.client_exceptions import ClientResponseError
 import backoff
-from gql import Client, gql
-from gql.transport.aiohttp import AIOHTTPTransport
 
 from pyhilo.const import (
     ANDROID_CLIENT_ENDPOINT,
@@ -45,7 +43,6 @@
 )
 from pyhilo.device import DeviceAttribute, HiloDevice, get_device_attributes
 from pyhilo.exceptions import InvalidCredentialsError, RequestError
-from pyhilo.graphql import GraphQlHelper
 from pyhilo.util.state import (
     StateDict,
     WebsocketDict,
@@ -513,22 +510,6 @@ async def get_devices(self, location_id: int) -> list[dict[str, Any]]:
             devices.append(callback())
         return devices
 
-    async def call_get_location_query(self, location_hilo_id: string) -> Dict[str, Any]:
-        access_token = await self.async_get_access_token()
-        transport = AIOHTTPTransport(
-            url="https://platform.hiloenergie.com/api/digital-twin/v3/graphql",
-            headers={"Authorization": f"Bearer {access_token}"},
-        )
-        client = Client(transport=transport, fetch_schema_from_transport=True)
-        query = gql(GraphQlHelper.query_get_location())
-
-        async with client as session:
-            result = await session.execute(
-                query, variable_values={"locationHiloId": location_hilo_id}
-            )
-        LOG.info(result)
-        return result
-
     async def _set_device_attribute(
         self,
         device: HiloDevice,

pyhilo/const.py

@@ -7,7 +7,7 @@
 LOG: Final = logging.getLogger(__package__)
 DEFAULT_STATE_FILE: Final = "hilo_state.yaml"
 REQUEST_RETRY: Final = 9
-PYHILO_VERSION: Final = "2025.2.02"
+PYHILO_VERSION: Final = "2025.4.01"
 # TODO: Find a way to keep previous line in sync with pyproject.toml automatically
 
 CONTENT_TYPE_FORM: Final = "application/x-www-form-urlencoded"

pyhilo/device/__init__.py

@@ -87,9 +87,7 @@ def update(self, **kwargs: Dict[str, Union[str, int, Dict]]) -> None:
                     new_val.append(DeviceAttribute("Disconnected", "null"))
             elif att == "provider":
                 att = "manufacturer"
-                new_val = HILO_PROVIDERS.get(
-                    int(val), f"Unknown ({val})"
-                )  # type: ignore
+                new_val = HILO_PROVIDERS.get(int(val), f"Unknown ({val})")  # type: ignore
             else:
                 if att == "serial":
                     att = "identifier"
@@ -234,9 +232,7 @@ def __init__(self, **kwargs: Dict[str, Any]):
         #       attr='intensity',
         #       value_type='%')
         # }
-        kwargs["timeStamp"] = from_utc_timestamp(
-            kwargs.pop("timeStampUTC", "")
-        )  # type: ignore
+        kwargs["timeStamp"] = from_utc_timestamp(kwargs.pop("timeStampUTC", ""))  # type: ignore
         self.id = 0
         self.value: Union[int, bool, str] = 0
         self.device_id = 0