From 843adc5520a3a7bfb891828ea341c79e8e8efbdb Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 22:15:02 -0700 Subject: [PATCH] refactor(homeassistant,sms,google_chat): HA template table for state descriptions; drop single-use wrappers --- plugins/platforms/google_chat/oauth.py | 23 +++------ plugins/platforms/homeassistant/adapter.py | 57 ++++++++-------------- plugins/platforms/sms/adapter.py | 5 +- 3 files changed, 29 insertions(+), 56 deletions(-) diff --git a/plugins/platforms/google_chat/oauth.py b/plugins/platforms/google_chat/oauth.py index 4574cc189f..6f0f7641b6 100644 --- a/plugins/platforms/google_chat/oauth.py +++ b/plugins/platforms/google_chat/oauth.py @@ -65,11 +65,6 @@ _REQUIRED_PACKAGES = [ _REDIRECT_URI = "http://localhost:1" -def _hermes_home() -> Path: - """Resolve HERMES_HOME at call time (late-binding for tests / profile switches).""" - return get_hermes_home() - - def _sanitize_email(email: str) -> str: cleaned = _EMAIL_FS_RE.sub("_", (email or "").strip().lower()) return cleaned or "_unknown_" @@ -81,27 +76,25 @@ def _token_rel(email: Optional[str]) -> str: def _user_tokens_dir() -> Path: - return _hermes_home() / "google_chat_user_tokens" + return get_hermes_home() / "google_chat_user_tokens" def _token_path(email: Optional[str] = None) -> Path: """Per-user token path for ``email``, or the legacy single-user path.""" - return _hermes_home() / _token_rel(email) + return get_hermes_home() / _token_rel(email) def _client_secret_path() -> Path: - return _hermes_home() / "google_chat_user_client_secret.json" + return get_hermes_home() / "google_chat_user_client_secret.json" def _pending_auth_path(email: Optional[str] = None) -> Path: if email: - return _hermes_home() / "google_chat_user_oauth_pending" / f"{_sanitize_email(email)}.json" - return _hermes_home() / "google_chat_user_oauth_pending.json" + return get_hermes_home() / "google_chat_user_oauth_pending" / f"{_sanitize_email(email)}.json" + return get_hermes_home() / "google_chat_user_oauth_pending.json" -# ============================================================================= -# Library API — called from the adapter at runtime -# ============================================================================= +# -- Library API — called from the adapter at runtime ------------------------- def _refresh_and_persist(creds: Any, token_path: Path, request_cls: Any, *, failure_msg: str) -> Optional[Any]: @@ -192,9 +185,7 @@ def _persist_credentials(creds: Any, token_path: Path) -> None: logger.debug("[google_chat_user_oauth] failed to persist credentials at %s", token_path, exc_info=True) -# ============================================================================= -# CLI commands — driven by the agent via /setup-files -# ============================================================================= +# -- CLI commands — driven by the agent via /setup-files ---------------------- def _normalize_authorized_user_payload(payload: dict) -> dict: diff --git a/plugins/platforms/homeassistant/adapter.py b/plugins/platforms/homeassistant/adapter.py index a358a7b6de..a19b704887 100644 --- a/plugins/platforms/homeassistant/adapter.py +++ b/plugins/platforms/homeassistant/adapter.py @@ -14,7 +14,7 @@ import os import time import uuid from datetime import datetime -from typing import Any, Callable, Dict, Optional, Set +from typing import Any, Dict, Optional, Set try: import aiohttp @@ -52,38 +52,21 @@ def _triggered(val: str) -> str: return "triggered" if val == "on" else "cleared" -def _describe_climate(name, old_val, new_val, attrs) -> str: - temp = attrs.get("current_temperature", "?") - target = attrs.get("temperature", "?") - return ( - f"[Home Assistant] {name}: HVAC mode changed from " - f"'{old_val}' to '{new_val}' (current: {temp}, target: {target})" - ) - - -def _describe_sensor(name, old_val, new_val, attrs) -> str: - unit = attrs.get("unit_of_measurement", "") - return f"[Home Assistant] {name}: changed from {old_val}{unit} to {new_val}{unit}" - - -def _describe_on_off(name, old_val, new_val, attrs) -> str: - return f"[Home Assistant] {name}: turned {_on_off(new_val)}" - - -# domain -> (name, old_val, new_val, attrs) -> human-readable description -_DOMAIN_DESCRIBERS: Dict[str, Callable[..., str]] = { - "climate": _describe_climate, - "sensor": _describe_sensor, - "binary_sensor": lambda name, old_val, new_val, attrs: ( - f"[Home Assistant] {name}: {_triggered(new_val)} (was {_triggered(old_val)})" - ), - "light": _describe_on_off, - "switch": _describe_on_off, - "fan": _describe_on_off, - "alarm_control_panel": lambda name, old_val, new_val, attrs: ( - f"[Home Assistant] {name}: alarm state changed from '{old_val}' to '{new_val}'" +# domain -> description template; see ``_format_state_change`` for the fields. +_TURNED = "[Home Assistant] {name}: turned {on_off}" +_DOMAIN_TEMPLATES = { + "climate": ( + "[Home Assistant] {name}: HVAC mode changed from " + "'{old}' to '{new}' (current: {temp}, target: {target})" ), + "sensor": "[Home Assistant] {name}: changed from {old}{unit} to {new}{unit}", + "binary_sensor": "[Home Assistant] {name}: {triggered} (was {was_triggered})", + "light": _TURNED, + "switch": _TURNED, + "fan": _TURNED, + "alarm_control_panel": "[Home Assistant] {name}: alarm state changed from '{old}' to '{new}'", } +_DEFAULT_TEMPLATE = "[Home Assistant] {name} ({entity_id}): changed from '{old}' to '{new}'" class HomeAssistantAdapter(BasePlatformAdapter): @@ -280,11 +263,13 @@ class HomeAssistantAdapter(BasePlatformAdapter): if old_val == new_val: return None attrs = new_state.get("attributes", {}) - name = attrs.get("friendly_name", entity_id) - describe = _DOMAIN_DESCRIBERS.get(_domain_of(entity_id)) - if describe is not None: - return describe(name, old_val, new_val, attrs) - return f"[Home Assistant] {name} ({entity_id}): changed from '{old_val}' to '{new_val}'" + template = _DOMAIN_TEMPLATES.get(_domain_of(entity_id), _DEFAULT_TEMPLATE) + return template.format( + name=attrs.get("friendly_name", entity_id), entity_id=entity_id, old=old_val, new=new_val, + temp=attrs.get("current_temperature", "?"), target=attrs.get("temperature", "?"), + unit=attrs.get("unit_of_measurement", ""), on_off=_on_off(new_val), + triggered=_triggered(new_val), was_triggered=_triggered(old_val), + ) # -- Outbound messaging ------------------------------------------------- diff --git a/plugins/platforms/sms/adapter.py b/plugins/platforms/sms/adapter.py index 99208cc59c..fb8e42cfa6 100644 --- a/plugins/platforms/sms/adapter.py +++ b/plugins/platforms/sms/adapter.py @@ -101,9 +101,6 @@ class SmsAdapter(BasePlatformAdapter): self._runner = None self._http_session: Optional["aiohttp.ClientSession"] = None - def _basic_auth_header(self) -> str: - return _basic_auth(self._account_sid, self._auth_token) - # -- Lifecycle ----------------------------------------------------------- async def connect(self, *, is_reconnect: bool = False) -> bool: @@ -169,7 +166,7 @@ class SmsAdapter(BasePlatformAdapter): chunks = self.truncate_message(self.format_message(content)) last_result = SendResult(success=True) url = f"{TWILIO_API_BASE}/{self._account_sid}/Messages.json" - headers = {"Authorization": self._basic_auth_header()} + headers = {"Authorization": _basic_auth(self._account_sid, self._auth_token)} session = self._http_session or _new_session(trust_env=gateway_trust_env()) try: for chunk in chunks: