dev: load real .pca fixtures into the mock-panel
Some checks failed
Validate / Hassfest (push) Has been cancelled
Validate / HACS validation (push) Has been cancelled

Adds an OMNI_PCA_FIXTURE escape hatch so the mock can serve real
panel data instead of the synthetic five-zone state. With this in
place the dev stack is wire-indistinguishable from the source
panel for everything the HA integration touches: 330 programs,
16 zones, 44 units, 2 thermostats etc. from our test fixture.

- run_mock_panel.py: --pca / OMNI_PCA_FIXTURE accepts a path; the
  decryption key is auto-derived from a sibling PCA01.CFG when one
  exists (the common PC Access export layout), with --pca-key /
  OMNI_PCA_FIXTURE_KEY as override. Falls back to KEY_EXPORT for
  vanilla unsigned exports.
- docker-compose.yml: mount /home/kdm/home-auto/HAI as /fixtures
  read-only and surface OMNI_PCA_FIXTURE so dev/.env can drive it.
- dev/README.md: new section documenting fixture loading.
- dev/screenshot_overview.py: quick playwright helper for capturing
  the side-panel landing page with whatever fixture is loaded.
- dev/artifacts/screenshots/2026-05-17/real-pca-overview.png: snapshot
  of the Omni Programs side panel against the real .pca fixture
  (330 programs).
This commit is contained in:
Ryan Malloy 2026-05-17 13:06:19 -06:00
parent 8a0fb1e4fe
commit 9726ee36bb
5 changed files with 208 additions and 3 deletions

View File

@ -44,6 +44,32 @@ make dev-down # stop the stack
make dev-reset # wipe HA config and start fresh make dev-reset # wipe HA config and start fresh
``` ```
## Load real `.pca` data into the mock
By default the mock serves a small synthetic state (five zones, four
units, …). Point `OMNI_PCA_FIXTURE` at a real `.pca` file to make the
mock indistinguishable on the wire from the source panel:
```bash
# dev/.env (gitignored)
OMNI_PCA_FIXTURE=/fixtures/path/to/Account.pca
```
The host directory `/home/kdm/home-auto/HAI` is mounted at `/fixtures`
inside the mock-panel container (see `docker-compose.yml`); adjust the
mount if your `.pca` lives elsewhere.
The decryption key is auto-derived from a sibling `PCA01.CFG` if one
exists (this is how PC Access exports usually ship). To override:
```bash
OMNI_PCA_FIXTURE_KEY=0xC1A280B2 # or --pca-key on the command line
```
`MockState.from_pca` populates zones, units, areas, thermostats,
buttons, programs, model byte, and firmware version from the file —
everything the HA integration reads at discovery time.
## Notes ## Notes
- The HA container mounts `../custom_components/omni_pca/` read-only, so - The HA container mounts `../custom_components/omni_pca/` read-only, so

Binary file not shown.

After

Width:  |  Height:  |  Size: 113 KiB

View File

@ -35,8 +35,14 @@ services:
volumes: volumes:
- ../src:/tmp/mock/src:ro - ../src:/tmp/mock/src:ro
- ./run_mock_panel.py:/tmp/mock/run_mock_panel.py:ro - ./run_mock_panel.py:/tmp/mock/run_mock_panel.py:ro
# Mount the captured .pca fixtures read-only so the mock can
# optionally seed its state from a real export. Set
# OMNI_PCA_FIXTURE in dev/.env (or pass on the command line) to
# activate; left unset, the mock uses the hard-coded sample.
- /home/kdm/home-auto/HAI:/fixtures:ro
environment: environment:
PYTHONPATH: /tmp/mock/src PYTHONPATH: /tmp/mock/src
OMNI_PCA_FIXTURE: ${OMNI_PCA_FIXTURE:-}
command: command:
- sh - sh
- -c - -c

View File

