Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 23 additions & 6 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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.

---

Expand Down Expand Up @@ -192,8 +194,8 @@ You can:
Position scale:

```
0 = Fully Open
100 = Fully Closed
0 = Fully Closed
100 = Fully Open
```

---
Expand Down Expand Up @@ -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
Expand All @@ -319,4 +336,4 @@ This value has been tuned for reliable real-world accuracy.

# License

MIT License
MIT License
51 changes: 31 additions & 20 deletions custom_components/blebox_smartwindowbox/cover.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down Expand Up @@ -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)
Expand All @@ -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 {}
Expand Down Expand Up @@ -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"),
Expand All @@ -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,
Expand Down Expand Up @@ -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:
Expand All @@ -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:
Expand Down Expand Up @@ -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
return
14 changes: 8 additions & 6 deletions custom_components/blebox_smartwindowbox/manifest.json
Original file line number Diff line number Diff line change
@@ -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"
}
"requirements": [],
"version": "0.0.1"
}
89 changes: 89 additions & 0 deletions tests/test_cover.py
Original file line number Diff line number Diff line change
@@ -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()