Skip to content
HAI Omni Pro II — omni-pca docs

Architecture overview

omni-pca architecture overview Diagram of how the omni_pca library, the Home Assistant integration, the mock panel, and the test harness fit together. The library wraps a four-layer protocol stack (crypto, packet, message, opcodes) with a higher-level OmniClient. The HA integration's coordinator uses the client. The mock panel is a controller-side implementation of the same protocol used both for development and for the HA integration tests. LIBRARY · omni_pca crypto AES + whitening packet outer framing message CRC-16, opcodes opcodes v1 / v2 IntEnums OmniConnection handshake, sequence, reader task OmniClient typed methods, events() models 21 dataclasses events 26 SystemEvent types HA INTEGRATION · custom_components/omni_pca Coordinator discover + poll + push 8 entity platforms alarm · binary_sensor · button climate · event · light · sensor · switch TEST SURFACE MockPanel controller-side async TCP server HA test harness in-process HA 12 tests e2e tests 17 client ↔ mock round-trips unit tests 300+ on primitives + helpers uses TCP/4369 (encrypted) drives Solid arrows = function calls in-process · accent arrows = real TCP traffic. The mock panel is the same thing the real panel is, on the wire — it lets every test run without hardware.

The project has four moving parts:

  1. The Python library (omni_pca) — protocol, client, models.
  2. The Home Assistant custom component (custom_components/omni_pca/) — eight entity platforms on top of the library.
  3. The mock panel (omni_pca.mock_panel) — a controller-side emulator that speaks the same protocol as a real panel.
  4. The test harness — pytest suites that exercise (1) against (3) over real TCP, and (2) against (3) inside a real in-process Home Assistant.

Plus a docker dev stack that wires HA + mock together for manual smoke testing and screenshot capture.

The library

Section titled “The library”
omni_pca/
├── crypto.py        AES-128-ECB + per-block XOR pre-whitening + SessionKey derivation
├── opcodes.py       PacketType, OmniLinkMessageType, OmniLink2MessageType IntEnums
├── packet.py        Outer 4-byte-header + payload framing
├── message.py       Inner Message + CRC-16/MODBUS
├── connection.py    OmniConnection: async TCP + handshake + reader task
├── client.py        OmniClient: typed methods on top of OmniConnection
├── commands.py      Command IntEnum + CommandFailedError
├── events.py        SystemEvent hierarchy + EventStream iterator
├── models.py        21 frozen-slots dataclasses for every panel object
├── pca_file.py      Borland LCG cipher + .pca / .CFG parsers
├── mock_panel.py    Stateful controller-side emulator
└── __main__.py      omni-pca CLI: decode-pca, mock-panel, version

Three layers, each shorter than the next:

  • Protocol layer — crypto, packet, message, opcodes. Pure byte-mangling. No I/O.
  • Connection layer — connection. Async TCP, secure-session handshake, per-direction sequence numbers, reader task that dispatches solicited replies to Futures and unsolicited messages to a queue.
  • Client layer — client, commands, events, models. Typed methods, parsed dataclasses, typed event stream.

OmniClient is the surface most users want. OmniConnection is exposed for power users who need raw Message round-trips. The protocol layer is exposed because it’s useful for testing.

The HA integration

Section titled “The HA integration”
custom_components/omni_pca/
├── __init__.py            setup_entry / unload_entry
├── manifest.json          iot_class: local_push, requires omni-pca==2026.5.10
├── coordinator.py         OmniDataUpdateCoordinator: long-lived OmniClient + event listener
├── config_flow.py         User + reauth flows (host/port/key, hex validation)
├── helpers.py             Pure functions for everything HA-shape-dependent
├── services.py            Idempotent service registration + voluptuous schemas
├── services.yaml          UI-side service descriptions
├── diagnostics.py         Redacted snapshot dump for bug reports
├── alarm_control_panel.py
├── binary_sensor.py
├── button.py
├── climate.py
├── event.py
├── light.py
├── sensor.py
└── switch.py

OmniDataUpdateCoordinator keeps a single long-lived OmniClient, runs a one-time discovery pass at first refresh (enumerates zones, units, areas, thermostats, buttons), and starts a background task consuming client.events(). Every push event mutates the in-memory state dict and calls async_set_updated_data(), which fans out to the entity platforms. A 30-second poll backstops anything that didn’t push.

The eight entity platforms are thin: each constructs entities from the coordinator’s discovered objects and reads state from the live state dict. Service handlers in services.py translate HA service calls into client method calls.

helpers.py is a strict no-HA-imports zone. Every translation between Omni’s wire encoding and HA’s UI encoding (zone-type → device-class, brightness conversion, HVAC mode round-trip, alarm state) lives there as a pure function. 61 unit tests cover it; they run in <100ms because they don’t have to boot HA.

