From 93b7e1f604c39712834f648c45df53b7c9f2cebc Mon Sep 17 00:00:00 2001 From: Ryan Malloy Date: Sun, 10 May 2026 15:09:31 -0600 Subject: [PATCH] Mock: add Thermostat + Button RequestProperties handlers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The HA coordinator walks ObjectType.THERMOSTAT (6) and ObjectType.BUTTON (3) via raw RequestProperties to discover them — the high-level get_object_properties() path only knows zones/units/areas in v1.0. The mock was returning Nak for both, which made HA discover zero thermostats and zero buttons no matter how MockState was seeded. src/omni_pca/mock_panel.py: - New MockButtonState dataclass (just a name) - MockState gains buttons: dict[int, MockButtonState] (with the same bare-string -> dataclass __post_init__ promotion as the others) - _OBJ_BUTTON = 3, _BUTTON_NAME_LEN = 12, _THERMOSTAT_NAME_LEN = 12 constants - thermostat_name_bytes() / button_name_bytes() helpers - _build_thermostat_properties() emits the 23-byte Properties body matching ThermostatProperties.parse offsets (object number BE u16, communicating flag, current temp, heat/cool setpoints, system/fan/ hold modes, thermostat type, 12-byte NUL-padded name) - _build_button_properties() emits the 15-byte body (object number BE u16 + 12-byte name) - _reply_properties / _object_store dispatch both new types tests/test_e2e_client_mock.py — two new e2e tests drive raw RequestProperties walks for thermostats and buttons against a seeded mock and assert ThermostatProperties / ButtonProperties parse cleanly, mirroring what the HA coordinator's _walk_properties() does. 333 tests pass (was 331); ruff clean. Mock surface now matches every opcode the HA coordinator and entity platforms actually call. --- src/omni_pca/mock_panel.py | 78 +++++++++++++++++++++++++++++++++++ tests/test_e2e_client_mock.py | 71 +++++++++++++++++++++++++++++++ 2 files changed, 149 insertions(+) diff --git a/src/omni_pca/mock_panel.py b/src/omni_pca/mock_panel.py index 086ffb5..6a42b79 100644 --- a/src/omni_pca/mock_panel.py +++ b/src/omni_pca/mock_panel.py @@ -59,6 +59,7 @@ _log = logging.getLogger(__name__) # enuObjectType (clsOmniLink2.cs / enuObjectType.cs) _OBJ_ZONE = 1 _OBJ_UNIT = 2 +_OBJ_BUTTON = 3 _OBJ_AREA = 5 _OBJ_THERMOSTAT = 6 @@ -66,6 +67,8 @@ _OBJ_THERMOSTAT = 6 _ZONE_NAME_LEN = 15 _UNIT_NAME_LEN = 12 _AREA_NAME_LEN = 12 +_BUTTON_NAME_LEN = 12 +_THERMOSTAT_NAME_LEN = 12 _PHONE_LEN = 24 # Per-object-type record sizes for the basic Status (opcode 35) reply. @@ -156,6 +159,13 @@ class MockZoneState: return val & 0xFF +@dataclass +class MockButtonState: + """One programmable button macro (no live state — buttons just fire programs).""" + + name: str = "" + + @dataclass class MockThermostatState: """One programmable thermostat. Defaults are sane Omni Pro II values.""" @@ -197,6 +207,7 @@ class MockState: units: dict[int, MockUnitState] = field(default_factory=dict) areas: dict[int, MockAreaState] = field(default_factory=dict) thermostats: dict[int, MockThermostatState] = field(default_factory=dict) + buttons: dict[int, MockButtonState] = field(default_factory=dict) # User-code table for ExecuteSecurityCommand validation. # Mapping is ``{code_index: 4-digit pin}``; the panel returns the @@ -230,6 +241,7 @@ class MockState: self.units = _promote_dict(self.units, MockUnitState) self.areas = _promote_dict(self.areas, MockAreaState) self.thermostats = _promote_dict(self.thermostats, MockThermostatState) + self.buttons = _promote_dict(self.buttons, MockButtonState) # ---- name-bytes helpers (kept for back-compat with old callers) ----- @@ -245,6 +257,14 @@ class MockState: a = self.areas.get(idx) return _name_bytes(a.name if a else "", _AREA_NAME_LEN) + def thermostat_name_bytes(self, idx: int) -> bytes: + t = self.thermostats.get(idx) + return _name_bytes(t.name if t else "", _THERMOSTAT_NAME_LEN) + + def button_name_bytes(self, idx: int) -> bytes: + b = self.buttons.get(idx) + return _name_bytes(b.name if b else "", _BUTTON_NAME_LEN) + def _promote_dict( raw: dict[int, object], @@ -618,6 +638,10 @@ class MockPanel: return self._build_unit_properties(target) if obj_type == _OBJ_AREA: return self._build_area_properties(target) + if obj_type == _OBJ_THERMOSTAT: + return self._build_thermostat_properties(target) + if obj_type == _OBJ_BUTTON: + return self._build_button_properties(target) return _build_nak(OmniLink2MessageType.RequestProperties) def _object_store(self, obj_type: int) -> dict[int, object] | None: @@ -629,6 +653,8 @@ class MockPanel: return self.state.areas # type: ignore[return-value] if obj_type == _OBJ_THERMOSTAT: return self.state.thermostats # type: ignore[return-value] + if obj_type == _OBJ_BUTTON: + return self.state.buttons # type: ignore[return-value] return None def _build_zone_properties(self, index: int) -> Message: @@ -678,6 +704,58 @@ class MockPanel: ) return encode_v2(OmniLink2MessageType.Properties, body) + def _build_thermostat_properties(self, index: int) -> Message: + # Properties.Data layout for Thermostat (Data[0]=opcode, body starts + # at Data[1]. ``payload`` here strips the opcode; payload[i]==Data[i+1]): + # payload[0] object type (Thermostat = 6) + # payload[1..2] object number (BE u16) + # payload[3] communicating flag + # payload[4] temperature raw + # payload[5] heat setpoint raw + # payload[6] cool setpoint raw + # payload[7] mode + # payload[8] fan mode + # payload[9] hold mode + # payload[10] thermostat type + # payload[11..22] 12-byte name + t = self.state.thermostats.get(index) + body = ( + bytes( + [ + _OBJ_THERMOSTAT, + (index >> 8) & 0xFF, + index & 0xFF, + t.status if t else 0, # communicating flag + t.temperature_raw if t else 0, + t.heat_setpoint_raw if t else 0, + t.cool_setpoint_raw if t else 0, + t.system_mode if t else 0, + t.fan_mode if t else 0, + t.hold_mode if t else 0, + 1, # thermostat type: AUTO_HEAT_COOL + ] + ) + + self.state.thermostat_name_bytes(index) + ) + return encode_v2(OmniLink2MessageType.Properties, body) + + def _build_button_properties(self, index: int) -> Message: + # Properties.Data layout for Button: + # payload[0] object type (Button = 3) + # payload[1..2] object number (BE u16) + # payload[3..14] 12-byte name (NUL-padded) + body = ( + bytes( + [ + _OBJ_BUTTON, + (index >> 8) & 0xFF, + index & 0xFF, + ] + ) + + self.state.button_name_bytes(index) + ) + return encode_v2(OmniLink2MessageType.Properties, body) + def _build_area_properties(self, index: int) -> Message: # Properties.Data for Area: # [0]=opcode, [1]=ObjectType, [2..3]=ObjectNumber, diff --git a/tests/test_e2e_client_mock.py b/tests/test_e2e_client_mock.py index 5d20aa2..ae5833b 100644 --- a/tests/test_e2e_client_mock.py +++ b/tests/test_e2e_client_mock.py @@ -18,6 +18,7 @@ from omni_pca.connection import HandshakeError from omni_pca.events import ArmingChanged, UnitStateChanged from omni_pca.mock_panel import ( MockAreaState, + MockButtonState, MockPanel, MockState, MockThermostatState, @@ -272,6 +273,76 @@ async def test_e2e_unit_command_pushes_unit_state_changed_event() -> None: assert ev.is_on is True +async def test_e2e_thermostat_properties_discovery() -> None: + """The HA coordinator walks thermostats via raw RequestProperties; ensure + the mock answers and the response parses cleanly into ThermostatProperties. + """ + from omni_pca.models import ObjectType as ObjType + from omni_pca.models import ThermostatProperties + from omni_pca.opcodes import OmniLink2MessageType + + state = MockState( + thermostats={ + 1: MockThermostatState(name="LIVING_ROOM"), + 3: MockThermostatState(name="MASTER"), + } + ) + panel = MockPanel(controller_key=CONTROLLER_KEY, state=state) + async with ( + panel.serve() as (host, port), + OmniClient(host=host, port=port, controller_key=CONTROLLER_KEY) as cli, + ): + found: dict[int, str] = {} + cursor = 0 + for _ in range(10): + payload = bytes( + [int(ObjType.THERMOSTAT), (cursor >> 8) & 0xFF, cursor & 0xFF, 1, 0, 0, 0] + ) + reply = await cli.connection.request( + OmniLink2MessageType.RequestProperties, payload + ) + if reply.opcode == int(OmniLink2MessageType.EOD): + break + t = ThermostatProperties.parse(reply.payload) + found[t.index] = t.name + cursor = t.index + assert found == {1: "LIVING_ROOM", 3: "MASTER"} + + +async def test_e2e_button_properties_discovery() -> None: + """Same idea for button discovery (the HA coordinator drives this too).""" + from omni_pca.models import ButtonProperties + from omni_pca.models import ObjectType as ObjType + from omni_pca.opcodes import OmniLink2MessageType + + state = MockState( + buttons={ + 1: MockButtonState(name="GOOD_MORNING"), + 5: MockButtonState(name="MOVIE_MODE"), + } + ) + panel = MockPanel(controller_key=CONTROLLER_KEY, state=state) + async with ( + panel.serve() as (host, port), + OmniClient(host=host, port=port, controller_key=CONTROLLER_KEY) as cli, + ): + found: dict[int, str] = {} + cursor = 0 + for _ in range(10): + payload = bytes( + [int(ObjType.BUTTON), (cursor >> 8) & 0xFF, cursor & 0xFF, 1, 0, 0, 0] + ) + reply = await cli.connection.request( + OmniLink2MessageType.RequestProperties, payload + ) + if reply.opcode == int(OmniLink2MessageType.EOD): + break + b = ButtonProperties.parse(reply.payload) + found[b.index] = b.name + cursor = b.index + assert found == {1: "GOOD_MORNING", 5: "MOVIE_MODE"} + + async def test_e2e_acknowledge_alerts() -> None: panel = MockPanel(controller_key=CONTROLLER_KEY) async with (