From 6338a286e7ae4b24c405f79a9805709641806796 Mon Sep 17 00:00:00 2001 From: Bartlomiej Kobus Date: Thu, 6 Aug 2026 08:02:40 +0200 Subject: [PATCH] feat: expose gateBox extra button output as a button entity --- blebox_uniapi/box_types.py | 10 +- blebox_uniapi/button.py | 102 ++++++++++++++--- blebox_uniapi/cover.py | 31 +++++- tests/test_button.py | 219 ++++++++++++++++++++++++++++++++++++- tests/test_cover.py | 72 ++++++++++++ 5 files changed, 410 insertions(+), 24 deletions(-) diff --git a/blebox_uniapi/box_types.py b/blebox_uniapi/box_types.py index 0fb22a9..fde922d 100644 --- a/blebox_uniapi/box_types.py +++ b/blebox_uniapi/box_types.py @@ -1,3 +1,4 @@ +from .button import GateBoxSecondOutput, TvLift from .cover import Gate, GateBox, GateBoxB, Shutter from typing import Union, Any @@ -50,9 +51,8 @@ def get_latest_api_level(product_type: str) -> Union[dict, int]: "set": lambda command: ("GET", f"/s/c/{command}") }, # dictionary with interaction methods "buttons": [ - "tvLift", - {"lift": ""}, - ], # key used to set platform, list elements used in cls init, e.g. [, {"path": "state_value"}] + ["tvLift", {}, TvLift], + ], # key used to set platform, list elements used in cls init, e.g. [, {"path": "state_value"}, ] } }, # airSensor @@ -135,11 +135,15 @@ def get_latest_api_level(product_type: str) -> Union[dict, int]: { "position": "gate.currentPos", "gate_type": "gate.gateType", + "extra_button_type": "gate.extraButtonType", }, "gatebox", GateBoxB, ] ], + "buttons": [ + ["second_output", {}, GateBoxSecondOutput], + ], }, }, # gateController diff --git a/blebox_uniapi/button.py b/blebox_uniapi/button.py index 8a8f343..14eef12 100644 --- a/blebox_uniapi/button.py +++ b/blebox_uniapi/button.py @@ -1,5 +1,6 @@ +from .cover import GateBoxControlType, GateBoxExtraButtonType from .feature import Feature -from typing import TYPE_CHECKING, Optional +from typing import TYPE_CHECKING, Any, Optional, Type from enum import Enum, auto @@ -23,33 +24,98 @@ class ControlType(Enum): CLOSE = auto() +class Buttons: + """Buttons defines how a product derives its buttons from extended state.""" + + @staticmethod + def many_from_extended_state( + button_cls: Type["Button"], + product: "Box", + alias: str, + methods: dict, + extended_state: dict, + ) -> list["Button"]: + raise NotImplementedError # pragma: no cover + + +class TvLift(Buttons): + @staticmethod + def many_from_extended_state( + button_cls: Type["Button"], + product: "Box", + alias: str, + methods: dict, + extended_state: dict, + ) -> list["Button"]: + control_type = extended_state.get("tvLift", {}).get("controlType") + endpoints = TV_LIFT_CONTROL_TYPES_API.get(control_type, {}) + return [ + button_cls(product, f"{alias}_{endpoint}", methods, endpoint) + for endpoint in endpoints.values() + ] + + +class GateBoxSecondOutput(Buttons): + """GateBoxSecondOutput exposes the gateBox extra button output as a button. + + WALK_IN and OTHER share one button name because what the output does is a + device setting the user can change at any time, not a hardware trait. + """ + + @staticmethod + def many_from_extended_state( + button_cls: Type["Button"], + product: "Box", + alias: str, + methods: dict, + extended_state: dict, + ) -> list["Button"]: + gate = extended_state.get("gate", {}) + if gate.get("extraButtonType") not in ( + GateBoxExtraButtonType.WALK_IN, + GateBoxExtraButtonType.OTHER, + ): + return [] + + # Api levels below 20230102 report no mode and cannot be wired that way. + if gate.get("openCloseMode") == GateBoxControlType.OPEN_CLOSE: + return [] + + return [button_cls(product, alias, methods, "second_output", "secondary")] + + class Button(Feature): def __init__( - self, product: "Box", alias: str, methods: dict, query_string: str + self, + product: "Box", + alias: str, + methods: dict, + query_string: str, + api_command: str = "set", ) -> None: super().__init__(product, alias, methods) self._device_class = "UPDATE" self._query_string: str = query_string + self._api_command: str = api_command @classmethod - def many_from_config(cls, product, box_type_config, extended_state): - object_list = list() - if len(box_type_config) > 0: - alias = box_type_config[0] - if isinstance(extended_state, dict) and extended_state is not None: - lift_mode = extended_state.get("tvLift", {}).get("controlType", None) - for row in TV_LIFT_CONTROL_TYPES_API[lift_mode].items(): - indicator, endpoint = row - object_list.append( - cls(product, alias + "_" + endpoint, {}, endpoint) - ) - - return object_list - else: + def many_from_config( + cls, product: "Box", box_type_config: list, extended_state: Any + ) -> list["Button"]: + if not isinstance(extended_state, dict): return [] - async def set(self): - await self.async_api_command("set", self.query_string) + features: list[Button] = [] + for alias, methods, buttons in box_type_config: + features.extend( + buttons.many_from_extended_state( + cls, product, alias, methods, extended_state + ) + ) + return features + + async def set(self) -> None: + await self.async_api_command(self._api_command, self.query_string) def after_update(self) -> None: pass diff --git a/blebox_uniapi/cover.py b/blebox_uniapi/cover.py index a6c7295..da28547 100644 --- a/blebox_uniapi/cover.py +++ b/blebox_uniapi/cover.py @@ -65,6 +65,19 @@ class GateBoxControlType(IntEnum): OPEN_CLOSE = 2 +class GateBoxExtraButtonType(IntEnum): + """GateBoxExtraButtonType defines gateBox extra button semantics. + + Only meaningful outside OPEN_CLOSE mode, which takes the extra button output + for the close action and leaves this field unused. + """ + + DISABLED = 0 + STOP = 1 + WALK_IN = 2 + OTHER = 3 + + class GateBoxGateType(IntEnum): """GateBoxGateType defines possible gate/cover types reported by gateBox""" @@ -305,7 +318,7 @@ def read_has_stop(self, alias: str, raw_value: Any, product: "Box") -> bool: if button_type is None: return False - return button_type == 1 + return button_type == GateBoxExtraButtonType.STOP class GateBoxB(GateBox): @@ -328,9 +341,23 @@ def read_desired(self, alias: str, raw_value: Any, product: "Box") -> Optional[i return raw_value("position") def read_has_stop(self, alias: str, raw_value: Any, product: "Box") -> bool: + if product.last_data is None: + return False + # note: if control type is unknown we assume it is not open/close # and has the stop feature via secondary button command. - return self._control_type != GateBoxControlType.OPEN_CLOSE + if self._control_type == GateBoxControlType.OPEN_CLOSE: + return False + + # stop_command issues the same secondary command as the second_output button. + extra_button_type = raw_value("extra_button_type") + if extra_button_type in ( + GateBoxExtraButtonType.WALK_IN, + GateBoxExtraButtonType.OTHER, + ): + return False + + return True def read_cover_type( self, alias: str, raw_value: Any, product: "Box" diff --git a/tests/test_button.py b/tests/test_button.py index 384a4ac..5e76b7b 100644 --- a/tests/test_button.py +++ b/tests/test_button.py @@ -3,7 +3,15 @@ from blebox_uniapi.button import Button from blebox_uniapi.box import Box -from blebox_uniapi.box_types import BOX_TYPE_CONF +from blebox_uniapi.box_types import BOX_TYPE_CONF, get_latest_api_level +from blebox_uniapi.cover import GateBoxControlType, GateBoxExtraButtonType + +from .conftest import CommonEntity, DefaultBoxTest, future_date + + +class BleBoxButtonEntity(CommonEntity): + async def async_press(self): + return await self._feature.set() @pytest.fixture @@ -47,3 +55,212 @@ async def test_tv_lift_1_box_pressed(tv_lift_box_1: Button, product: Box): await tv_lift_box_1.set() product.async_api_command.assert_called_with("set", "close_or_stop") assert tv_lift_box_1.control_type + + +def gate_box_buttons(product: Box, extended_state) -> list[Button]: + product.type = "gateBox" + return Button.many_from_config( + product, + BOX_TYPE_CONF["gateBox"][20200831]["buttons"], + extended_state=extended_state, + ) + + +@pytest.fixture +def gate_box_second_output(product: Box) -> Button: + many = gate_box_buttons( + product, + { + "gate": { + "extraButtonType": GateBoxExtraButtonType.WALK_IN, + "openCloseMode": GateBoxControlType.STEP_BY_STEP, + } + }, + ) + assert len(many) == 1 + return many[0] + + +async def test_gate_box_second_output_pressed( + gate_box_second_output: Button, product: Box +): + await gate_box_second_output.set() + product.async_api_command.assert_called_with("secondary", "second_output") + + +def test_gate_box_second_output_alias(gate_box_second_output: Button): + assert gate_box_second_output.alias == "second_output" + assert gate_box_second_output.control_type is None + + +@pytest.mark.parametrize( + "extra_button_type", + [GateBoxExtraButtonType.WALK_IN, GateBoxExtraButtonType.OTHER], +) +@pytest.mark.parametrize( + "open_close_mode", + [GateBoxControlType.STEP_BY_STEP, GateBoxControlType.ONLY_OPEN], +) +def test_gate_box_second_output_created( + product: Box, + extra_button_type: GateBoxExtraButtonType, + open_close_mode: GateBoxControlType, +): + many = gate_box_buttons( + product, + { + "gate": { + "extraButtonType": extra_button_type, + "openCloseMode": open_close_mode, + } + }, + ) + assert len(many) == 1 + + +@pytest.mark.parametrize( + "extra_button_type", + [GateBoxExtraButtonType.WALK_IN, GateBoxExtraButtonType.OTHER], +) +def test_gate_box_second_output_created_without_open_close_mode( + product: Box, extra_button_type: GateBoxExtraButtonType +): + """Test api levels below 20230102, which never report openCloseMode.""" + + many = gate_box_buttons(product, {"gate": {"extraButtonType": extra_button_type}}) + assert len(many) == 1 + + +@pytest.mark.parametrize( + "gate", + [ + { + "extraButtonType": GateBoxExtraButtonType.WALK_IN, + "openCloseMode": GateBoxControlType.OPEN_CLOSE, + }, + { + "extraButtonType": GateBoxExtraButtonType.OTHER, + "openCloseMode": GateBoxControlType.OPEN_CLOSE, + }, + { + "extraButtonType": GateBoxExtraButtonType.DISABLED, + "openCloseMode": GateBoxControlType.STEP_BY_STEP, + }, + { + "extraButtonType": GateBoxExtraButtonType.STOP, + "openCloseMode": GateBoxControlType.STEP_BY_STEP, + }, + {"extraButtonType": GateBoxExtraButtonType.DISABLED}, + {"extraButtonType": GateBoxExtraButtonType.STOP}, + {"openCloseMode": GateBoxControlType.STEP_BY_STEP}, + {}, + ], +) +def test_gate_box_second_output_not_created(product: Box, gate: dict): + assert gate_box_buttons(product, {"gate": gate}) == [] + + +@pytest.mark.parametrize( + ("box_type", "api_level"), [("gateBox", 20200831), ("tvLiftBox", 20200518)] +) +@pytest.mark.parametrize("extended_state", [None, {}, "not a dict"]) +def test_no_buttons_without_usable_extended_state( + product: Box, box_type: str, api_level: int, extended_state +): + product.type = box_type + many = Button.many_from_config( + product, + BOX_TYPE_CONF[box_type][api_level]["buttons"], + extended_state=extended_state, + ) + assert many == [] + + +class TestGateBoxSecondOutput(DefaultBoxTest): + """Tests for a gateBox exposing its extra button output as a button.""" + + DEVCLASS = "buttons" + ENTITY_CLASS = BleBoxButtonEntity + DEV_INFO_PATH = "state/extended" + + DEVICE_INFO = { + "device": { + "deviceName": "My gateBox 1", + "type": "gateBox", + "product": "gateBox", + "fv": "0.1010", + "hv": "9.1d", + "id": "1afe34d27e4f", + "apiLevel": "20230102", + } + } + DEVICE_INFO_FUTURE = { + "device": {**DEVICE_INFO["device"], "apiLevel": future_date()} + } + DEVICE_INFO_LATEST = { + "device": { + **DEVICE_INFO["device"], + "apiLevel": get_latest_api_level("gateBox"), + } + } + DEVICE_INFO_UNSUPPORTED = DEVICE_INFO + DEVICE_INFO_UNSPECIFIED_API = None # already handled as default case + + STATE_DEFAULT = { + "gate": { + "currentPos": 0, + "openCloseMode": 0, + "gateType": 1, + "gatePulseTimeMs": 1500, + "gateOutputState": 0, + "extraButtonType": 2, + "extraButtonPulseTimeMs": 1500, + "extraButtonOutputState": 0, + "inputsType": 0, + } + } + DEVICE_EXTENDED_INFO = STATE_DEFAULT + DEVICE_EXTENDED_INFO_PATH = "/state/extended" + + async def test_init(self, aioclient_mock): + """Test that a usable extra button output yields exactly one button.""" + + await self.allow_get_info(aioclient_mock) + entities = await self.async_entities(aioclient_mock) + + assert len(entities) == 1 + entity = entities[0] + assert entity.name == "My gateBox 1 (gateBox#second_output)" + assert entity.unique_id == "BleBox-gateBox-1afe34d27e4f-second_output" + + async def test_pressed(self, aioclient_mock): + """Test that pressing the button pulses the extra button output.""" + + await self.allow_get_info(aioclient_mock) + entity = (await self.async_entities(aioclient_mock))[0] + + self.allow_get(aioclient_mock, "/s/s", self.STATE_DEFAULT) + await entity.async_press() + + +class TestGateBoxSecondOutputWithoutOpenCloseMode(TestGateBoxSecondOutput): + """Tests for a gateBox on an api level that never reports openCloseMode.""" + + DEVICE_INFO = { + "device": { + **TestGateBoxSecondOutput.DEVICE_INFO["device"], + "apiLevel": "20220713", + } + } + DEVICE_INFO_UNSUPPORTED = DEVICE_INFO + + STATE_DEFAULT = { + "gate": { + "currentPos": 0, + "gateType": 1, + "gatePulseTimeMs": 1500, + "extraButtonType": 2, + "extraButtonPulseTimeMs": 1500, + } + } + DEVICE_EXTENDED_INFO = STATE_DEFAULT diff --git a/tests/test_cover.py b/tests/test_cover.py index f49d7cc..9f5b9b1 100644 --- a/tests/test_cover.py +++ b/tests/test_cover.py @@ -732,6 +732,7 @@ async def test_stop(self, aioclient_mock): entity = await self.updated(aioclient_mock, self.STATE_STOPPED) self.assert_state(entity, STATE_OPEN) + assert entity.supported_features & SUPPORT_STOP async def test_closed(self, aioclient_mock): """Test cover closed.""" @@ -745,6 +746,77 @@ async def test_unkown_position(self, aioclient_mock): self.assert_state(entity, None) +class TestGateBoxBSecondOutput(CoverTest): + """Tests for a GateBoxB whose extra button output is exposed as a button.""" + + DEV_INFO_PATH = "state/extended" + + DEVICE_INFO = json.loads(""" + { + "device": { + "deviceName":"My gateBox 1", + "type":"gateBox", + "product":"gateBox", + "hv":"9.1d", + "fv":"0.1010", + "universe":0, + "apiLevel":"20230102", + "id":"1afe34d27e4f", + "ip":"192.168.4.1", + "availableFv":null + } + } + """) + + DEVICE_INFO_FUTURE = jmerge(DEVICE_INFO, patch_version(future_date())) + DEVICE_INFO_LATEST = jmerge( + DEVICE_INFO, patch_version(get_latest_api_level("gateBox")) + ) + DEVICE_INFO_UNSUPPORTED = DEVICE_INFO + + DEVICE_INFO_UNSPECIFIED_API = None # already handled as default case + + STATE_DEFAULT = json.loads(""" + { + "gate": { + "currentPos": 0, + "openCloseMode": 0, + "gateType": 1, + "gatePulseTimeMs": 1500, + "gateOutputState": 0, + "extraButtonType": 2, + "extraButtonPulseTimeMs": 1500, + "extraButtonOutputState": 0, + "inputsType": 0 + } + } + """) + + DEVICE_EXTENDED_INFO_PATH = "/state/extended" + DEVICE_EXTENDED_INFO = STATE_DEFAULT + + async def test_init(self, aioclient_mock): + """Test cover default state.""" + + await self.allow_get_info(aioclient_mock) + entity = (await self.async_entities(aioclient_mock))[0] + + assert entity.name == "My gateBox 1 (gateBox#position)" + assert entity.unique_id == "BleBox-gateBox-1afe34d27e4f-position" + assert entity.device_class == DEVICE_CLASS_DOOR + assert entity.supported_features & SUPPORT_OPEN + assert entity.supported_features & SUPPORT_CLOSE + + async def test_second_output_claims_stop(self, aioclient_mock): + """Test that a second_output button removes the cover's stop action. + + Both would issue the same secondary command. + """ + + entity = await self.updated(aioclient_mock, self.STATE_DEFAULT) + assert not entity.supported_features & SUPPORT_STOP + + class TestGateController(CoverTest): """Tests for cover devices representing a BleBox gateController."""