The mock panel

Section titled “The mock panel”
omni_pca/mock_panel.py
  ├── MockUnitState, MockAreaState, MockZoneState, MockThermostatState
  ├── user_codes table for security validation
  ├── handshake handler (same crypto as the client)
  ├── opcode handlers:
  │     RequestSystemInformation / SystemStatus
  │     RequestProperties (Zone/Unit/Area/Thermostat/Button)
  │     RequestStatus / RequestExtendedStatus
  │     Command (with state mutation)
  │     ExecuteSecurityCommand (with code validation)
  │     AcknowledgeAlerts
  └── synthesised SystemEvents (opcode 55) on every state change

The mock is a TCP server that runs the controller half of the protocol. Same handshake, same key derivation, same per-block XOR pre-whitening, same CRC, same opcodes. State changes push synthesised SystemEvents packets back to the client with seq=0 (the unsolicited semantics).

The point is not to be a complete production simulator (it is not — many opcodes are stubbed or unimplemented). The point is to be a bidirectionally faithful protocol counterpart for the surface the library actually uses, so the test suite can prove the stack roundtrips without needing real hardware.

The HA test harness

Section titled “The HA test harness”

pytest-homeassistant-custom-component runs a real Home Assistant in-process per test. Tests boot HA, run the omni_pca config flow against the mock panel, exercise services, and assert on entity state. The full HA-side suite is 12 tests and runs in 0.74 seconds.

The combined test surface:

  • Library unit tests — crypto KAT vectors, CRC, packet/message ser-de, .pca decrypt, command payloads, event parsing.
  • Library e2e tests (17) — OmniClient ↔ MockPanel over real TCP. Proves the handshake, encryption, framing, and sequencing all agree bidirectionally.
  • HA helpers unit tests (61) — pure-function translations, no HA imports.
  • HA-side integration tests (12) — real in-process HA driving the integration against MockPanel.

351 tests pass, 1 skipped (a gitignored .pca fixture). Ruff clean.

The dev docker stack

Section titled “The dev docker stack”
dev/
├── docker-compose.yml     HA 2026.5 + MockPanel sidecar
├── Makefile               make dev-up / dev-down / dev-logs / dev-reset
├── run_mock_panel.py      Long-running mock seeded with realistic data
└── screenshot.py          Onboard HA via REST + drive playwright for screenshots

make dev-up brings up real Home Assistant in a container with the integration mounted read-only, plus a sidecar running the mock panel on port 14369. screenshot.py POSTs to HA’s onboarding API, runs the config flow via REST, and uses headless playwright to deep-link six pages for the README screenshots. The whole flow is 100% scripted; no manual clicking.

Request lifecycle

Section titled “Request lifecycle”

What happens when a user toggles a light in the HA UI:

1. HA UI emits         light.turn_on(entity_id="light.front_porch", brightness=128)
2. light.py            OmniLight.async_turn_on() called
3. helpers.py          ha_brightness_to_omni_percent(128) → 50
4. coordinator         await client.set_unit_level(unit_index=12, percent=50)
5. client.py           execute_command(Command.UNIT_LEVEL, parameter1=50, parameter2=12)
6. client.py           build payload [0x09, 0x32, 0x00, 0x0C]
7. connection.py       wrap as inner Message (StartChar + length + payload + CRC)
8. crypto.py           zero-pad to 16, XOR-whiten with seq, AES-encrypt
9. connection.py       frame as 4-byte header + ciphertext, asyncio.write
10. -- TCP -->         to MockPanel (or real panel) at 192.168.1.9:4369
11. mock_panel.py      read header + first 16 bytes, AES-decrypt, peek MessageLength
12. mock_panel.py      read rest, AES-decrypt + un-whiten, parse Command
13. mock_panel.py      mutate MockUnitState[12].state = 150 (=100+50)
14. mock_panel.py      send Ack reply on the same seq number
15. mock_panel.py      synthesise SystemEvents (opcode 55) with UnitStateChanged
16. -- TCP <--         to client
17. connection.py      reader task: classify Ack as solicited → resolve Future for step (4)
18. connection.py      reader task: classify SystemEvents as unsolicited → push to queue
19. events.py          EventStream.__anext__() yields UnitStateChanged
20. coordinator        background task receives event, patches state dict
21. coordinator        async_set_updated_data() fires
22. light.py           OmniLight._handle_coordinator_update() reads new state
23. HA UI              re-renders the light card

Steps 4-13 happen in <50ms over a real TCP socket. Steps 15-22 happen in a similar window. The user sees the light card update on the UI essentially immediately.