diff --git a/pyhilo/api.py b/pyhilo/api.py index 9c03d4f..958c80d 100755 --- a/pyhilo/api.py +++ b/pyhilo/api.py @@ -393,6 +393,7 @@ async def refresh_ws_token(self) -> None: await self.websocket_manager.refresh_token(self.websocket_manager.challengehub) async def get_websocket_params(self) -> None: + """Retrieves and constructs WebSocket connection parameters from the negotiation endpoint.""" uri = parse.urlparse(self.ws_url) LOG.debug("Getting websocket params") LOG.debug(f"Getting uri {uri}") @@ -415,10 +416,11 @@ async def get_websocket_params(self) -> None: "available_transports": transport_dict, "full_ws_url": self.full_ws_url, } - LOG.debug("Calling set_state websocket_params") + LOG.debug("Calling set_state from get_websocket_params") await set_state(self._state_yaml, "websocket", websocket_dict) async def fb_install(self, fb_id: str) -> None: + """Registers a Firebase installation and stores the authentication token state.""" LOG.debug("Posting firebase install") body = { "fid": fb_id, @@ -441,7 +443,7 @@ async def fb_install(self, fb_id: str) -> None: raise RequestError(err) from err LOG.debug(f"FB Install data: {resp}") auth_token = resp.get("authToken", {}) - LOG.debug("Calling set_state fb_install") + LOG.debug("Calling set_state from fb_install") await set_state( self._state_yaml, "firebase", @@ -493,6 +495,7 @@ async def android_register(self) -> None: ) async def get_location_ids(self) -> tuple[int, str]: + """Gets location id from an API call""" url = f"{API_AUTOMATION_ENDPOINT}/Locations" LOG.debug(f"LocationId URL is {url}") req: list[dict[str, Any]] = await self.async_request("get", url) @@ -516,6 +519,7 @@ async def _set_device_attribute( key: DeviceAttribute, value: Union[str, float, int, None], ) -> None: + """Sets device attributes""" url = self._get_url(f"Devices/{device.id}/Attributes", device.location_id) LOG.debug(f"Device Attribute URL is {url}") await self.async_request("put", url, json={key.hilo_attribute: value}) @@ -611,7 +615,7 @@ async def get_gd_events( } } """ - + # ic-dev21 need to check but this is probably dead code url = self._get_url("Events", location_id, True) if not event_id: url += "?active=true" @@ -645,6 +649,7 @@ async def get_seasons(self, location_id: int) -> dict[str, Any]: return cast(dict[str, Any], await self.async_request("get", url)) async def get_gateway(self, location_id: int) -> dict[str, Any]: + """Gets info about the Hilo hub (gateway)""" url = self._get_url("Gateways/Info", location_id) LOG.debug(f"Gateway URL is {url}") req = await self.async_request("get", url) diff --git a/pyhilo/devices.py b/pyhilo/devices.py index 1187098..f9257d8 100644 --- a/pyhilo/devices.py +++ b/pyhilo/devices.py @@ -11,6 +11,7 @@ class Devices: def __init__(self, api: API): + """Initialize.""" self._api = api self.devices: list[HiloDevice] = [] self.location_id: int = 0 @@ -37,6 +38,9 @@ def attributes_list(self) -> list[Union[int, dict[int, list[str]]]]: ] def parse_values_received(self, values: list[dict[str, Any]]) -> list[HiloDevice]: + """Places value received in a dict while removing null attributes, + this returns values to be mapped to devices. + """ readings = [] for val in values: val["device_attribute"] = self._api.dev_atts( @@ -48,6 +52,7 @@ def parse_values_received(self, values: list[dict[str, Any]]) -> list[HiloDevice def _map_readings_to_devices( self, readings: list[DeviceReading] ) -> list[HiloDevice]: + """Uses the dict from parse_values_received to map the values to devices.""" updated_devices = [] for reading in readings: device_identifier = reading.device_id @@ -65,11 +70,15 @@ def _map_readings_to_devices( return updated_devices def find_device(self, device_identifier: int | str) -> HiloDevice: + """Makes sure the devices received have an identifier, this means some need to be hardcoded + like the unknown power meter. + """ if isinstance(device_identifier, int): return next((d for d in self.devices if d.id == device_identifier), None) return next((d for d in self.devices if d.hilo_id == device_identifier), None) def generate_device(self, device: dict) -> HiloDevice: + """Generate all devices from the list received.""" device["location_id"] = self.location_id if dev := self.find_device(device["id"]): dev.update(**device) @@ -101,6 +110,7 @@ async def update(self) -> None: async def update_devicelist_from_signalr( self, values: list[dict[str, Any]] ) -> list[HiloDevice]: + #ic-dev21 not sure if this is dead code? new_devices = [] for raw_device in values: LOG.debug(f"Generating device {raw_device}") diff --git a/pyhilo/event.py b/pyhilo/event.py index d9e5d30..9afa91c 100755 --- a/pyhilo/event.py +++ b/pyhilo/event.py @@ -25,6 +25,7 @@ class Event: recovery_end: datetime def __init__(self, **event: dict[str, Any]): + """Initialize.""" self._convert_phases(cast(dict[str, Any], event.get("phases"))) params: dict[str, Any] = event.get("parameters", {}) devices: list[dict[str, Any]] = params.get("devices", []) @@ -79,6 +80,7 @@ def should_check_for_allowed_wh(self) -> bool: return 1800 <= time_since_preheat_start <= 2700 and not already_has_allowed_wh def as_dict(self) -> dict[str, Any]: + """Formats the information received as a dictionary""" rep = {k: getattr(self, k) for k in self.dict_items} rep["phases"] = {k: getattr(self, k) for k in self.phases_list} rep["state"] = self.state @@ -134,6 +136,7 @@ def invalid(self) -> bool: @property def current_phase_times(self) -> dict[str, datetime]: + """Defines timestamps for Hilo Challenge""" if self.state in ["completed", "off", "scheduled", "unknown"]: return {} phase_timestamp = self._phase_time_mapping.get(self.state, self.state) diff --git a/pyhilo/graphql.py b/pyhilo/graphql.py index f29ac6b..6b4dc62 100644 --- a/pyhilo/graphql.py +++ b/pyhilo/graphql.py @@ -529,6 +529,7 @@ async def async_init(self) -> None: }""" async def call_get_location_query(self, location_hilo_id: str) -> None: + """This functions calls the digital-twin and requests location id""" access_token = await self._get_access_token() transport = AIOHTTPTransport( url="https://platform.hiloenergie.com/api/digital-twin/v3/graphql", @@ -594,6 +595,7 @@ async def _get_access_token(self) -> str: return await self._api.async_get_access_token() def _handle_query_result(self, result: Dict[str, Any]) -> None: + """This receives query results and maps them to the proper device.""" devices_values: list[any] = result["getLocation"]["devices"] attributes = self.mapper.map_query_values(devices_values) self._devices.parse_values_received(attributes) diff --git a/pyhilo/oauth2helper.py b/pyhilo/oauth2helper.py index 7177de9..3ac0ff1 100644 --- a/pyhilo/oauth2helper.py +++ b/pyhilo/oauth2helper.py @@ -17,10 +17,12 @@ def __init__(self) -> None: # Ref : https://blog.sanghviharshit.com/reverse-engineering-private-api-oauth-code-flow-with-pkce/ def _get_code_verifier(self) -> str: + """Generates a random cryptographic key string to be used as a code verifier in PKCE.""" code = base64.urlsafe_b64encode(os.urandom(40)).decode("utf-8") return re.sub("[^a-zA-Z0-9]+", "", code) def _get_code_challenge(self, verifier: str) -> str: + """Generates a SHA-256 code challenge for PKCE""" sha_verifier = hashlib.sha256(verifier.encode("utf-8")).digest() code = base64.urlsafe_b64encode(sha_verifier).decode("utf-8") return code.replace("=", "") diff --git a/pyhilo/websocket.py b/pyhilo/websocket.py index 8446ffd..9c6dcfc 100755 --- a/pyhilo/websocket.py +++ b/pyhilo/websocket.py @@ -311,6 +311,7 @@ async def async_connect(self) -> None: schedule_callback(callback) async def _clean_queue(self) -> None: + """Removes queued tasks.""" for task in self._queued_tasks: task.cancel() @@ -368,6 +369,24 @@ async def _async_pong(self) -> None: async def async_invoke( self, arg: list, target: str, inv_id: int, inv_type: WSMsgType = WSMsgType.TEXT ) -> None: + """ + Sends an invocation message over the WebSocket connection. + + Waits for the WebSocket to be ready if it is not already, then sends a message + containing the target method, arguments, and invocation ID. + + Args: + arg (list): The list of arguments to send with the invocation. + target (str): The name of the method or action being invoked on the server. + inv_id (int): A unique identifier for this invocation message. + inv_type (WSMsgType, optional): The WebSocket message type. Defaults to WSMsgType.TEXT. + + Returns: + None + + Notes: + If the WebSocket is not ready within 10 seconds, the invocation is skipped. + """ if not self._ready: LOG.warning( f"Delaying invoke {target} {inv_id} {arg}: Websocket not ready."