From 697d7e54d0d1336b5bc83b837f7e48567ea58ed5 Mon Sep 17 00:00:00 2001 From: Pepijn Nichting Date: Tue, 13 Oct 2026 12:56:10 +0000 Subject: [PATCH] Fix cover position scale and native endpoint movements --- README.md | 29 ++++-- .../blebox_smartwindowbox/cover.py | 51 ++++++----- .../blebox_smartwindowbox/manifest.json | 14 +-- tests/test_cover.py | 89 +++++++++++++++++++ 4 files changed, 151 insertions(+), 32 deletions(-) create mode 100644 tests/test_cover.py diff --git a/README.md b/README.md index 46c9012..716cd6c 100644 --- a/README.md +++ b/README.md @@ -42,15 +42,17 @@ The smartWindowBox motor is exposed as a standard Home Assistant **cover entity* * `Open` * `Close` * `Stop` -* `Set position (0=open, 100=closed)` +* `Set position (0=closed, 100=open)` -Since the official BleBox API does **not** support direct position setting, this integration emulates it by: +For intermediate positions, the integration emulates position setting because the BleBox API does **not** support direct position setting: 1. Sending open or close command 2. Polling device state 3. Stopping the motor once the target position is reached -This method is precise and tuned for real-world motor timing. +Fully closed (0%) and fully open (100%) use the native close/open commands. The device completes its calibrated travel without an early stop based on a reported percentage. + +All exposed position attributes use the Home Assistant scale (0% closed, 100% open). `device_position` retains the raw BleBox scale (0% open, 100% closed). Existing automations using numeric positions must invert their targets and comparisons (`100 - old_position`) when upgrading. --- @@ -192,8 +194,8 @@ You can: Position scale: ``` -0 = Fully Open -100 = Fully Closed +0 = Fully Closed +100 = Fully Open ``` --- @@ -307,6 +309,21 @@ This value has been tuned for reliable real-world accuracy. --- +# Development checks + +With Home Assistant installed in the active Python environment, run the cover +regression tests from the repository root: + +```sh +python -m unittest discover -s tests -v +``` + +These cover the Home Assistant position scale, intermediate movements, and native +endpoint commands. They use simulated responses; physical travel still needs to +be verified on a real device. The GitHub workflow also runs Hassfest validation. + +--- + # Future Improvements (Ideas) * HACS support @@ -319,4 +336,4 @@ This value has been tuned for reliable real-world accuracy. # License -MIT License \ No newline at end of file +MIT License diff --git a/custom_components/blebox_smartwindowbox/cover.py b/custom_components/blebox_smartwindowbox/cover.py index 9a22c9b..4bc7150 100644 --- a/custom_components/blebox_smartwindowbox/cover.py +++ b/custom_components/blebox_smartwindowbox/cover.py @@ -32,6 +32,14 @@ def _motor_list(data: dict[str, Any]) -> list[dict[str, Any]]: return (data.get("window") or {}).get("motors") or [] +def _ha_position(device_position: Any) -> int | None: + """Convert BleBox's closed percentage to Home Assistant's open percentage.""" + try: + return 100 - int(device_position) + except (TypeError, ValueError): + return None + + async def async_setup_entry( hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback ) -> None: @@ -97,18 +105,14 @@ def name(self) -> str | None: return f"{self._name_prefix} {base}" return str(base) - # Your device scale: - # 0 = fully open, 100 = fully closed + # Home Assistant: 0 = fully closed, 100 = fully open. @property def current_cover_position(self) -> int | None: motor = self._motor() if not motor: return None pos = (motor.get("currentPos") or {}).get("position") - try: - return int(pos) - except (TypeError, ValueError): - return None + return _ha_position(pos) # Your observed mapping: # 0 closing (moving to 100) @@ -131,7 +135,7 @@ def is_closed(self) -> bool | None: pos = self.current_cover_position if pos is None: return None - return pos >= 100 + return pos <= 0 def _full_travel_time_s(self, motor: dict[str, Any], direction: str) -> float: calib = motor.get("calibrationParameters") or {} @@ -240,9 +244,10 @@ def extra_state_attributes(self) -> dict[str, Any]: return { "enabled": motor.get("enabled"), "motor_state": motor.get("state"), - "current_position": cur_i, - "desired_position": des_i, - "favorite_position": fav.get("position"), + "current_position": _ha_position(cur_i), + "device_position": cur_i, + "desired_position": _ha_position(des_i), + "favorite_position": _ha_position(fav.get("position")), "at_favorite": at_fav, "control_type": motor.get("controlType"), "icon_set": motor.get("iconSet"), @@ -252,8 +257,8 @@ def extra_state_attributes(self) -> dict[str, Any]: # movement metadata (any movement) "moving": self._tracked_moving, "move_direction": self._move_direction, - "move_target_position": self._move_target, - "move_start_position": self._move_start_pos, + "move_target_position": _ha_position(self._move_target), + "move_start_position": _ha_position(self._move_start_pos), "move_elapsed_s": move_elapsed_s, "estimated_total_s": estimated_total_s, "estimated_remaining_s": estimated_remaining_s, @@ -283,7 +288,8 @@ async def async_next_step(self) -> None: async def async_set_cover_position(self, **kwargs: Any) -> None: """ Emulate set position by moving and stopping when target reached. - Position scale: 0=open, 100=closed + Position scale: 0=closed, 100=open. + Endpoints use native commands so the device can finish its travel. """ target = kwargs.get("position") if target is None: @@ -295,13 +301,18 @@ async def async_set_cover_position(self, **kwargs: Any) -> None: return target = max(0, min(100, target)) - # Special-case endpoints for better accuracy + # A reported endpoint may precede the physical end of travel. + # Do not stop the motor or skip the command based on cached position. + if target == 0: + await self.async_close_cover() + return if target == 100: - tol = 0 - elif target == 0: - tol = 0 - else: - tol = POSITION_TOLERANCE + await self.async_open_cover() + return + + # Keep the polling loop in the native BleBox coordinate system. + target = 100 - target + tol = POSITION_TOLERANCE motor = self._motor() if not motor: @@ -362,4 +373,4 @@ async def async_set_cover_position(self, **kwargs: Any) -> None: await asyncio.sleep(EXTRA_STOP_DELAY_SEC) await self.coordinator.api.send_motor_command(self._channel, CMD_STOP) await self.coordinator.async_request_refresh() - return \ No newline at end of file + return diff --git a/custom_components/blebox_smartwindowbox/manifest.json b/custom_components/blebox_smartwindowbox/manifest.json index e9b58d8..69a38d3 100644 --- a/custom_components/blebox_smartwindowbox/manifest.json +++ b/custom_components/blebox_smartwindowbox/manifest.json @@ -1,11 +1,13 @@ { "domain": "blebox_smartwindowbox", "name": "BleBox smartWindowBox", - "version": "0.0.1", - "documentation": "https://github.com/Sjnaaf/bleBoxSmartWindowBox/blob/main/README.md", - "requirements": [], - "codeowners": ["@sjnaaf"], + "codeowners": [ + "@sjnaaf" + ], "config_flow": true, + "documentation": "https://github.com/Sjnaaf/bleBoxSmartWindowBox/blob/main/README.md", + "integration_type": "device", "iot_class": "local_polling", - "integration_type": "device" -} \ No newline at end of file + "requirements": [], + "version": "0.0.1" +} diff --git a/tests/test_cover.py b/tests/test_cover.py new file mode 100644 index 0000000..d17b935 --- /dev/null +++ b/tests/test_cover.py @@ -0,0 +1,89 @@ +"""Regression checks for upstream issues #1 and #2. + +Run with: python -m unittest discover -s tests -v +Requires Home Assistant in the active Python environment. +""" +import unittest +from types import SimpleNamespace +from unittest.mock import AsyncMock, patch + +from custom_components.blebox_smartwindowbox.cover import BleBoxMotorCover + + +class CoverTests(unittest.IsolatedAsyncioTestCase): + def setUp(self): + self.motor = { + 'id': 0, 'state': 2, 'currentPos': {'position': 40}, + 'desiredPos': {'position': 80}, 'favPos': {'position': 30}, + } + self.data = {'window': {'motors': [self.motor]}} + self.api = SimpleNamespace( + send_motor_command=AsyncMock(), + window_extended_state=AsyncMock(return_value=self.data), + ) + self.coordinator = SimpleNamespace( + data=self.data, api=self.api, config_entry=None, + async_request_refresh=AsyncMock(), + ) + self.cover = BleBoxMotorCover(self.coordinator, 'entry', 0, None) + + def test_reported_positions_and_closed_state(self): + for raw, expected, closed in [(100, 0, True), (0, 100, False), (40, 60, False)]: + with self.subTest(raw=raw): + self.motor['currentPos']['position'] = raw + self.assertEqual(self.cover.current_cover_position, expected) + self.assertEqual(self.cover.extra_state_attributes['current_position'], expected) + self.assertEqual(self.cover.is_closed, closed) + self.assertEqual(self.cover.extra_state_attributes['device_position'], raw) + + def test_unknown_position(self): + for value in (None, 'invalid'): + self.motor['currentPos']['position'] = value + self.assertIsNone(self.cover.current_cover_position) + self.assertIsNone(self.cover.is_closed) + self.assertIsNone(self.cover.extra_state_attributes['current_position']) + + def test_position_metadata_uses_home_assistant_scale(self): + self.cover._move_target = 80 + self.cover._move_start_pos = 40 + attributes = self.cover.extra_state_attributes + self.assertEqual(attributes['desired_position'], 20) + self.assertEqual(attributes['favorite_position'], 70) + self.assertEqual(attributes['move_target_position'], 20) + self.assertEqual(attributes['move_start_position'], 60) + + async def test_endpoints_do_not_stop_or_skip_motor(self): + # Cached/reported positions near or at an endpoint must still send + # the native command, leaving physical completion to the device. + for target, command in [(0, 'd'), (100, 'u')]: + for raw in (0, 1, 50, 99, 100): + with self.subTest(target=target, raw=raw): + self.api.send_motor_command.reset_mock() + self.motor['currentPos']['position'] = raw + await self.cover.async_set_cover_position(position=target) + self.api.send_motor_command.assert_awaited_once_with(0, command) + self.api.window_extended_state.assert_not_awaited() + + async def test_intermediate_positions_move_in_correct_direction_and_stop(self): + for target, initial, polled, command in [(70, 60, 30, 'u'), (20, 40, 80, 'd')]: + with self.subTest(target=target): + self.motor['currentPos']['position'] = initial + self.api.send_motor_command.reset_mock() + self.api.window_extended_state.return_value = {'window': {'motors': [ + {'id': 0, 'currentPos': {'position': polled}} + ]}} + with patch('custom_components.blebox_smartwindowbox.cover.asyncio.sleep', new=AsyncMock()): + await self.cover.async_set_cover_position(position=target) + self.assertEqual( + [call.args for call in self.api.send_motor_command.await_args_list], + [(0, command), (0, 's'), (0, 's')], + ) + + async def test_matching_intermediate_position_does_not_move(self): + await self.cover.async_set_cover_position(position=60) + self.api.send_motor_command.assert_not_awaited() + + async def test_invalid_target_does_not_move(self): + for target in (None, 'invalid'): + await self.cover.async_set_cover_position(position=target) + self.api.send_motor_command.assert_not_awaited()