Add Sonoff SNZB-06P24 presence sensor - #4907
Conversation
…ocus tests on cluster behavior
…ocus tests on cluster behavior
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## dev #4907 +/- ##
==========================================
- Coverage 93.07% 91.39% -1.69%
==========================================
Files 401 417 +16
Lines 13306 14543 +1237
==========================================
+ Hits 12385 13291 +906
- Misses 921 1252 +331 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
There was a problem hiding this comment.
Pull request overview
Adds a new ZHA quirk implementation for the Sonoff SNZB-06P24 presence sensor, introducing support for the device’s manufacturer-specific 0xFC11 cluster and exposing configuration/learning-related entities in Home Assistant.
Changes:
- Introduces
SonoffSNZB06P24FC11Clusterwith custom command handling, payload parsing, and virtual zone attributes backed by a bitmap. - Registers a v2
QuirkBuilderto expose numbers, a command button, sensors, and per-zone enable switches.
| class SpatialLearningUiState(t.enum8): | ||
| """Spatial learning UI state enum.""" | ||
|
|
||
| IDLE = 0x00 | ||
| START = 0x01 | ||
| COUNTDOWN = 0x02 | ||
| TIMEOUT = 0x03 | ||
|
|
There was a problem hiding this comment.
Enum members in SpatialLearningUiState use ALL_CAPS (IDLE/START/COUNTDOWN/TIMEOUT) while other enums in this repo (including SpatialLearningState just above) consistently use PascalCase. This inconsistency makes the enum harder to use and is likely to leak into entity state strings. Rename these members to PascalCase and update all references accordingly.
| from zigpy.quirks.v2.homeassistant import EntityPlatform, EntityType, UnitOfTime | ||
| import zigpy.types as t | ||
| from zigpy.zcl import foundation | ||
|
|
There was a problem hiding this comment.
OccupancySensing is imported but never used in this module. Please remove the unused import to keep linting clean.
| ) | ||
| self._update_attribute(ATTR_SONOFF_SPATIAL_COUNTDOWN, 0) | ||
|
|
||
| def request( |
There was a problem hiding this comment.
_last_spatial_start_ts_ms is written in multiple places but never read. Either remove it or use it to derive UI state/countdown behavior (otherwise it’s dead state that can confuse future maintenance).
| """Intercept read_attributes to handle virtual zone switches.""" | ||
| virtual_attrs = set() | ||
| real_attributes = [] | ||
|
|
||
| requested_ids = set() | ||
|
|
||
| for a in attributes: | ||
| aid = a | ||
| if isinstance(a, str): | ||
| if a in self.attributes_by_name: | ||
| aid = self.attributes_by_name[a].id | ||
|
|
||
| requested_ids.add(aid) | ||
| if isinstance(aid, int) and 0x1000 <= aid <= 0x1007: | ||
| virtual_attrs.add(aid) | ||
| else: | ||
| real_attributes.append(a) | ||
|
|
||
| if not virtual_attrs: | ||
| return await super().read_attributes( | ||
| attributes, allow_cache, only_cache, manufacturer | ||
| ) | ||
|
|
||
| # Ensure we read the bitmap if we need it for virtual attributes | ||
| if ATTR_SONOFF_ZONE_ENABLE not in requested_ids: | ||
| real_attributes.append(ATTR_SONOFF_ZONE_ENABLE) | ||
|
|
||
| success, failure = await super().read_attributes( | ||
| real_attributes, allow_cache, only_cache, manufacturer | ||
| ) | ||
|
|
||
| if ATTR_SONOFF_ZONE_ENABLE in success: | ||
| bitmap_val = success[ATTR_SONOFF_ZONE_ENABLE] | ||
| for v_id in virtual_attrs: | ||
| idx = v_id - 0x1000 | ||
| if idx == 0: | ||
| val = bool(bitmap_val & 0x03) | ||
| else: | ||
| val = bool(bitmap_val & (1 << idx)) | ||
| success[v_id] = t.Bool(val) | ||
| self._update_attribute(v_id, t.Bool(val)) | ||
|
|
||
| if ATTR_SONOFF_ZONE_ENABLE in failure: | ||
| err = failure[ATTR_SONOFF_ZONE_ENABLE] |
There was a problem hiding this comment.
read_attributes() builds success/failure entries for virtual zone attributes using attribute IDs (success[v_id] = ...). Zigpy generally returns results keyed by the same form the caller requested (name vs id). If a caller requests e.g. "zone_0_enable", this implementation will return the value under 0x1000 instead of the name. Preserve and return results keyed by the original requested attribute identifiers.
| """Intercept write_attributes to handle virtual zone switches.""" | ||
|
|
||
| virtual_updates = {} | ||
| real_attributes = {} |
There was a problem hiding this comment.
write_attributes() overrides the base method but does not accept **kwargs (and its attributes type omits foundation.ZCLAttributeDef). Callers in this codebase often pass extra kwargs (e.g. disable_default_response) and may pass attribute defs; this override can raise TypeError. Update the signature to accept/forward **kwargs and include foundation.ZCLAttributeDef in the accepted key types to match zigpy/repo patterns.
| self._update_attribute(0x1001, t.Bool(state)) | ||
| else: | ||
| self._update_attribute(0x1000 + idx, t.Bool(state)) | ||
|
|
||
| # Perform the actual write | ||
| res = await super().write_attributes(real_attributes, manufacturer) | ||
|
|
||
| # Normalize response records. |
There was a problem hiding this comment.
This method updates the virtual zone attributes in the local cache before performing the bitmap write. If the device write fails, the cache/UI will temporarily show the wrong state (and you don’t roll it back). Update the cache only after confirming a SUCCESS status for the bitmap write, or revert the optimistic updates on failure.
| if res and isinstance(res[0], list): | ||
| records = res[0] | ||
| else: | ||
| records = res | ||
| elif hasattr(res, "status_records"): | ||
| records = res.status_records | ||
| else: | ||
| records = [res] | ||
|
|
||
| # Check if the bitmap write was successful | ||
| bitmap_status = foundation.Status.SUCCESS | ||
|
|
||
| for record in records: | ||
| if isinstance(record, foundation.WriteAttributesStatusRecord): | ||
| if record.attrid == ATTR_SONOFF_ZONE_ENABLE: | ||
| bitmap_status = record.status | ||
| break | ||
|
|
||
| # Generate records for virtual updates | ||
| if virtual_updates: | ||
| for idx in virtual_updates: | ||
| v_attr_id = 0x1000 + idx | ||
| # If bitmap write failed, report failure for virtual attrs too | ||
| records.append( | ||
| foundation.WriteAttributesStatusRecord(bitmap_status, v_attr_id) | ||
| ) | ||
|
|
||
| return [records] | ||
|
|
||
| def _update_attribute(self, attrid, value): | ||
| """Update attribute value in cache and push to listeners.""" | ||
| super()._update_attribute(attrid, value) | ||
|
|
There was a problem hiding this comment.
The response “normalization” here can drop data: if super().write_attributes(...) returns multiple response chunks (list of lists), this code only keeps/returns the first chunk (records = res[0]) and discards the rest. Preserve the original response structure and append virtual status records to the corresponding response list without losing other records.
| mode="slider", | ||
| translation_key="fine_tune_sensitivity", | ||
| fallback_name="Fine-tune Sensitivity", | ||
| ) | ||
| # Spatial Learning Button | ||
| # Cmd 0x04, SubCmd=0 (Start), Sequence=timestamp_ms | ||
| .command_button( | ||
| command_name="start_learning_now", | ||
| cluster_id=SonoffSNZB06P24FC11Cluster.cluster_id, | ||
| command_args=(), |
There was a problem hiding this comment.
fallback_name should be sentence case (per repo conventions). "Fine-tune Sensitivity" should be "Fine-tune sensitivity" to avoid inconsistent entity naming in Home Assistant.
|
|
||
| if ( | ||
| hdr.frame_control.frame_type == foundation.FrameType.CLUSTER_COMMAND | ||
| and hdr.command_id == CMD_SPATIAL_LEARNING | ||
| ): | ||
| self._handle_spatial_learning_payload(bytes(payload)) | ||
| return hdr, [] | ||
|
|
||
| return super().deserialize(hdr.serialize() + payload) | ||
|
|
||
| def _handle_spatial_learning_payload(self, payload: bytes) -> None: | ||
| """Handle incoming spatial learning command payloads.""" | ||
| if not payload: | ||
| return | ||
|
|
||
| cmd_type = payload[0] | ||
|
|
||
| if cmd_type == 0x01 and len(payload) >= 17: | ||
| start_ts = int.from_bytes(payload[1:9], "little") | ||
| end_ts = int.from_bytes(payload[9:17], "little") | ||
| self._last_spatial_start_ts_ms = start_ts | ||
|
|
||
| duration_ms = max(0, end_ts - start_ts) | ||
| duration_sec = int(duration_ms // 1000) | ||
|
|
||
| self._set_spatial_ui_state(SpatialLearningUiState.COUNTDOWN) | ||
| self._start_spatial_countdown(duration_sec) | ||
| self._start_spatial_timeout() | ||
| elif cmd_type == 0x02 and len(payload) >= 11: | ||
| self._stop_spatial_tasks() | ||
| self._set_spatial_countdown(0) | ||
| self._set_spatial_ui_state(SpatialLearningUiState.IDLE) | ||
|
|
||
| def _set_spatial_ui_state(self, state: SpatialLearningUiState) -> None: | ||
| self._update_attribute(ATTR_SONOFF_SPATIAL_UI_STATE, state) | ||
|
|
||
| def _set_spatial_countdown(self, seconds: int) -> None: | ||
| self._update_attribute(ATTR_SONOFF_SPATIAL_COUNTDOWN, max(0, int(seconds))) | ||
|
|
||
| def _stop_spatial_tasks(self) -> None: | ||
| if self._spatial_countdown_task and not self._spatial_countdown_task.done(): | ||
| self._spatial_countdown_task.cancel() | ||
| if self._spatial_timeout_task and not self._spatial_timeout_task.done(): | ||
| self._spatial_timeout_task.cancel() | ||
|
|
||
| def _start_spatial_countdown(self, seconds: int) -> None: | ||
| if self._spatial_countdown_task and not self._spatial_countdown_task.done(): | ||
| self._spatial_countdown_task.cancel() | ||
| self._spatial_countdown_task = asyncio.create_task( | ||
| self._spatial_countdown_worker(seconds) | ||
| ) | ||
|
|
||
| async def _spatial_countdown_worker(self, seconds: int) -> None: | ||
| remaining = max(0, int(seconds)) | ||
| self._set_spatial_countdown(remaining) | ||
| while remaining > 0: | ||
| await asyncio.sleep(1) | ||
| remaining -= 1 | ||
| self._set_spatial_countdown(remaining) | ||
|
|
||
| def _start_spatial_timeout(self) -> None: | ||
| if self._spatial_timeout_task and not self._spatial_timeout_task.done(): | ||
| self._spatial_timeout_task.cancel() | ||
| self._spatial_timeout_task = asyncio.create_task(self._spatial_timeout_worker()) | ||
|
|
||
| async def _spatial_timeout_worker(self) -> None: | ||
| await asyncio.sleep(60) | ||
| if self._spatial_countdown_task and not self._spatial_countdown_task.done(): | ||
| self._spatial_countdown_task.cancel() | ||
| self._set_spatial_countdown(0) | ||
| self._set_spatial_ui_state(SpatialLearningUiState.TIMEOUT) | ||
| await asyncio.sleep(5) | ||
| self._set_spatial_ui_state(SpatialLearningUiState.IDLE) | ||
|
|
||
| async def read_attributes( | ||
| self, attributes, allow_cache=False, only_cache=False, manufacturer=None | ||
| ): |
There was a problem hiding this comment.
This quirk introduces substantial custom logic (custom deserialize() parsing, virtual attributes backed by a bitmap, and custom read/write interception). There should be dedicated unit tests (similar to existing Sonoff quirk tests in tests/test_sonoff.py) covering: virtual zone read/write behavior (including failed writes), and spatial learning payload handling updating the UI state/countdown.
| @@ -0,0 +1,385 @@ | |||
| """Sonoff SNZB-09P - Zigbee alarm sensor.""" | |||
|
|
|||
| from typing import Any | |||
| ALARM_STATE_OFF = 0x00 | ||
| ALARM_STATE_MANUAL = 0x01 | ||
| ALARM_STATE_SCENE = 0x02 |
| SUBCMD_START_MANUAL_ALARM if alarm_switch_state else SUBCMD_STOP_ALARM | ||
| ) | ||
|
|
||
| try: | ||
| await self.command( | ||
| CMD_SOUND_AND_LIGHT_ALARM_SETTINGS, | ||
| command_arg, | ||
| manufacturer=manufacturer or SONOFF_MANUFACTURER_CODE, | ||
| ) | ||
| except Exception: | ||
| records.append( | ||
| foundation.WriteAttributesStatusRecord( | ||
| foundation.Status.FAILURE, |
| Idle = 0x00 | ||
| Learning = 0x01 | ||
| Success = 0x02 | ||
| Failure = 0x03 |
| timestamp_ms = int(time.time() * 1000) | ||
| args = (0, timestamp_ms) | ||
| command_id = CMD_SPATIAL_LEARNING | ||
| manufacturer = 0x1286 |
| self._handle_spatial_learning_payload(bytes(payload)) | ||
| return hdr, [] | ||
|
|
||
| return super().deserialize(hdr.serialize() + payload) |
| # Remove bitmap if not requested (by ID or name) | ||
| if ATTR_SONOFF_ZONE_ENABLE not in requested_ids: | ||
| # Also check if user requested it by name "zone_enable" | ||
| # But we only track IDs in requested_ids. | ||
| # If 0x2016 isn't in requested_ids, we pop it. |
| # Optimistically update virtual attributes in cache | ||
| for idx, state in virtual_updates.items(): | ||
| if idx == 0: | ||
| self._update_attribute(0x1000, t.Bool(state)) | ||
| self._update_attribute(0x1001, t.Bool(state)) | ||
| else: | ||
| self._update_attribute(0x1000 + idx, t.Bool(state)) |
|
Love this presence sensor. When will it be merged, please? @TheJulianJES |
|
Tried the quirk localy and it seems to work just fine. |
|
Lately i was getting a warning on ha logs I think it started after upgrading to 2026.7.1 Tried some code tweaks using some llm added And changed to No warning after this change , not sure its the correct one... |
Migrate the spatial learning commands to BaseCommandDefs and ZCLCommandDef so zigpy derives the correct server and client directions. This removes the incorrect direction warnings without changing command IDs, names, or payload schemas.
Proposed change
Add a quirk for the Sonoff SNZB-06P24 presence sensor.
This quirk adds support for the device's manufacturer-specific FC11 cluster and exposes the following controls and entities in ZHA:
The quirk also implements handling for:
Additional information
This change is intended to add support for the Sonoff SNZB-06P24 in ZHA by exposing device-specific configuration and spatial learning features that are not available through the standard clusters alone.
This is not intended to be a breaking change for existing devices.
Device diagnostics
zha-01KNJTVMGFY0P6H2TCKT692JY6-SONOFF SNZB-06P24-6ced29d613a94bc061cbaaa40250d577.json
Checklist
pre-commitchecks pass / the code has been formatted using Black