Graphql query get location

pyhilo/__init__.py

@@ -1,14 +1,14 @@
 """Define the hilo package."""
 from pyhilo.api import API
-from pyhilo.devices import Devices
+from pyhilo.const import UNMONITORED_DEVICES
 from pyhilo.device import HiloDevice
+from pyhilo.device.switch import Switch
+from pyhilo.devices import Devices
 from pyhilo.event import Event
 from pyhilo.exceptions import HiloError, InvalidCredentialsError, WebsocketError
 from pyhilo.oauth2 import AuthCodeWithPKCEImplementation
 from pyhilo.util import from_utc_timestamp, time_diff
 from pyhilo.websocket import WebsocketEvent
-from pyhilo.const import UNMONITORED_DEVICES
-from pyhilo.device.switch import Switch
 
 __all__ = [
     "API",

pyhilo/api.py

@@ -6,12 +6,14 @@
 import random
 import string
 import sys
-from typing import Any, Callable, Union, cast
+from typing import Any, Callable, Dict, 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,
@@ -43,6 +45,7 @@
 )
 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,
@@ -492,11 +495,11 @@ async def android_register(self) -> None:
             },
         )
 
-    async def get_location_id(self) -> int:
+    async def get_location_ids(self) -> tuple[int, str]:
         url = f"{API_AUTOMATION_ENDPOINT}/Locations"
         LOG.debug(f"LocationId URL is {url}")
         req: list[dict[str, Any]] = await self.async_request("get", url)
-        return int(req[0]["id"])
+        return (req[0]["id"], req[0]["locationHiloId"])
 
     async def get_devices(self, location_id: int) -> list[dict[str, Any]]:
         """Get list of all devices"""
@@ -510,6 +513,22 @@ 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/device/__init__.py

@@ -1,4 +1,5 @@
 """Define devices"""
+
 from __future__ import annotations
 
 from dataclasses import dataclass, field
@@ -36,6 +37,7 @@ def __init__(
     ) -> None:
         self._api = api
         self.id = 0
+        self.hilo_id: str = ""
         self.location_id = 0
         self.type = "Unknown"
         self.name = "Unknown"
@@ -59,6 +61,7 @@ def update(self, **kwargs: Dict[str, Union[str, int, Dict]]) -> None:
                     value = val.get("value")
                 reading = {
                     "deviceId": self.id,
+                    "hiloId": self.hilo_id,
                     "locationId": self.location_id,
                     "timeStampUTC": datetime.utcnow().isoformat(),
                     "value": value,
@@ -86,7 +89,8 @@ def update(self, **kwargs: Dict[str, Union[str, int, Dict]]) -> None:
             elif att == "provider":
                 att = "manufacturer"
                 new_val = HILO_PROVIDERS.get(
-                    int(val), f"Unknown ({val})")  # type: ignore
+                    int(val), f"Unknown ({val})"
+                )  # type: ignore
             else:
                 if att == "serial":
                     att = "identifier"
@@ -232,10 +236,12 @@ def __init__(self, **kwargs: Dict[str, Any]):
         #       value_type='%')
         # }
         kwargs["timeStamp"] = from_utc_timestamp(
-            kwargs.pop("timeStampUTC", ""))  # type: ignore
+            kwargs.pop("timeStampUTC", "")
+        )  # type: ignore
         self.id = 0
         self.value: Union[int, bool, str] = 0
         self.device_id = 0
+        self.hilo_id: str = ""
         self.device_attribute: DeviceAttribute
         self.__dict__.update({camel_to_snake(k): v for k, v in kwargs.items()})
         self.unit_of_measurement = (
@@ -244,8 +250,7 @@ def __init__(self, **kwargs: Dict[str, Any]):
             else ""
         )
         if not self.device_attribute:
-            LOG.warning(
-                f"Received invalid reading for {self.device_id}: {kwargs}")
+            LOG.warning(f"Received invalid reading for {self.device_id}: {kwargs}")
 
     def __repr__(self) -> str:
         return f"<Reading {self.device_attribute.attr} {self.value}{self.unit_of_measurement}>"

pyhilo/device/climate.py

@@ -1,5 +1,6 @@
 """Climate object."""
 from __future__ import annotations
+
 from typing import Any, cast
 
 from pyhilo import API
@@ -15,12 +16,14 @@ class Climate(HiloDevice):
     devices such as thermostats.
     """
 
-    def __init__(self, api: API, **kwargs: dict[str, str | int | dict[Any, Any]]) -> None:
+    def __init__(
+        self, api: API, **kwargs: dict[str, str | int | dict[Any, Any]]
+    ) -> None:
         """Initialize the Climate object.
 
-            Args:
-                api: The Hilo API instance.
-                **kwargs: Keyword arguments containing device data.
+        Args:
+            api: The Hilo API instance.
+            **kwargs: Keyword arguments containing device data.
         """
         super().__init__(api, **kwargs)
         LOG.debug("Setting up Climate device: %s", self.name)