Library API

omni_pca is split into a thin protocol layer (crypto, packet, message, connection), a domain layer (models, commands, events, opcodes), a high-level client (client), and a controller-side emulator (mock_panel). This page is the surface index. Source has type hints and docstrings on every public symbol.

`omni_pca.client.OmniClient`

High-level async client. Handles the full secure-session handshake on __aenter__, dispatches solicited replies via Futures, and exposes typed methods that send the right opcode and parse the reply payload into a dataclass. Use as an async context manager.

Lifecycle

Method	Description
`OmniClient(host, port=4369, *, controller_key, timeout=5.0)`	Construct. Doesn’t connect yet.
`async __aenter__()`	Run the four-step handshake; transition to OnlineSecure.
`async __aexit__(...)`	Cancel subscriber task; close the TCP socket.
`connection` (property)	The underlying `OmniConnection` for advanced use.

System info & status

Method	Returns
`await get_system_information()`	`SystemInformation` (model byte, firmware, local phone).
`await get_system_status()`	`SystemStatus` (panel time, battery, area-alarm bytes).

Object discovery

Method	Returns
`await get_object_properties(object_type, index)`	One `*Properties` dataclass for the given object.
`await list_zone_names()`	`dict[int, str]` of named zones.
`await list_unit_names()`	`dict[int, str]` of named units.
`await list_area_names()`	`dict[int, str]` of named areas.

Status queries

Method	Returns
`await get_object_status(object_type, start, end=None)`	List of basic-Status records (fixed per-type sizes).
`await get_extended_status(object_type, start, end=None)`	List of ExtendedStatus records (length-prefixed).

Commands

Method	Wire mapping
`await execute_command(command, parameter1=0, parameter2=0)`	Generic Command (opcode 20).
`await execute_security_command(area, mode, code)`	ExecuteSecurityCommand (opcode 74).
`await acknowledge_alerts()`	AcknowledgeAlerts (opcode 60).
`await turn_unit_on(index)`	`Command.UNIT_ON`
`await turn_unit_off(index)`	`Command.UNIT_OFF`
`await set_unit_level(index, percent)`	`Command.UNIT_LEVEL`, parameter1 = percent.
`await bypass_zone(index, code=0)`	`Command.BYPASS_ZONE`
`await restore_zone(index, code=0)`	`Command.RESTORE_ZONE`
`await set_thermostat_system_mode(index, mode)`	`Command.SET_THERMOSTAT_SYSTEM_MODE`
`await set_thermostat_fan_mode(index, mode)`	`Command.SET_THERMOSTAT_FAN_MODE`
`await set_thermostat_hold_mode(index, mode)`	`Command.SET_THERMOSTAT_HOLD_MODE`
`await set_thermostat_heat_setpoint_raw(index, raw)`	`Command.SET_THERMOSTAT_HEAT_SETPOINT`
`await set_thermostat_cool_setpoint_raw(index, raw)`	`Command.SET_THERMOSTAT_COOL_SETPOINT`
`await execute_button(index)`	`Command.EXECUTE_BUTTON`
`await execute_program(index)`	`Command.EXECUTE_PROGRAM`
`await show_message(index, beep=True)`	`Command.SHOW_MESSAGE_WITH_BEEP` / `_NO_BEEP`
`await clear_message(index)`	`Command.CLEAR_MESSAGE`

Events

Method	Returns
`events()`	`AsyncIterator[SystemEvent]` over decoded push notifications.
`await subscribe(callback)`	Background task that hands every unsolicited `Message` to `callback`.

`omni_pca.connection.OmniConnection`

Low-level primitives. You usually want OmniClient instead.

Method	Description
`await connect()`	Open TCP, run secure-session handshake.
`await close()`	Cancel reader task, send terminator, close socket.
`await request(opcode, payload=b"")`	Send one inner `Message`, return the matching reply.
`unsolicited()`	`AsyncIterator[Message]` of all push messages (any opcode).
`is_connected` (property)	True between successful `connect()` and `close()`.

Custom exceptions: OmniConnectionError, HandshakeError, InvalidEncryptionKeyError, ProtocolError, RequestTimeoutError.

`omni_pca.models`

Frozen-slots dataclasses for every Omni object’s Properties + Status payload. Each carries a parse(payload: bytes) -> Self classmethod that decodes one record.

Class	Used for
`SystemInformation`	RequestSystemInformation reply.
`SystemStatus`	RequestSystemStatus reply.
`ZoneProperties` / `ZoneStatus`	Per-zone configuration / live state.
`UnitProperties` / `UnitStatus`	Per-unit (light/relay/scene).
`AreaProperties` / `AreaStatus`	Per-area (security mode + alarms).
`ThermostatProperties` / `ThermostatStatus`	Per-thermostat config + readings.
`ButtonProperties`	Index + name (buttons carry no state).
`ProgramProperties`	Index + raw 14-byte program body + optional remark.
`CodeProperties`	User code slot (name only — digits never exposed).
`MessageProperties`	Stored panel message.
`AuxSensorStatus`	Auxiliary temp/humidity/output sensor.
`AudioZoneProperties` / `AudioZoneStatus`	Audio zone (power, source, volume, mute).
`AudioSourceProperties` / `AudioSourceStatus`	Audio source metadata stream.
`UserSettingProperties` / `UserSettingStatus`	Programmable user-setting value.

