gib/Panama
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add Home Assistant light catalog and dimming

Browse Source
This commit is contained in:
Gabriel Brown
2026-08-17 16:01:50 -04:00
parent c42794c5e2
commit ff67387c42
4 changed files with 225 additions and 87 deletions
Download Patch File Download Diff File Expand all files Collapse all files
config/dot/quickshell/scripts/panama-home-assistant
+89 -36
View File
@@ -18,7 +18,6 @@ import urllib.parse
import urllib.request
from collections.abc import Callable, Mapping, Sequence
from dataclasses import dataclass
from typing import Any
ENV_KEYS = (
@@ -49,7 +48,7 @@ class Config:
@property
def configured(self) -> bool:
return bool(self.base_url and self.token and self.entity_ids)
return bool(self.base_url and self.token)
def compact_json(value: dict[str, object]) -> str:
@@ -226,7 +225,7 @@ def request_json(
config: Config,
method: str,
path: str,
payload: dict[str, str] | None = None,
payload: Mapping[str, object] | None = None,
) -> object:
if not config.base_url or not config.token:
raise BridgeError("not-configured")
@@ -266,71 +265,74 @@ def fallback_name(entity_id: str) -> str:
return entity_id.split(".", 1)[1].replace("_", " ").title()
def normalize_entities(
raw: list[dict[str, Any]],
configured: Sequence[str],
) -> list[dict[str, object]]:
by_id = {
str(item.get("entity_id", "")): item
for item in raw
if isinstance(item, dict)
}
def normalize_catalog(raw: Sequence[object]) -> list[dict[str, object]]:
result: list[dict[str, object]] = []
for entity_id in configured:
item = by_id.get(entity_id)
if not item:
for item in raw:
if not isinstance(item, dict):
continue
entity_id = item.get("entity_id")
attributes = item.get("attributes")
if not isinstance(attributes, dict):
if not isinstance(entity_id, str) or not entity_id.startswith("light."):
continue
if not ENTITY_ID.fullmatch(entity_id) or not isinstance(attributes, dict):
continue
state = str(item.get("state", "unavailable"))
available = state not in {"unknown", "unavailable"}
friendly_name = attributes.get("friendly_name")
name = (
friendly_name.strip()
if isinstance(friendly_name, str) and friendly_name.strip()
else fallback_name(entity_id)
active = available and state == "on"
raw_brightness = attributes.get("brightness")
brightness_pct = (
round(max(0, min(255, raw_brightness)) * 100 / 255)
if active
and isinstance(raw_brightness, (int, float))
and not isinstance(raw_brightness, bool)
else 0
)
modes = attributes.get("supported_color_modes", [])
dimmable = (
isinstance(modes, list) and any(mode != "onoff" for mode in modes)
) or isinstance(raw_brightness, (int, float))
source_name = attributes.get("friendly_name")
result.append(
{
"id": entity_id,
"name": name,
"domain": entity_id.split(".", 1)[0],
"sourceName": (
source_name.strip()
if isinstance(source_name, str) and source_name.strip()
else fallback_name(entity_id)
),
"state": state,
"available": available,
"active": available
and state not in {"off", "closed", "idle", "standby"},
"active": active,
"dimmable": dimmable,
"brightnessPct": brightness_pct,
}
)
return result
def ensure_configured(entity_id: str, configured: Sequence[str]) -> str:
if entity_id not in configured:
raise ValueError("entity-not-configured")
return entity_id
def collect_snapshot(config: Config) -> dict[str, object]:
def collect_catalog(config: Config) -> dict[str, object]:
legacy_entity_ids = list(config.entity_ids)
if not config.configured:
return {
"ok": False,
"configured": False,
"generatedAt": int(time.time()),
"entities": [],
"legacyEntityIds": legacy_entity_ids,
"error": "not-configured",
}
try:
raw = request_json(config, "GET", "/api/states")
if not isinstance(raw, list):
raise BridgeError("invalid-response")
entities = normalize_entities(raw, config.entity_ids)
entities = normalize_catalog(raw)
except BridgeError as error:
return {
"ok": False,
"configured": True,
"generatedAt": int(time.time()),
"entities": [],
"legacyEntityIds": legacy_entity_ids,
"error": str(error),
}
return {
@@ -338,10 +340,28 @@ def collect_snapshot(config: Config) -> dict[str, object]:
"configured": True,
"generatedAt": int(time.time()),
"entities": entities,
"legacyEntityIds": legacy_entity_ids,
"error": "",
}
def collect_snapshot(config: Config) -> dict[str, object]:
return collect_catalog(config)
def discovered_light_ids(config: Config) -> set[str]:
raw = request_json(config, "GET", "/api/states")
if not isinstance(raw, list):
raise BridgeError("invalid-response")
return {item["id"] for item in normalize_catalog(raw) if isinstance(item["id"], str)}
def ensure_discovered(config: Config, entity_id: str) -> str:
if entity_id not in discovered_light_ids(config):
raise ValueError("entity-not-discovered")
return entity_id
def probe(config: Config) -> dict[str, object]:
if not config.configured:
return {
@@ -368,7 +388,7 @@ def probe(config: Config) -> dict[str, object]:
def toggle(config: Config, entity_id: str) -> dict[str, object]:
entity_id = ensure_configured(entity_id, config.entity_ids)
entity_id = ensure_discovered(config, entity_id)
request_json(
config,
"POST",
@@ -378,6 +398,31 @@ def toggle(config: Config, entity_id: str) -> dict[str, object]:
return {"ok": True, "entityId": entity_id, "error": ""}
def set_brightness(
config: Config, entity_id: str, percent: int
) -> dict[str, object]:
if isinstance(percent, bool) or not isinstance(percent, int) or not 0 <= percent <= 100:
raise ValueError("invalid-brightness")
ensure_discovered(config, entity_id)
if percent == 0:
path = "/api/services/light/turn_off"
payload = {"entity_id": entity_id}
else:
path = "/api/services/light/turn_on"
payload = {"entity_id": entity_id, "brightness_pct": percent}
request_json(config, "POST", path, payload)
return {"ok": True, "entityId": entity_id, "brightnessPct": percent, "error": ""}
def parse_brightness(value: str) -> int:
if not re.fullmatch(r"(?:0|[1-9][0-9]{0,2})", value):
raise ValueError("invalid-brightness")
percent = int(value)
if percent > 100:
raise ValueError("invalid-brightness")
return percent
def open_home(config: Config) -> dict[str, object]:
if not config.base_url:
return {"ok": False, "error": "not-configured"}
@@ -409,8 +454,8 @@ def main(argv: list[str]) -> int:
if command == "probe" and len(argv) <= 1:
result = probe(config)
success = bool(result["reachable"])
elif command == "snapshot" and len(argv) == 1:
result = collect_snapshot(config)
elif command in {"catalog", "snapshot"} and len(argv) == 1:
result = collect_catalog(config)
success = bool(result["ok"])
elif command == "toggle" and len(argv) == 2:
try:
@@ -420,6 +465,14 @@ def main(argv: list[str]) -> int:
except BridgeError as error:
result = {"ok": False, "error": str(error)}
success = bool(result["ok"])
elif command == "brightness" and len(argv) == 3:
try:
result = set_brightness(config, argv[1], parse_brightness(argv[2]))
except ValueError as error:
result = {"ok": False, "error": str(error)}
except BridgeError as error:
result = {"ok": False, "error": str(error)}
success = bool(result["ok"])
elif command == "open" and len(argv) == 1:
result = open_home(config)
success = bool(result["ok"])
config/dot/quickshell/services/HomeAssistant.qml
+2 -2
View File
@@ -1,7 +1,7 @@
pragma Singleton
// Home Assistant state for the Control Center. Credentials and REST details
// remain behind the helper; QML receives configured favourites only.
// remain behind the helper; QML receives a normalized light catalog only.
import Quickshell
import Quickshell.Io
@@ -151,7 +151,7 @@ Singleton {
Process {
id: refreshProc
command: [root.helperPath, "snapshot"]
command: [root.helperPath, "catalog"]
stdout: StdioCollector {
onStreamFinished: root.consumeSnapshot(this.text)
}
tests/quickshell/home-assistant-helper-contract.sh
+9 -19
View File
@@ -12,24 +12,14 @@ helper="$project_root/config/dot/quickshell/scripts/panama-home-assistant"
[[ -x "$helper" ]] || fail 'helper is missing or not executable'
probe="$($helper probe)"
catalog="$($helper catalog)"
jq -e '
.configured == true and
.reachable == true and
(.entityCount | type == "number" and . > 0) and
.error == ""
' <<<"$probe" >/dev/null || fail 'live API probe failed'
.ok == true and .configured == true and .error == "" and
(.entities | type == "array" and length > 0) and
([.entities[] | (keys | sort) == (["active", "available", "brightnessPct", "dimmable", "id", "sourceName", "state"] | sort)] | all) and
([.entities[] | (.id | startswith("light.")) and (.brightnessPct >= 0 and .brightnessPct <= 100)] | all) and
(.legacyEntityIds | type == "array")
' <<<"$catalog" >/dev/null || fail 'live catalog shape is invalid'
snapshot="$($helper snapshot)"
jq -e --argjson expected "$(jq '.entityCount' <<<"$probe")" '
.ok == true and
.configured == true and
.error == "" and
(.generatedAt | type == "number") and
(.entities | type == "array" and length == $expected) and
([.entities[] | (keys | sort) == (["active", "available", "domain", "id", "name", "state"] | sort)] | all) and
([.entities[] | (.active | type == "boolean") and (.available | type == "boolean")] | all)
' <<<"$snapshot" >/dev/null || fail 'live snapshot shape is invalid'
printf 'Home Assistant helper contract: PASS (configured=true, %s favourites; contents redacted)\n' \
"$(jq '.entities | length' <<<"$snapshot")"
printf 'Home Assistant helper contract: PASS (configured=true, %s lights; contents redacted)\n' \
"$(jq '.entities | length' <<<"$catalog")"
tests/quickshell/home_assistant_bridge_test.py
+125 -30
View File
@@ -14,6 +14,7 @@ from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
ROOT = pathlib.Path(__file__).resolve().parents[2]
HELPER = ROOT / "config/dot/quickshell/scripts/panama-home-assistant"
QML_SERVICE = ROOT / "config/dot/quickshell/services/HomeAssistant.qml"
loader = importlib.machinery.SourceFileLoader("panama_home_assistant", str(HELPER))
spec = importlib.util.spec_from_loader(loader.name, loader)
@@ -54,20 +55,36 @@ class FakeHomeAssistant(BaseHTTPRequestHandler):
if self.path == "/api/states":
self._json(
[
{
"entity_id": "light.hall",
"state": "off",
"attributes": {"friendly_name": "Hall"},
},
{
"entity_id": "sensor.private",
"state": "1",
"attributes": {"friendly_name": "Private"},
},
{
"entity_id": "light.kitchen",
"state": "on",
"attributes": {"friendly_name": "Kitchen"},
"attributes": {
"friendly_name": "Kitchen",
"brightness": 128,
"supported_color_modes": ["brightness"],
},
},
{
"entity_id": "light.hall",
"state": "off",
"attributes": {
"friendly_name": "Hall",
"supported_color_modes": ["color_temp"],
},
},
{"entity_id": "sensor.private", "state": "1", "attributes": {"token": "never-return"}},
{
"entity_id": "light.malformed",
"state": "on",
"attributes": "invalid",
},
{
"entity_id": "light.corner",
"state": "unavailable",
"attributes": {
"friendly_name": "Corner",
"supported_color_modes": ["brightness"],
},
},
]
)
@@ -78,7 +95,11 @@ class FakeHomeAssistant(BaseHTTPRequestHandler):
length = int(self.headers.get("Content-Length", "0"))
body = self.rfile.read(length)
self._record(body)
if self.path == "/api/services/homeassistant/toggle":
if self.path in {
"/api/services/homeassistant/toggle",
"/api/services/light/turn_on",
"/api/services/light/turn_off",
}:
self._json([])
return
self._json({"message": "not found"}, 404)
@@ -123,33 +144,71 @@ class HomeAssistantBridgeTest(unittest.TestCase):
self.assertEqual(config.base_url, "https://home.example")
self.assertEqual(config.entity_ids, ("light.kitchen", "light.hall"))
def test_snapshot_filters_and_preserves_configured_order(self) -> None:
snapshot = bridge.collect_snapshot(self.config())
self.assertTrue(snapshot["ok"])
self.assertEqual(
[item["name"] for item in snapshot["entities"]],
["Kitchen", "Hall"],
def test_config_does_not_require_legacy_entity_ids(self) -> None:
self.assertTrue(
bridge.Config(
base_url="https://home.example",
token="fixture-token",
entity_ids=(),
).configured
)
self.assertNotIn("sensor.private", json.dumps(snapshot))
def test_legacy_entity_ids_remain_migration_metadata(self) -> None:
config = bridge.resolve_config(
{
"PANAMA_HOME_ASSISTANT_URL": "https://home.example",
"PANAMA_HOME_ASSISTANT_TOKEN": "fixture-token",
},
legacy=lambda: bridge.Config(
base_url="https://legacy.example",
token="legacy-token",
entity_ids=("light.kitchen",),
),
)
self.assertEqual(config.base_url, "https://home.example")
self.assertEqual(config.token, "fixture-token")
self.assertEqual(config.entity_ids, ("light.kitchen",))
def test_catalog_filters_and_normalizes_discovered_lights(self) -> None:
catalog = bridge.collect_catalog(self.config())
self.assertTrue(catalog["ok"])
self.assertEqual(
[item["sourceName"] for item in catalog["entities"]],
["Kitchen", "Hall", "Corner"],
)
self.assertEqual(catalog["entities"][0]["brightnessPct"], 50)
self.assertEqual(catalog["entities"][1]["brightnessPct"], 0)
self.assertTrue(all(item["dimmable"] for item in catalog["entities"]))
rendered = json.dumps(catalog)
self.assertNotIn("sensor.private", rendered)
self.assertNotIn("light.malformed", rendered)
self.assertNotIn("token", rendered)
def test_snapshot_uses_bearer_authentication(self) -> None:
bridge.collect_snapshot(self.config())
bridge.collect_catalog(self.config())
request = FakeHomeAssistant.requests[-1]
self.assertEqual(request["method"], "GET")
self.assertEqual(request["path"], "/api/states")
self.assertEqual(request["authorization"], "Bearer fixture-token")
def test_toggle_rejects_an_unconfigured_entity(self) -> None:
with self.assertRaisesRegex(ValueError, "entity-not-configured"):
bridge.ensure_configured(
"light.office",
("light.kitchen",),
)
def test_snapshot_remains_a_catalog_compatibility_alias(self) -> None:
snapshot = bridge.collect_snapshot(self.config())
def test_toggle_calls_the_homeassistant_service(self) -> None:
result = bridge.toggle(self.config(), "light.kitchen")
self.assertTrue(snapshot["ok"])
self.assertEqual(
[item["sourceName"] for item in snapshot["entities"]],
["Kitchen", "Hall", "Corner"],
)
def test_action_rejects_an_entity_not_in_the_live_catalog(self) -> None:
with self.assertRaisesRegex(ValueError, "entity-not-discovered"):
bridge.toggle(self.config(), "light.office")
def test_toggle_authorizes_against_discovered_catalog(self) -> None:
result = bridge.toggle(self.config(), "light.corner")
self.assertTrue(result["ok"])
request = FakeHomeAssistant.requests[-1]
@@ -157,9 +216,45 @@ class HomeAssistantBridgeTest(unittest.TestCase):
self.assertEqual(request["path"], "/api/services/homeassistant/toggle")
self.assertEqual(
json.loads(request["body"]),
{"entity_id": "light.kitchen"},
{"entity_id": "light.corner"},
)
def test_brightness_uses_turn_on_for_positive_percent(self) -> None:
bridge.set_brightness(self.config(), "light.kitchen", 62)
request = FakeHomeAssistant.requests[-1]
self.assertEqual(request["path"], "/api/services/light/turn_on")
self.assertEqual(
json.loads(request["body"]),
{"entity_id": "light.kitchen", "brightness_pct": 62},
)
def test_brightness_zero_uses_turn_off(self) -> None:
bridge.set_brightness(self.config(), "light.hall", 0)
request = FakeHomeAssistant.requests[-1]
self.assertEqual(request["path"], "/api/services/light/turn_off")
self.assertEqual(
json.loads(request["body"]),
{"entity_id": "light.hall"},
)
def test_brightness_rejects_invalid_values_before_any_request(self) -> None:
for value in (-1, 101, 1.5, "bright"):
with self.subTest(value=value):
with self.assertRaisesRegex(ValueError, "invalid-brightness"):
bridge.set_brightness(self.config(), "light.kitchen", value)
self.assertEqual(FakeHomeAssistant.requests, [])
def test_parse_brightness_rejects_invalid_cli_values(self) -> None:
for value in ("-1", "101", "1.5", "bright"):
with self.subTest(value=value):
with self.assertRaisesRegex(ValueError, "invalid-brightness"):
bridge.parse_brightness(value)
def test_qml_refreshes_through_the_catalog_command(self) -> None:
self.assertIn('command: [root.helperPath, "catalog"]', QML_SERVICE.read_text())
def test_authentication_error_is_redacted(self) -> None:
self.assertEqual(
bridge.public_http_error(401, "sensitive response"),