"""Hubject API client utilities. This module encapsulates loading the Hubject specific configuration block from ``config.json`` and provides a small asynchronous client for calling the Hubject OICP endpoints. The client is intentionally lightweight so it can be re-used by both the OCPP server and broker without introducing heavy third-party dependencies beyond ``aiohttp`` which is already part of the project. """ from __future__ import annotations import asyncio import json import logging import ssl from datetime import datetime, timezone from contextlib import asynccontextmanager from dataclasses import dataclass from pathlib import Path from typing import Any, Mapping, MutableMapping from urllib.parse import urljoin, urlparse import aiohttp try: import aiomysql # type: ignore[import] except ModuleNotFoundError: # pragma: no cover - optional dependency aiomysql = None # type: ignore[assignment] logger = logging.getLogger(__name__) AIOMYSQL_AVAILABLE = aiomysql is not None if not AIOMYSQL_AVAILABLE: logger.warning( "aiomysql is not installed; Hubject export logging will be disabled" ) __all__ = [ "HubjectClient", "HubjectClientError", "HubjectConfig", "HubjectConfigurationError", "create_ssl_context", "load_hubject_config", ] CONFIG_FILE = "config.json" try: _raw_config = json.loads(Path(CONFIG_FILE).read_text(encoding="utf-8")) except FileNotFoundError: _raw_config = {} MYSQL_CONFIG = _raw_config.get( "mysql", { "host": "127.0.0.1", "user": "root", "password": "", "db": "op", "charset": "utf8", }, ) _HUBJECT_EXPORTS_TABLE_SQL = """ CREATE TABLE IF NOT EXISTS op_hubject_exports ( id INT AUTO_INCREMENT PRIMARY KEY, station_id VARCHAR(255), transaction_id VARCHAR(255), payload JSON, success TINYINT(1), response_status INT, response_body TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) """ _exports_table_ready = False _exports_table_lock: asyncio.Lock | None = None @asynccontextmanager async def db_connection(): if not AIOMYSQL_AVAILABLE: raise RuntimeError("aiomysql is required for Hubject export logging") conn = await aiomysql.connect(**MYSQL_CONFIG) # type: ignore[union-attr] try: yield conn finally: conn.close() async def _ensure_exports_table(conn: Any) -> None: global _exports_table_ready, _exports_table_lock if _exports_table_ready: return if _exports_table_lock is None: _exports_table_lock = asyncio.Lock() async with _exports_table_lock: if _exports_table_ready: return async with conn.cursor() as cur: await cur.execute(_HUBJECT_EXPORTS_TABLE_SQL) await conn.commit() _exports_table_ready = True class HubjectConfigurationError(RuntimeError): """Raised when the Hubject configuration is invalid or incomplete.""" class HubjectClientError(RuntimeError): """Raised when an HTTP error is returned by the Hubject platform.""" def __init__(self, status: int, message: str | None = None) -> None: super().__init__(message or f"Hubject request failed with status {status}") self.status = status @dataclass(slots=True) class HubjectConfig: """Configuration required for the Hubject API client.""" api_base_url: str authorization: str | None timeout: int client_cert: Path | None client_key: Path | None ca_bundle: Path | None operator_id: str | None = None def _as_path(value: str | Path | None, base_dir: Path) -> Path | None: if value is None: return None candidate = Path(value) if not candidate.is_absolute(): candidate = base_dir / candidate return candidate def _require_file(path: Path | None, description: str) -> Path | None: if path is None: return None if not path.exists(): raise HubjectConfigurationError(f"The {description} '{path}' does not exist") if not path.is_file(): raise HubjectConfigurationError(f"The {description} '{path}' is not a file") return path def load_hubject_config( config: Mapping[str, Any], *, config_dir: Path | None = None, certs_dir: Path | None = None, ) -> HubjectConfig: """Load and validate the Hubject configuration block. Parameters ---------- config: Mapping representing the JSON configuration. Typically the result of ``json.load`` on ``config.json``. config_dir: Base directory used to resolve relative certificate paths. If ``None`` the current working directory is used. certs_dir: Optional directory containing TLS material. Defaults to ``config_dir`` joined with ``"certs"`` if unspecified. Returns ------- HubjectConfig Normalised configuration dataclass instance. Raises ------ HubjectConfigurationError If the configuration is missing required fields or references missing certificate files. """ try: hubject_cfg = config["hubject"] except KeyError as exc: raise HubjectConfigurationError("Hubject configuration block missing") from exc if not isinstance(hubject_cfg, Mapping): raise HubjectConfigurationError("Hubject configuration must be a mapping") cfg_dir = Path(config_dir or Path.cwd()) certs_base = Path(certs_dir) if certs_dir is not None else cfg_dir / "certs" api_base_url = str(hubject_cfg.get("api_base_url", "")).strip() if not api_base_url: raise HubjectConfigurationError("hubject.api_base_url must be configured") parsed = urlparse(api_base_url) if parsed.scheme not in {"https", "http"} or not parsed.netloc: raise HubjectConfigurationError( "hubject.api_base_url must be an absolute HTTP(S) URL" ) # Prefer secure communication and make it explicit in the logs if HTTP is used. if parsed.scheme != "https": logging.warning("Hubject base URL %s does not use HTTPS", api_base_url) timeout_raw = hubject_cfg.get("timeout", 30) try: timeout = int(timeout_raw) except (TypeError, ValueError) as exc: raise HubjectConfigurationError("hubject.timeout must be an integer") from exc if timeout <= 0: raise HubjectConfigurationError("hubject.timeout must be positive") authorization = hubject_cfg.get("authorization") if authorization is not None: authorization = str(authorization) operator_id = hubject_cfg.get("operator_id") if operator_id is not None: operator_id = str(operator_id).strip() or None client_cert = _as_path(hubject_cfg.get("client_cert"), certs_base) client_key = _as_path(hubject_cfg.get("client_key"), certs_base) ca_bundle = _as_path(hubject_cfg.get("ca_bundle"), certs_base) if (client_cert is None) ^ (client_key is None): raise HubjectConfigurationError( "hubject.client_cert and hubject.client_key must be provided together" ) client_cert = _require_file(client_cert, "client certificate") client_key = _require_file(client_key, "client private key") ca_bundle = _require_file(ca_bundle, "CA bundle") return HubjectConfig( api_base_url=api_base_url.rstrip("/") + "/", authorization=authorization, timeout=timeout, client_cert=client_cert, client_key=client_key, ca_bundle=ca_bundle, operator_id=operator_id, ) def create_ssl_context(config: HubjectConfig) -> ssl.SSLContext | None: """Create an SSL context using the Hubject TLS configuration.""" if not any((config.client_cert, config.client_key, config.ca_bundle)): return None context = ssl.create_default_context() if config.ca_bundle is not None: context.load_verify_locations(cafile=str(config.ca_bundle)) if config.client_cert and config.client_key: context.load_cert_chain(str(config.client_cert), str(config.client_key)) return context def _operator_id_for_path(operator_id: str) -> str: """Return the Hubject operator identifier formatted for path usage.""" return operator_id.replace(" ", "") class HubjectClient: """Small helper around the Hubject HTTP API.""" def __init__(self, config: HubjectConfig) -> None: self._config = config self._ssl_context = create_ssl_context(config) self._log = logging.getLogger(self.__class__.__name__) @property def config(self) -> HubjectConfig: return self._config def _prepare_headers(self, headers: MutableMapping[str, str] | None = None) -> dict[str, str]: prepared: dict[str, str] = {"Content-Type": "application/json"} if headers: prepared.update(headers) if self._config.authorization: prepared.setdefault("Authorization", self._config.authorization) return prepared def _build_url(self, operation: str) -> str: operation = operation.strip() if not operation: raise ValueError("operation must not be empty") return urljoin(self._config.api_base_url, operation.lstrip("/")) def _cdr_path(self, operator_id: str | None = None) -> str: operator = (operator_id or self._config.operator_id or "").strip() if not operator: raise ValueError("operator_id must be provided to upload CDRs") return f"cdrmgmt/v22/operators/{_operator_id_for_path(operator)}/charge-detail-record" async def _request( self, operation: str, payload: Mapping[str, Any], *, method: str = "POST", headers: MutableMapping[str, str] | None = None, return_raw_text: bool = False, ) -> Any: url = self._build_url(operation) timeout = aiohttp.ClientTimeout(total=self._config.timeout) async with aiohttp.ClientSession(timeout=timeout) as session: async with session.request( method, url, json=dict(payload), headers=self._prepare_headers(headers), ssl=self._ssl_context, ) as response: raw_body = await response.text() if response.status >= 400: raise HubjectClientError(response.status, raw_body) data: Any if response.content_type == "application/json": try: data = json.loads(raw_body) except json.JSONDecodeError: data = raw_body else: data = raw_body if return_raw_text: return data, response.status, raw_body return data async def call_operation(self, operation: str, payload: Mapping[str, Any]) -> Any: """Call an arbitrary Hubject API operation.""" return await self._request(operation, payload, method="POST") async def authorize_start(self, payload: Mapping[str, Any]) -> Any: """Invoke the Hubject AuthorizeStart endpoint.""" return await self.call_operation("AuthorizeStart", payload) async def authorize_stop(self, payload: Mapping[str, Any]) -> Any: """Invoke the Hubject AuthorizeStop endpoint.""" return await self.call_operation("AuthorizeStop", payload) async def upload_cdr( self, payload: Mapping[str, Any], *, operator_id: str | None = None, ) -> Any: """Upload a charge detail record via the Hubject CDR endpoint.""" path = self._cdr_path(operator_id) return await self._request(path, payload, method="POST") async def push_evse_data( self, operator_id: str, payload: Mapping[str, Any], ) -> Any: """Transmit EVSE metadata to Hubject using eRoamingPushEvseData.""" operator = str(operator_id or "").strip() if not operator: raise ValueError("operator_id must be provided") request_payload: dict[str, Any] = dict(payload) request_payload.setdefault("OperatorID", operator) request_payload.pop("operator_id", None) action = request_payload.get("ActionType") if isinstance(action, str): stripped = action.strip() if stripped: request_payload["ActionType"] = stripped path = f"evsepush/v23/operators/{operator}/data-records" return await self._request(path, request_payload, method="POST") async def push_cdr( self, station_id: str, start_info: Mapping[str, Any], stop_info: Mapping[str, Any], ) -> tuple[bool, int, Any | None]: """Create and upload a Hubject-compliant CDR and persist the result.""" transaction_id = ( stop_info.get("transactionId") or start_info.get("transactionId") or start_info.get("idTag") ) session_id = str(transaction_id) if transaction_id is not None else None def _as_float(value: Any) -> float | None: try: if value is None: return None return float(value) except (TypeError, ValueError): return None meter_start = ( _as_float(start_info.get("meterStartWh")) or _as_float(start_info.get("meterStart")) ) meter_end = _as_float(stop_info.get("meterStop")) or _as_float( stop_info.get("meterStopWh") ) consumed = None if meter_start is not None and meter_end is not None: consumed = max(meter_end - meter_start, 0.0) id_token = ( start_info.get("idTokenOriginal") or start_info.get("idToken") or start_info.get("idTag") or stop_info.get("idTag") ) identification: dict[str, Any] | None = None if id_token: identification = { "RFIDMifareFamilyIdentification": { "UID": str(id_token), } } cdr_payload: dict[str, Any] = { "SessionID": session_id, "CPOPartnerSessionID": session_id, "EMPPartnerSessionID": session_id, "EVSEID": start_info.get("evseId") or station_id, "ChargingStart": start_info.get("sessionStartTimestamp") or start_info.get("timestamp"), "ChargingEnd": stop_info.get("timestamp"), "MeterValueStart": meter_start, "MeterValueEnd": meter_end, "MeterValueInBetween": [], "ConsumedEnergy": consumed, "Identification": identification, "SessionStart": start_info.get("sessionStartTimestamp") or start_info.get("timestamp"), "SessionEnd": stop_info.get("timestamp"), "IsCertifiedMeterValue": False, } if self._config.operator_id: cdr_payload.setdefault("HubOperatorID", self._config.operator_id) cdr_payload = {k: v for k, v in cdr_payload.items() if v is not None} payload = cdr_payload success = False status = 0 response_body: str | None = None response_data: Any | None = None try: response_data, status, raw_body = await self._request( self._cdr_path(), payload, method="POST", return_raw_text=True, ) success = True response_body = raw_body except HubjectClientError as exc: status = exc.status response_body = exc.args[0] if exc.args else "" logging.warning( "Hubject push_cdr failed with HTTP %s for station_id=%s", exc.status, station_id, ) except Exception: response_body = None logging.exception( "Unexpected error while pushing Hubject CDR for station_id=%s", station_id, ) truncated_body = (response_body or "")[:1000] if AIOMYSQL_AVAILABLE: try: async with db_connection() as conn: await _ensure_exports_table(conn) async with conn.cursor() as cur: await cur.execute( """ INSERT INTO op_hubject_exports (station_id, transaction_id, payload, success, response_status, response_body) VALUES (%s, %s, %s, %s, %s, %s) """, ( station_id, session_id, json.dumps(payload, ensure_ascii=False), 1 if success else 0, status, truncated_body, ), ) await conn.commit() except Exception: logging.debug( "Failed to log Hubject CDR upload for station_id=%s", station_id, exc_info=True, ) else: logging.debug( "aiomysql not available; skipping Hubject CDR log for station_id=%s", station_id, ) return success, status, response_data def _build_evse_status_payload( self, evse_id: str, status: str, timestamp: datetime | None = None, ) -> dict[str, Any]: if not evse_id or not str(evse_id).strip(): raise ValueError("evse_id must be provided") status_text = str(status or "").strip() if not status_text: raise ValueError("status must be provided") ts = timestamp or datetime.now(timezone.utc) if ts.tzinfo is None: ts = ts.replace(tzinfo=timezone.utc) else: ts = ts.astimezone(timezone.utc) payload: dict[str, Any] = { "ActionType": "update", "EVSEStatusRecords": [ { "EvseID": str(evse_id), "EvseStatus": status_text, "LastUpdate": ts.isoformat().replace("+00:00", "Z"), } ], } if self._config.operator_id: payload["OperatorID"] = self._config.operator_id return payload async def set_evse_status( self, evse_id: str, status: str, *, timestamp: datetime | None = None, ) -> tuple[Any, int, str]: payload = self._build_evse_status_payload(evse_id, status, timestamp) data, status_code, raw_body = await self._request( "/evse/status/record", payload, method="POST", return_raw_text=True, ) self._log.info( "Hubject EVSE status %s for %s -> HTTP %s", payload["EVSEStatusRecords"][0]["EvseStatus"], payload["EVSEStatusRecords"][0]["EvseID"], status_code, ) return data, status_code, raw_body async def set_evse_occupied( self, evse_id: str, *, timestamp: datetime | None = None ) -> tuple[Any, int, str]: return await self.set_evse_status(evse_id, "Occupied", timestamp=timestamp) async def set_evse_available( self, evse_id: str, *, timestamp: datetime | None = None ) -> tuple[Any, int, str]: return await self.set_evse_status(evse_id, "Available", timestamp=timestamp) async def load_client_from_file( config_path: str | Path = "config.json", *, certs_dir: str | Path | None = None ) -> HubjectClient: """Convenience helper that instantiates :class:`HubjectClient` from disk.""" path = Path(config_path) config_data = json.loads(path.read_text(encoding="utf-8")) cfg = load_hubject_config(config_data, config_dir=path.parent, certs_dir=certs_dir) return HubjectClient(cfg) def sync_load_client_from_file( config_path: str | Path = "config.json", *, certs_dir: str | Path | None = None ) -> HubjectClient: """Synchronous wrapper around :func:`load_client_from_file` for convenience.""" return asyncio.get_event_loop().run_until_complete( load_client_from_file(config_path, certs_dir=certs_dir) )