Plus enums: ObjectType, SecurityMode, HvacMode, FanMode, HoldMode, ThermostatKind, ZoneType, UserSettingKind.

Two temperature converters mirror clsText.DecodeTempRaw:

omni_temp_to_celsius(raw)     # °C = raw / 2 - 40
omni_temp_to_fahrenheit(raw)  # °F = int(raw * 9 / 10 + 0.5) - 40   (matches PC Access display)

Dispatch tables: OBJECT_TYPE_TO_PROPERTIES, OBJECT_TYPE_TO_STATUS. Type unions: PropertiesReply, StatusReply.

`omni_pca.commands.Command`

The Command IntEnum (64 values) sourced from enuUnitCommand.cs. Every member cites the line in the C# source where its byte value is defined. Members are grouped: unit/lighting (UNIT_OFF..RADIO_RA_PHANTOM_ON), security (SECURITY_OFF..SECURITY_ARMING_NIGHT_DELAYED), energy (ENERGY_OFF/ENERGY_ON), thermostat (SET_THERMOSTAT_HEAT_SETPOINT.. SET_THERMOSTAT_DEHUMIDIFY_SETPOINT), display messages (SHOW_MESSAGE_WITH_BEEP..EMAIL_MESSAGE), scenes/misc (SCENE_OFF..STOP), audio (AUDIO_ZONE..AUDIO_KEY_PRESS).

Source: omni_pca/commands.py.

Companion types:

SecurityCommandResponse — status byte returned by ExecuteSecurityCommand (SUCCESS, INVALID_CODE, INVALID_SECURITY_MODE, INVALID_AREA, ZONES_NOT_READY, INSTALLER_RESTORE_NEEDED, CODE_LOCKED_OUT, INVALID).
CommandFailedError(ProtocolError) — raised on Nak or non-zero ExecuteSecurityCommandResponse status. failure_code attribute carries the raw status byte.

`omni_pca.events.SystemEvent`

Base class for every typed push event. The panel batches notifications into a single SystemEvents (opcode 55) message; each batched event is a single 16-bit big-endian word in the message payload, classified by bit-field into one of 26 typed subclasses:

UserMacroButton, ProLinkMessage, CentraLiteSwitch, AlarmActivated, AlarmCleared, ZoneStateChanged, UnitStateChanged, X10CodeReceived, AllOnOff, PhoneLineDead, PhoneLineRinging, PhoneLineOffHook, PhoneLineOnHook, AcLost, AcRestored, BatteryLow, BatteryRestored, DcmTrouble, DcmOk, EnergyCostChanged, CameraTrigger, AccessReaderEvent, UpbLinkEvent, ArmingChanged, plus an UnknownEvent catch-all.

Helpers:

parse_events(message) — decode one SystemEvents message into a list of typed events.
EventStream(source) — async iterator that flattens batches across an underlying connection’s unsolicited() stream. Yields one typed event per iterator step.
EVENT_REGISTRY: dict[int, type[SystemEvent]] — discriminator-to-class map for routing.

`omni_pca.opcodes`

Three IntEnums, byte-exact to the C# decompilation:

PacketType — outer-frame packet type (12 values).
OmniLinkMessageType — inner v1 (legacy) opcode (104 values).
OmniLink2MessageType — inner v2 (modern TCP) opcode (83 values).

Plus ConnectionType (None/Modem/Serial/Network_UDP/Network_TCP) and ProtocolVersion (V1/V2).

`omni_pca.mock_panel.MockPanel`

A controller-side emulator: a TCP server that speaks the same Omni-Link II protocol as a real panel. Handshake, encryption, framing all match bidirectionally. Used by the test suite (17 e2e tests connect a real OmniClient to a real MockPanel over real TCP) and by the dev docker stack for clicking around the HA integration without hardware.

import asyncio
from omni_pca.mock_panel import MockPanel

async def main():
    async with MockPanel(
        controller_key=bytes.fromhex("000102030405060708090a0b0c0d0e0f"),
    ).serve(port=14369):
        await asyncio.sleep(3600)  # serve until cancelled

asyncio.run(main())

The mock seeds itself with a small set of zones, units, areas, thermostats, buttons, codes; mutates state on Command / ExecuteSecurityCommand; and synthesizes SystemEvents push messages on every state change.

CLI

The package ships a omni-pca console script:

Command	Description
`omni-pca decode-pca FILE [--field NAME] [--include-pii]`	Decrypt and parse a `.pca` file; print one field or the whole structure.
`omni-pca mock-panel [--port 14369] [--key HEX]`	Run the mock panel as a standalone server.
`omni-pca version`	Print package version.

For full details, the source has type hints and docstrings everywhere; see git.supported.systems/warehack.ing/omni-pca.