From e57fbc41e356c9e7035d911cd5e9abe4ac639a7e Mon Sep 17 00:00:00 2001 From: Ryan Malloy Date: Tue, 12 May 2026 19:15:32 -0600 Subject: [PATCH] HA: optional .pca file as alternate source for panel programs Adds CONF_PCA_PATH + CONF_PCA_KEY config-flow fields. When set, the coordinator parses programs from the .pca file at that path instead of streaming them over the wire on every entry refresh. Useful for: * deployments where wire enumeration is slow (1500-slot iteration) * offline snapshots when the panel is unreachable * deterministic test setups against a known fixture The config-flow validates the file is readable and decrypts cleanly, surfacing pca_not_found / pca_decode_failed errors via the strings/ en.json translations. The .pca path is checked first in _discover_programs; if absent the wire path runs as before. So existing deployments are unaffected. Tests cover the success path (live fixture, 330 programs) and the two validation failures (missing file, garbage bytes). --- custom_components/omni_pca/__init__.py | 6 + custom_components/omni_pca/config_flow.py | 71 ++++++++-- custom_components/omni_pca/const.py | 7 + custom_components/omni_pca/coordinator.py | 62 +++++++-- custom_components/omni_pca/strings.json | 9 +- .../omni_pca/translations/en.json | 9 +- tests/ha_integration/test_pca_source.py | 131 ++++++++++++++++++ 7 files changed, 268 insertions(+), 27 deletions(-) create mode 100644 tests/ha_integration/test_pca_source.py diff --git a/custom_components/omni_pca/__init__.py b/custom_components/omni_pca/__init__.py index 042e790..39a908d 100644 --- a/custom_components/omni_pca/__init__.py +++ b/custom_components/omni_pca/__init__.py @@ -16,6 +16,8 @@ from homeassistant.exceptions import ConfigEntryNotReady from .const import ( CONF_CONTROLLER_KEY, + CONF_PCA_KEY, + CONF_PCA_PATH, CONF_TRANSPORT, DEFAULT_TRANSPORT, DOMAIN, @@ -57,6 +59,8 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: return False transport: str = entry.data.get(CONF_TRANSPORT, DEFAULT_TRANSPORT) + pca_path: str = entry.data.get(CONF_PCA_PATH, "") or "" + pca_key: int = entry.data.get(CONF_PCA_KEY, 0) coordinator = OmniDataUpdateCoordinator( hass, entry, @@ -64,6 +68,8 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: port=port, controller_key=controller_key, transport=transport, + pca_path=pca_path or None, + pca_key=pca_key, ) try: diff --git a/custom_components/omni_pca/config_flow.py b/custom_components/omni_pca/config_flow.py index 6dc5c90..d2a6fee 100644 --- a/custom_components/omni_pca/config_flow.py +++ b/custom_components/omni_pca/config_flow.py @@ -20,6 +20,8 @@ from omni_pca.connection import ( from .const import ( CONF_CONTROLLER_KEY, + CONF_PCA_KEY, + CONF_PCA_PATH, CONF_TRANSPORT, CONTROLLER_KEY_HEX_LEN, DEFAULT_PORT, @@ -70,6 +72,15 @@ _USER_SCHEMA = vol.Schema( vol.Required(CONF_TRANSPORT, default=DEFAULT_TRANSPORT): vol.In( [TRANSPORT_TCP, TRANSPORT_UDP] ), + # Optional: load panel programs from a saved .pca file on the HA + # filesystem (e.g. /config/pca/My_House.pca) instead of streaming + # them from the panel on every restart. Useful when wire + # enumeration is slow or unreliable. CONF_PCA_KEY is the + # per-install key from PCA01.CFG (a uint32, 0 for plain-text). + vol.Optional(CONF_PCA_PATH, default=""): str, + vol.Optional(CONF_PCA_KEY, default=0): vol.All( + vol.Coerce(int), vol.Range(min=0, max=0xFFFFFFFF) + ), } ) @@ -90,6 +101,8 @@ class OmniConfigFlow(ConfigFlow, domain=DOMAIN): host: str = user_input[CONF_HOST].strip() port: int = user_input[CONF_PORT] transport: str = user_input.get(CONF_TRANSPORT, DEFAULT_TRANSPORT) + pca_path: str = (user_input.get(CONF_PCA_PATH) or "").strip() + pca_key: int = user_input.get(CONF_PCA_KEY, 0) unique_id = f"{host}:{port}" await self.async_set_unique_id(unique_id) @@ -101,19 +114,24 @@ class OmniConfigFlow(ConfigFlow, domain=DOMAIN): LOGGER.debug("controller key rejected: %s", err) errors[CONF_CONTROLLER_KEY] = "invalid_key" else: - title, error = await self._probe(host, port, key, transport) - if error is not None: - errors["base"] = error - else: - return self.async_create_entry( - title=title or f"Omni Panel ({host})", - data={ - CONF_HOST: host, - CONF_PORT: port, - CONF_CONTROLLER_KEY: key.hex(), - CONF_TRANSPORT: transport, - }, - ) + if pca_path and (pca_err := await self._validate_pca(pca_path, pca_key)): + errors[CONF_PCA_PATH] = pca_err + if not errors: + title, error = await self._probe(host, port, key, transport) + if error is not None: + errors["base"] = error + else: + return self.async_create_entry( + title=title or f"Omni Panel ({host})", + data={ + CONF_HOST: host, + CONF_PORT: port, + CONF_CONTROLLER_KEY: key.hex(), + CONF_TRANSPORT: transport, + CONF_PCA_PATH: pca_path, + CONF_PCA_KEY: pca_key, + }, + ) return self.async_show_form( step_id="user", @@ -121,6 +139,33 @@ class OmniConfigFlow(ConfigFlow, domain=DOMAIN): errors=errors, ) + async def _validate_pca(self, path: str, key: int) -> str | None: + """Validate the user-supplied .pca path is readable and decrypts. + + Returns ``None`` on success or an error code string on failure + (matches the {code: message} keys in strings.json). + """ + from pathlib import Path + + from omni_pca.pca_file import parse_pca_file + + p = Path(path) + if not p.is_file(): + return "pca_not_found" + try: + data = await self.hass.async_add_executor_job(p.read_bytes) + acct = await self.hass.async_add_executor_job( + lambda: parse_pca_file(data, key=key) + ) + except Exception as err: + LOGGER.debug("pca file rejected: %s", err) + return "pca_decode_failed" + # Sanity: programs block decoded cleanly. Empty is allowed + # (legitimate brand-new install with no programs). + if not isinstance(acct.programs, tuple): + return "pca_decode_failed" + return None + async def async_step_reauth( self, entry_data: Mapping[str, Any] ) -> ConfigFlowResult: diff --git a/custom_components/omni_pca/const.py b/custom_components/omni_pca/const.py index 9f7de9a..9d93c03 100644 --- a/custom_components/omni_pca/const.py +++ b/custom_components/omni_pca/const.py @@ -14,6 +14,13 @@ DEFAULT_TIMEOUT: Final = 5.0 CONF_CONTROLLER_KEY: Final = "controller_key" CONF_TRANSPORT: Final = "transport" +# Optional: when set, load panel programs from a .pca file at this path +# instead of enumerating them over the wire on every entry refresh. The +# .pca file is decrypted with CONF_PCA_KEY (the per-install key from +# PCA01.CFG, or 0 for a plain-text dump). Both must be set together. +CONF_PCA_PATH: Final = "pca_path" +CONF_PCA_KEY: Final = "pca_key" + TRANSPORT_TCP: Final = "tcp" TRANSPORT_UDP: Final = "udp" DEFAULT_TRANSPORT: Final = TRANSPORT_TCP diff --git a/custom_components/omni_pca/coordinator.py b/custom_components/omni_pca/coordinator.py index 4b90898..51cbef7 100644 --- a/custom_components/omni_pca/coordinator.py +++ b/custom_components/omni_pca/coordinator.py @@ -138,6 +138,8 @@ class OmniDataUpdateCoordinator(DataUpdateCoordinator[OmniData]): port: int, controller_key: bytes, transport: str = "tcp", + pca_path: str | None = None, + pca_key: int = 0, ) -> None: super().__init__( hass, @@ -150,6 +152,8 @@ class OmniDataUpdateCoordinator(DataUpdateCoordinator[OmniData]): self._port = port self._controller_key = controller_key self._transport = transport + self._pca_path = pca_path + self._pca_key = pca_key self._client: OmniClient | None = None self._discovery_done = False self._discovered: OmniData | None = None @@ -383,19 +387,25 @@ class OmniDataUpdateCoordinator(DataUpdateCoordinator[OmniData]): async def _discover_programs( self, client: OmniClient ) -> dict[int, Program]: - """Enumerate defined panel programs via the appropriate wire path. + """Enumerate defined panel programs. - * v2 (TCP): ``client.iter_programs()`` drives UploadProgram with - request_reason=1 ("next defined after slot"), stopping at EOD. - * v1 (UDP): the adapter forwards to OmniClientV1.iter_programs(), - which is a bare UploadPrograms stream ack-walked to EOD. + Two sources, in order of preference: - Both surfaces yield :class:`omni_pca.programs.Program` and skip - empty slots, so the resulting dict only carries defined programs. - Errors during enumeration are logged and swallowed — programs - are a non-critical part of discovery, so a partial list is better - than blocking the entry setup. + 1. ``CONF_PCA_PATH`` is configured → parse the .pca file and + extract the programs block. Avoids streaming 1500 records on + every entry refresh and works against an offline snapshot. + 2. Otherwise → enumerate over the wire: + * v2 (TCP): ``client.iter_programs()`` drives UploadProgram + with request_reason=1 ("next defined after slot"). + * v1 (UDP): adapter forwards to OmniClientV1.iter_programs(), + a bare UploadPrograms stream ack-walked to EOD. + + Both paths yield :class:`omni_pca.programs.Program` and skip + empty slots. Errors are logged and swallowed — programs are + non-critical discovery, so a partial list beats blocking setup. """ + if self._pca_path: + return await self._discover_programs_from_pca() out: dict[int, Program] = {} try: async for prog in client.iter_programs(): @@ -409,6 +419,38 @@ class OmniDataUpdateCoordinator(DataUpdateCoordinator[OmniData]): ) return out + async def _discover_programs_from_pca(self) -> dict[int, Program]: + """Parse the configured .pca file and pull out its programs block. + + Runs the disk I/O on the executor since :mod:`pca_file` does + sync reads. Any failure (missing file, bad key, malformed block) + is logged and downgraded to an empty dict — the rest of + discovery still works. + """ + from pathlib import Path + + from omni_pca.pca_file import parse_pca_file + + try: + data = await self.hass.async_add_executor_job( + Path(self._pca_path).read_bytes + ) + acct = await self.hass.async_add_executor_job( + lambda: parse_pca_file(data, key=self._pca_key) + ) + except Exception: + LOGGER.warning( + "failed to load programs from %s — falling back to empty list", + self._pca_path, exc_info=True, + ) + return {} + out: dict[int, Program] = {} + for prog in acct.programs: + if prog.slot is None or prog.is_empty(): + continue + out[prog.slot] = prog + return out + async def _walk_properties( self, client: OmniClient, diff --git a/custom_components/omni_pca/strings.json b/custom_components/omni_pca/strings.json index d2234b5..78c1401 100644 --- a/custom_components/omni_pca/strings.json +++ b/custom_components/omni_pca/strings.json @@ -3,11 +3,14 @@ "step": { "user": { "title": "Connect to your Omni panel", - "description": "Enter the IP/hostname of your HAI/Leviton Omni Pro II controller and the 32-character hex Controller Key. Use `omni-pca decode-pca .pca --field controller_key` to extract the key from a PC Access export.", + "description": "Enter the IP/hostname of your HAI/Leviton Omni Pro II controller and the 32-character hex Controller Key. Use `omni-pca decode-pca .pca --field controller_key` to extract the key from a PC Access export. Optionally provide a path to a saved `.pca` file to load panel programs from disk instead of streaming them from the controller.", "data": { "host": "Host", "port": "Port", - "controller_key": "Controller Key (32 hex chars)" + "controller_key": "Controller Key (32 hex chars)", + "transport": "Transport", + "pca_path": ".pca file path (optional)", + "pca_key": ".pca per-install key (integer; 0 if plain-text)" } }, "reauth_confirm": { @@ -22,6 +25,8 @@ "cannot_connect": "Could not reach the panel. Check the host, port, and that TCP/4369 is open.", "invalid_auth": "The Controller Key was rejected by the panel.", "invalid_key": "Controller Key must be exactly 32 hexadecimal characters (16 bytes).", + "pca_not_found": "No file at the supplied .pca path.", + "pca_decode_failed": "Could not decode the .pca file. Check the per-install key.", "unknown": "An unexpected error occurred. Check the Home Assistant logs." }, "abort": { diff --git a/custom_components/omni_pca/translations/en.json b/custom_components/omni_pca/translations/en.json index d2234b5..78c1401 100644 --- a/custom_components/omni_pca/translations/en.json +++ b/custom_components/omni_pca/translations/en.json @@ -3,11 +3,14 @@ "step": { "user": { "title": "Connect to your Omni panel", - "description": "Enter the IP/hostname of your HAI/Leviton Omni Pro II controller and the 32-character hex Controller Key. Use `omni-pca decode-pca .pca --field controller_key` to extract the key from a PC Access export.", + "description": "Enter the IP/hostname of your HAI/Leviton Omni Pro II controller and the 32-character hex Controller Key. Use `omni-pca decode-pca .pca --field controller_key` to extract the key from a PC Access export. Optionally provide a path to a saved `.pca` file to load panel programs from disk instead of streaming them from the controller.", "data": { "host": "Host", "port": "Port", - "controller_key": "Controller Key (32 hex chars)" + "controller_key": "Controller Key (32 hex chars)", + "transport": "Transport", + "pca_path": ".pca file path (optional)", + "pca_key": ".pca per-install key (integer; 0 if plain-text)" } }, "reauth_confirm": { @@ -22,6 +25,8 @@ "cannot_connect": "Could not reach the panel. Check the host, port, and that TCP/4369 is open.", "invalid_auth": "The Controller Key was rejected by the panel.", "invalid_key": "Controller Key must be exactly 32 hexadecimal characters (16 bytes).", + "pca_not_found": "No file at the supplied .pca path.", + "pca_decode_failed": "Could not decode the .pca file. Check the per-install key.", "unknown": "An unexpected error occurred. Check the Home Assistant logs." }, "abort": { diff --git a/tests/ha_integration/test_pca_source.py b/tests/ha_integration/test_pca_source.py new file mode 100644 index 0000000..6ee267b --- /dev/null +++ b/tests/ha_integration/test_pca_source.py @@ -0,0 +1,131 @@ +"""HA-side integration: optional .pca file source for panel programs. + +When ``CONF_PCA_PATH`` is set in the entry data, the coordinator should +parse the .pca file at that path (with ``CONF_PCA_KEY`` as the per-install +key) and use those programs *instead* of streaming them over the wire. +The wire-based discovery for everything else (zones, units, etc.) is +unaffected. +""" + +from __future__ import annotations + +from collections.abc import AsyncIterator +from pathlib import Path +from typing import Any + +import pytest +from custom_components.omni_pca.const import ( + CONF_CONTROLLER_KEY, + CONF_PCA_KEY, + CONF_PCA_PATH, + DOMAIN, +) +from homeassistant.config_entries import ConfigEntry +from homeassistant.const import CONF_HOST, CONF_PORT +from homeassistant.core import HomeAssistant +from pytest_homeassistant_custom_component.common import MockConfigEntry + +from .conftest import CONTROLLER_KEY_HEX + +LIVE_FIXTURE_PLAIN = Path( + "/home/kdm/home-auto/HAI/pca-re/extracted/Our_House.pca.plain" +) + + +def _materialize_encrypted_fixture(tmp_path: Path) -> tuple[Path, int]: + """Re-encrypt the plain fixture so parse_pca_file can decrypt it. + + parse_pca_file always runs the XOR keystream. The plain dump bypasses + that, so we re-apply the keystream with KEY_EXPORT and write the + result to tmp_path. Returns (file_path, key) the coordinator should use. + """ + from omni_pca.pca_file import KEY_EXPORT, decrypt_pca_bytes + + plain = LIVE_FIXTURE_PLAIN.read_bytes() + # XOR is symmetric — "decrypt" of plain bytes with the export key + # produces a valid encrypted .pca that parse_pca_file can read back. + encrypted = decrypt_pca_bytes(plain, KEY_EXPORT) + fixture = tmp_path / "Test_House.pca" + fixture.write_bytes(encrypted) + return fixture, KEY_EXPORT + + +@pytest.fixture +async def configured_with_pca( + hass: HomeAssistant, panel: tuple[Any, str, int], tmp_path: Path +) -> AsyncIterator[ConfigEntry]: + """Config entry pointing at a .pca file fixture for programs.""" + if not LIVE_FIXTURE_PLAIN.is_file(): + pytest.skip(f"live .pca fixture missing: {LIVE_FIXTURE_PLAIN}") + + fixture_path, pca_key = _materialize_encrypted_fixture(tmp_path) + + _, host, port = panel + entry = MockConfigEntry( + domain=DOMAIN, + data={ + CONF_HOST: host, + CONF_PORT: port, + CONF_CONTROLLER_KEY: CONTROLLER_KEY_HEX, + CONF_PCA_PATH: str(fixture_path), + CONF_PCA_KEY: pca_key, + }, + title=f"Mock Omni @ {host}:{port} (with .pca)", + unique_id=f"{host}:{port}", + ) + entry.add_to_hass(hass) + assert await hass.config_entries.async_setup(entry.entry_id) + await hass.async_block_till_done() + try: + yield entry + finally: + if entry.entry_id in hass.data.get(DOMAIN, {}): + await hass.config_entries.async_unload(entry.entry_id) + await hass.async_block_till_done() + + +async def test_pca_source_overrides_wire_programs( + hass: HomeAssistant, configured_with_pca: ConfigEntry +) -> None: + """The fixture .pca has 330 defined programs (Phase 1 recon). The mock + panel only seeded 3 in conftest. When pca_path is set, the .pca count + wins — proving the coordinator routed through _discover_programs_from_pca, + not iter_programs.""" + coordinator = hass.data[DOMAIN][configured_with_pca.entry_id] + assert len(coordinator.data.programs) == 330 + + # Sanity: the diagnostic sensor reflects the .pca count, not the mock seed. + sensors = [ + s for s in hass.states.async_all("sensor") + if "panel_programs" in s.entity_id + ] + assert len(sensors) == 1 + assert int(sensors[0].state) == 330 + + +async def test_pca_path_validation_rejects_missing_file( + hass: HomeAssistant, tmp_path: Path +) -> None: + """The config-flow validator returns ``pca_not_found`` for an absent + file. We exercise the helper directly to avoid spinning a full mock + panel just for the validation branch.""" + from custom_components.omni_pca.config_flow import OmniConfigFlow + + flow = OmniConfigFlow() + flow.hass = hass + err = await flow._validate_pca(str(tmp_path / "does-not-exist.pca"), 0) + assert err == "pca_not_found" + + +async def test_pca_path_validation_rejects_garbage( + hass: HomeAssistant, tmp_path: Path +) -> None: + """A file that doesn't decode as a .pca returns ``pca_decode_failed``.""" + from custom_components.omni_pca.config_flow import OmniConfigFlow + + garbage = tmp_path / "garbage.pca" + garbage.write_bytes(b"not a real pca file" * 1000) + flow = OmniConfigFlow() + flow.hass = hass + err = await flow._validate_pca(str(garbage), 0) + assert err == "pca_decode_failed"