@ -10,8 +10,10 @@ from __future__ import annotations
import argparse import argparse
import asyncio import asyncio
import logging import logging
import os
import signal import signal
import sys import sys
from pathlib import Path
from omni_pca.mock_panel import ( from omni_pca.mock_panel import (
MockAreaState, MockAreaState,
@ -22,10 +24,55 @@ from omni_pca.mock_panel import (
MockUnitState, MockUnitState,
MockZoneState, MockZoneState,
) )
from omni_pca.commands import Command
from omni_pca.pca_file import KEY_EXPORT, parse_pca01_cfg
from omni_pca.programs import Days, Program, ProgramType
DEFAULT_KEY_HEX = "000102030405060708090a0b0c0d0e0f" DEFAULT_KEY_HEX = "000102030405060708090a0b0c0d0e0f"
def _seed_programs() -> dict[int, bytes]:
"""A handful of programs covering compact-form + clausal-chain shapes.
Slot 200..202 is a chain with a structured-AND condition whose Arg2
is itself a Thermostat reference exercises the Arg2-as-object
editor controls.
"""
programs: dict[int, Program] = {
12: Program(
slot=12, prog_type=int(ProgramType.TIMED),
cmd=int(Command.UNIT_ON), pr2=1,
hour=6, minute=0, days=int(Days.MONDAY | Days.FRIDAY),
),
42: Program(
slot=42, prog_type=int(ProgramType.TIMED),
cmd=int(Command.UNIT_OFF), pr2=2,
hour=22, minute=30, days=int(Days.SUNDAY),
),
# Chain: WHEN zone 1 not-ready, AND IF Thermostat 1.Temp >
# Thermostat 2.Temp, THEN turn ON unit 3. The AND record is a
# structured-OP comparison with Arg2 as a Thermostat reference.
200: Program(
slot=200, prog_type=int(ProgramType.WHEN),
month=0x04, day=0x01,
),
201: Program(
slot=201, prog_type=int(ProgramType.AND),
cond=(4 << 8) | 4, # op=GT (4), arg1Type=Thermostat (4)
cond2=1, # arg1Ix=1
cmd=1, # arg1Field=current temp
par=4, # arg2Type=Thermostat (4)
pr2=2, # arg2Ix=2
month=1, # arg2Field=current temp
),
202: Program(
slot=202, prog_type=int(ProgramType.THEN),
cmd=int(Command.UNIT_ON), pr2=3,
),
}
return {slot: p.encode_wire_bytes() for slot, p in programs.items()}
def _populated_state() -> MockState: def _populated_state() -> MockState:
"""A small but representative set of objects so HA shows real entities.""" """A small but representative set of objects so HA shows real entities."""
return MockState( return MockState(
@ -56,11 +103,50 @@ def _populated_state() -> MockState:
3: MockButtonState(name="GOODNIGHT"), 3: MockButtonState(name="GOODNIGHT"),
}, },
user_codes={1: 1234, 2: 5678}, user_codes={1: 1234, 2: 5678},
programs=_seed_programs(),
) )
async def _serve(host: str, port: int, key: bytes) -> None: def _key_for_pca(path: Path, override: int | None) -> int:
panel = MockPanel(controller_key=key, state=_populated_state()) """Pick the decryption key for a .pca file.
Priority:
1. Explicit override (CLI / env var).
2. Per-installation key from a sibling ``PCA01.CFG`` (most common
PC Access ships each export with a matching config file).
3. ``KEY_EXPORT`` as a last resort for vanilla exports.
"""
if override is not None:
return override
cfg_path = path.parent / "PCA01.CFG"
if cfg_path.is_file():
cfg = parse_pca01_cfg(cfg_path.read_bytes())
logging.info("derived pca_key from %s: 0x%08X", cfg_path.name, cfg.pca_key)
return cfg.pca_key
logging.info("no sibling PCA01.CFG; falling back to KEY_EXPORT")
return KEY_EXPORT
def _state_from_pca(path: Path, key: int) -> MockState:
"""Seed a MockState from a real .pca file."""
state = MockState.from_pca(str(path), key=key)
logging.info(
"loaded %s: %d zones, %d units, %d areas, %d thermostats, %d programs",
path.name,
len(state.zones), len(state.units), len(state.areas),
len(state.thermostats), len(state.programs),
)
return state
async def _serve(
host: str, port: int, key: bytes, pca: Path | None, pca_key: int | None,
) -> None:
if pca is not None:
state = _state_from_pca(pca, _key_for_pca(pca, pca_key))
else:
state = _populated_state()
panel = MockPanel(controller_key=key, state=state)
async with panel.serve(host=host, port=port) as (bound_host, bound_port): async with panel.serve(host=host, port=port) as (bound_host, bound_port):
logging.info("MockPanel listening on %s:%d", bound_host, bound_port) logging.info("MockPanel listening on %s:%d", bound_host, bound_port)
logging.info("Use this controller key in HA: %s", key.hex()) logging.info("Use this controller key in HA: %s", key.hex())
@ -86,6 +172,23 @@ def main() -> int:
default=DEFAULT_KEY_HEX, default=DEFAULT_KEY_HEX,
help="32 hex chars; default is the docker-compose value", help="32 hex chars; default is the docker-compose value",
) )
parser.add_argument(
"--pca",
default=os.environ.get("OMNI_PCA_FIXTURE"),
help="Path to a .pca file. When supplied, the mock seeds its "
"state from this file instead of the hard-coded sample. "
"Can also be set via OMNI_PCA_FIXTURE.",
)
parser.add_argument(
"--pca-key",
type=lambda s: int(s, 0),
default=(
int(os.environ["OMNI_PCA_FIXTURE_KEY"], 0)
if os.environ.get("OMNI_PCA_FIXTURE_KEY") else None
),
help="32-bit decryption key for --pca. Default: auto-derive from "
"a sibling PCA01.CFG, or fall back to KEY_EXPORT.",
)
args = parser.parse_args() args = parser.parse_args()
logging.basicConfig( logging.basicConfig(
@ -104,7 +207,14 @@ def main() -> int:
file=sys.stderr) file=sys.stderr)
return 2 return 2
asyncio.run(_serve(args.host, args.port, key)) pca_path: Path | None = None
if args.pca:
pca_path = Path(args.pca)
if not pca_path.is_file():
print(f"--pca path not found: {pca_path}", file=sys.stderr)
return 2
asyncio.run(_serve(args.host, args.port, key, pca_path, args.pca_key))
return 0 return 0

