Merge pull request #358 from dvd-dev/urn_extract

Urn extract

Author: ic-dev21

pyhilo/api.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import asyncio
+import base64
 from datetime import datetime, timedelta
 import json
 import random
@@ -92,6 +93,7 @@ def __init__(
         self.ws_url: str = ""
         self.ws_token: str = ""
         self.endpoint: str = ""
+        self._urn: str | None = None
 
     @classmethod
     async def async_create(
@@ -143,8 +145,37 @@ async def async_get_access_token(self) -> str:
         access_token = str(self._oauth_session.token["access_token"])
         LOG.debug("Websocket access token is %s", access_token)
 
+        urn = self.urn
+        LOG.debug("Extracted URN: %s", urn)
+
         return str(self._oauth_session.token["access_token"])
 
+    @property
+    def urn(self) -> str | None:
+        """Extract URN from the JWT access token."""
+        try:
+            if not self._oauth_session.valid_token:
+                return None
+            token = self._oauth_session.token["access_token"]
+            payload_part = token.split(".")[1]
+            # Add padding if necessary
+            padding = 4 - len(payload_part) % 4
+            if padding != 4:
+                payload_part += "=" * padding
+
+            decoded = base64.urlsafe_b64decode(payload_part)
+            claims = json.loads(decoded)
+            urn_claim = claims.get("urn:com:hiloenergie:profile:location_hilo_id")
+            if urn_claim and isinstance(urn_claim, list) and len(urn_claim) > 0:
+                self._urn = urn_claim[0]  # Get the first URN from the array
+            else:
+                self._urn = None
+                
+            return self._urn
+        except (IndexError, json.JSONDecodeError, KeyError):
+            LOG.error("Failed to extract URN from access token")
+            return None
+
     def dev_atts(
         self, attribute: str, value_type: Union[str, None] = None
     ) -> Union[DeviceAttribute, str]:

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.12.03"
+PYHILO_VERSION: Final = "2025.12.04"
 # TODO: Find a way to keep previous line in sync with pyproject.toml automatically
 
 CONTENT_TYPE_FORM: Final = "application/x-www-form-urlencoded"

pyproject.toml

@@ -40,7 +40,7 @@ exclude = ".venv/.*"
 
 [tool.poetry]
 name = "python-hilo"
-version = "2025.12.3"
+version = "2025.12.4"
 description = "A Python3, async interface to the Hilo API"
 readme = "README.md"
 authors = ["David Vallee Delisle <[email protected]>"]