#!/usr/bin/env python3 """Launch a long-running MockPanel suitable for the docker-compose dev stack. Reuses the mock fixture from the test suite so the behaviour matches what the HA integration tests prove out. Defaults match dev/docker-compose.yml. """ from __future__ import annotations import argparse import asyncio import logging import os import signal import sys from pathlib import Path from omni_pca.mock_panel import ( MockAreaState, MockButtonState, MockPanel, MockState, MockThermostatState, MockUnitState, 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" 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: """A small but representative set of objects so HA shows real entities.""" return MockState( zones={ 1: MockZoneState(name="FRONT_DOOR"), 2: MockZoneState(name="GARAGE_ENTRY"), 3: MockZoneState(name="BACK_DOOR"), 10: MockZoneState(name="LIVING_MOTION"), 11: MockZoneState(name="HALL_MOTION"), }, units={ 1: MockUnitState(name="LIVING_LAMP"), 2: MockUnitState(name="KITCHEN_OVERHEAD"), 3: MockUnitState(name="FRONT_PORCH"), 4: MockUnitState(name="BEDROOM_FAN"), }, areas={ 1: MockAreaState(name="MAIN"), 2: MockAreaState(name="GUEST"), }, thermostats={ 1: MockThermostatState(name="LIVING_ROOM"), 2: MockThermostatState(name="MASTER_BEDROOM"), }, buttons={ 1: MockButtonState(name="GOOD_MORNING"), 2: MockButtonState(name="MOVIE_MODE"), 3: MockButtonState(name="GOODNIGHT"), }, user_codes={1: 1234, 2: 5678}, programs=_seed_programs(), ) def _key_for_pca(path: Path, override: int | None) -> int: """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): logging.info("MockPanel listening on %s:%d", bound_host, bound_port) logging.info("Use this controller key in HA: %s", key.hex()) stop = asyncio.Event() def _on_signal() -> None: logging.info("shutdown signal received") stop.set() loop = asyncio.get_running_loop() for sig in (signal.SIGINT, signal.SIGTERM): loop.add_signal_handler(sig, _on_signal) await stop.wait() def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--host", default="0.0.0.0") parser.add_argument("--port", type=int, default=14369) parser.add_argument( "--controller-key", default=DEFAULT_KEY_HEX, 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() logging.basicConfig( level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s", ) try: key = bytes.fromhex(args.controller_key) except ValueError: print(f"controller-key must be 32 hex chars: {args.controller_key!r}", file=sys.stderr) return 2 if len(key) != 16: print(f"controller-key must decode to exactly 16 bytes (got {len(key)})", file=sys.stderr) return 2 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 if __name__ == "__main__": raise SystemExit(main())