View File

@ -0,0 +1,63 @@
#!/usr/bin/env python3
"""Quick screenshot of the Omni Programs side panel landing page."""
from __future__ import annotations
import asyncio
import sys
from datetime import datetime
from pathlib import Path
import httpx
from playwright.async_api import async_playwright
HA_URL = "http://localhost:8123"
USERNAME = "demo"
PASSWORD = "demo-password-1234"
async def _login_token() -> str:
async with httpx.AsyncClient(base_url=HA_URL, timeout=30) as c:
r = await c.post("/auth/login_flow", json={
"client_id": HA_URL, "handler": ["homeassistant", None],
"redirect_uri": HA_URL,
})
flow_id = r.json()["flow_id"]
r = await c.post(f"/auth/login_flow/{flow_id}", json={
"username": USERNAME, "password": PASSWORD, "client_id": HA_URL,
})
code = r.json()["result"]
r = await c.post("/auth/token", data={
"client_id": HA_URL, "grant_type": "authorization_code", "code": code,
})
return r.json()["access_token"]
async def amain(outdir: Path) -> None:
token = await _login_token()
outdir.mkdir(parents=True, exist_ok=True)
async with async_playwright() as p:
browser = await p.chromium.launch()
context = await browser.new_context(viewport={"width": 1400, "height": 900})
await context.add_init_script(f"""
window.localStorage.setItem('hassTokens', JSON.stringify({{
access_token: '{token}', token_type: 'Bearer', refresh_token: '',
expires: Date.now() + 3600000, hassUrl: '{HA_URL}', clientId: '{HA_URL}',
}}));
window.localStorage.setItem('selectedTheme', JSON.stringify({{dark: false}}));
""")
page = await context.new_page()
await page.goto(f"{HA_URL}/omni-panel-programs", wait_until="domcontentloaded")
await page.wait_for_timeout(8000)
path = outdir / "real-pca-overview.png"
await page.screenshot(path=str(path), full_page=True)
print(f" wrote {path}")
await browser.close()
if __name__ == "__main__":
outdir = Path(sys.argv[1]) if len(sys.argv) > 1 else (
Path(__file__).parent / "artifacts" / "screenshots" /
datetime.now().strftime("%Y-%m-%d")
)
asyncio.run(amain(outdir))