refactor(homeassistant,sms,google_chat): HA template table for state descriptions; drop single-use wrappers

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:

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 -------------------------------------------------
 

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: