hastarin
commented
1 year ago I've made a start on the custom quirk but it's still not working the way I'd like so requires more work when time permits.
Open hastarin opened 1 year ago
hastarin
commented
1 year ago I've made a start on the custom quirk but it's still not working the way I'd like so requires more work when time permits.
hastarin
commented
1 year ago If anyone can tell me how to get the log output showing via Home Assistant it would be most appreciated. I thought just enabling debug logging for zhaquirks would do the trick but it's still not showing up no matter what I try putting in for the name of the logger.
hastarin
commented
1 year ago Just adding some TS code for reference from https://github.com/Koenkk/zigbee-herdsman-converters/blob/master/src/devices/xiaomi.ts in case it helps me figure things out:
{
zigbeeModel: ['lumi.curtain'],
model: 'ZNCLDJ11LM',
description: 'Aqara curtain motor',
vendor: 'Xiaomi',
fromZigbee: [fz.xiaomi_basic, fz.xiaomi_curtain_position, fz.xiaomi_curtain_position_tilt],
toZigbee: [tz.xiaomi_curtain_position_state, tz.xiaomi_curtain_options],
onEvent: async (type, data, device) => {
if (type === 'message' && data.type === 'attributeReport' && data.cluster === 'genBasic' &&
data.data.hasOwnProperty('1028') && data.data['1028'] == 0) {
// Try to read the position after the motor stops, the device occasionally report wrong data right after stopping
// Might need to add delay, seems to be working without one but needs more tests.
await device.getEndpoint(1).read('genAnalogOutput', ['presentValue']);
}
},
exposes: [e.cover_position().setAccess('state', ea.ALL),
e.binary('running', ea.STATE, true, false)
.withDescription('Whether the motor is moving or not'),
e.enum('motor_state', ea.STATE, ['stopped', 'opening', 'closing'])
.withDescription('Motor state')],
ota: ota.zigbeeOTA,
},Original comment has been updated with a working custom quirk. I'll need to clean it up at some point and get a pull request sorted.
mgwrd
commented
1 year ago Thank you very much for this quirk! Could it be, that you are refering to the model ZNCLDJ12LM?
I had to tweak the signature for my Aqara device (ZNCLDJ11LM) accordingly (see commented values in INPUT_CLUSTERS and OUTPUT_CLUSTERS):
signature = {
MODELS_INFO: [(LUMI, "lumi.curtain")],
ENDPOINTS: {
# <SizePrefixedSimpleDescriptor endpoint=1 profile=260 device_type=514
# device_version=1
# input_clusters=["0x0000", "0x0001", "0x0003", "0x0004", "0x0005", "0x0006", "0x000a", "0x000d", "0x0013", "0x0102", "0x0406"]
# output_clusters=["0x0001", "0x0006", "0x000a", "0x000d", "0x0013", "0x0019", "0x0102", "0x0406"]>
1: {
PROFILE_ID: zha.PROFILE_ID,
DEVICE_TYPE: zha.DeviceType.WINDOW_COVERING_DEVICE,
INPUT_CLUSTERS: [
Basic.cluster_id,
# PowerConfiguration.cluster_id,
Identify.cluster_id,
Groups.cluster_id,
Scenes.cluster_id,
OnOff.cluster_id,
Time.cluster_id,
AnalogOutput.cluster_id,
MultistateOutput.cluster_id,
WindowCovering.cluster_id,
# OccupancySensing.cluster_id,
],
OUTPUT_CLUSTERS: [
# PowerConfiguration.cluster_id,
OnOff.cluster_id,
Time.cluster_id,
AnalogOutput.cluster_id,
MultistateOutput.cluster_id,
Ota.cluster_id,
WindowCovering.cluster_id,
# OccupancySensing.cluster_id,
],
},
},
}
replacement = {
ENDPOINTS: {
1: {
PROFILE_ID: zha.PROFILE_ID,
DEVICE_TYPE: zha.DeviceType.WINDOW_COVERING_DEVICE,
INPUT_CLUSTERS: [
Basic.cluster_id,
# PowerConfiguration.cluster_id,
OnOffB1,
AnalogOutputB1,
WindowCoveringB1,
],
OUTPUT_CLUSTERS: [
OnOffB1,
Time.cluster_id,
AnalogOutputB1,
Ota.cluster_id,
WindowCoveringB1,
],
},
},
}No I'm guessing maybe you're got a newer variant?
mgwrd
commented
1 year ago Yes, you're right. Mine seems to be manufacured in 2022 (HW-Versoin 17, app/ota-file version 36):
Programmierus
commented
1 year ago Can confirm this quirk working with my device, thank you!
poldim
commented
7 months ago I just migrated my curtains from the aqara gateway to ZHA and unfortunately ZHA is not correctly supporting these curtains. I'm going to try this quirk.
Is this difficult to PR into core HA?
gordio
commented
2 months ago My version required some deletions about PowerConfiguration.cluster_id and OccupancySensing.cluster_id.
In Manage Zigbee device
hw_version: 17
and final file looks like:
"""Aqara Curtain Driver B1 device."""
from __future__ import annotations
import logging
from typing import Any
from zigpy import types as t
from zigpy.profiles import zha
from zigpy.zcl import foundation
from zigpy.zcl.clusters.closures import WindowCovering
from zigpy.zcl.clusters.general import (
AnalogOutput,
Basic,
Groups,
Identify,
MultistateOutput,
OnOff,
Ota,
#PowerConfiguration,
Scenes,
Time,
)
from zigpy.zcl.clusters.measurement import OccupancySensing
from zhaquirks import CustomCluster
from zhaquirks.const import (
DEVICE_TYPE,
ENDPOINTS,
INPUT_CLUSTERS,
MODELS_INFO,
OUTPUT_CLUSTERS,
PROFILE_ID,
)
from zhaquirks.xiaomi import LUMI, XiaomiCustomDevice
_LOGGER = logging.getLogger(__name__)
class AnalogOutputB1(CustomCluster, AnalogOutput):
"""Analog output cluster, used to relay current_value to WindowCovering."""
def __init__(self, *args: Any, **kwargs: Any) -> None:
"""Init."""
_LOGGER.debug("AnalogOutputB1 Cluster init")
super().__init__(*args, **kwargs)
self._update_attribute(
self.attributes_by_name["max_present_value"].id, 100.0
) # max_present_value
self._update_attribute(
self.attributes_by_name["min_present_value"].id, 0.0
) # min_present_value
def _update_attribute(self, attrid: int, value: Any) -> None:
_LOGGER.debug("AnalogOutput update attribute %04x to %s... ", attrid, value)
super()._update_attribute(attrid, value)
if attrid == self.attributes_by_name["present_value"].id:
self.endpoint.window_covering._update_attribute( # pylint: disable=protected-access
WindowCovering.attributes_by_name[
"current_position_lift_percentage"
].id,
(100 - value),
)
self.endpoint.on_off._update_attribute( # pylint: disable=protected-access
OnOff.attributes_by_name["on_off"].id, value > 0
)
class WindowCoveringB1(CustomCluster, WindowCovering):
"""Window covering cluster, used to cause commands to update the AnalogOutput present_value."""
stop_called = bool(False)
def __init__(self, *args: Any, **kwargs: Any) -> None:
"""Init."""
_LOGGER.debug("WindowCoveringB1 Cluster init")
super().__init__(*args, **kwargs)
def _update_attribute(self, attrid: int, value: Any) -> None:
_LOGGER.debug(
"WindowCovering update attribute %04x to %s and stop_called is %s",
attrid,
value,
self.stop_called,
)
if self.stop_called:
_LOGGER.debug("Recently stopped, updating AnalogOutput value")
self.stop_called = bool(False)
self.endpoint.analog_output._update_attribute( # pylint: disable=protected-access
AnalogOutput.attributes_by_name["present_value"].id, value
)
super()._update_attribute(attrid, value)
async def command(
self,
command_id: foundation.GeneralCommand | int | t.uint8_t,
*args: Any,
manufacturer: int | t.uint16_t | None = None,
expect_reply: bool = True,
tsn: int | t.uint8_t | None = None,
**kwargs: Any,
) -> Any:
"""Overwrite the commands.
Overwrite analog_output's current_value
value to make the curtain work as expected.
"""
_LOGGER.debug("WindowCovering command %04x", command_id)
if command_id == self.commands_by_name["up_open"].id:
(res,) = await self.endpoint.analog_output.write_attributes(
{"present_value": 100}
)
return foundation.GENERAL_COMMANDS[
foundation.GeneralCommand.Default_Response
].schema(command_id=command_id, status=res[0].status)
if command_id == self.commands_by_name["down_close"].id:
(res,) = await self.endpoint.analog_output.write_attributes(
{"present_value": 0}
)
return foundation.GENERAL_COMMANDS[
foundation.GeneralCommand.Default_Response
].schema(command_id=command_id, status=res[0].status)
if command_id == self.commands_by_name["go_to_lift_percentage"].id:
_LOGGER.debug("go_to_lift_percentage %d", args[0])
(res,) = await self.endpoint.analog_output.write_attributes(
{"present_value": (100 - args[0])}
)
return foundation.GENERAL_COMMANDS[
foundation.GeneralCommand.Default_Response
].schema(command_id=command_id, status=res[0].status)
if command_id == self.commands_by_name["stop"].id:
self.stop_called = bool(True)
return await super().command(
command_id,
*args,
manufacturer=manufacturer,
expect_reply=expect_reply,
tsn=tsn,
**kwargs,
)
class OnOffB1(CustomCluster, OnOff):
"""On Off Cluster, used to update state based on AnalogOutput"""
def __init__(self, *args: Any, **kwargs: Any) -> None:
"""Init."""
_LOGGER.debug("OnOffB1 Cluster init")
super().__init__(*args, **kwargs)
def _update_attribute(self, attrid: int, value: Any) -> None:
_LOGGER.debug("OnOff update attribute %04x to %s", attrid, value)
super()._update_attribute(attrid, value)
class CurtainB1(XiaomiCustomDevice):
"""Aqara Curtain Driver B1 device."""
def __init__(self, *args: Any, **kwargs: Any) -> None:
"""Init."""
super().__init__(*args, **kwargs) # type: ignore
_LOGGER.info("CurtainB1 custom quirk loaded for model ZNCLDJ11LM")
signature = {
MODELS_INFO: [(LUMI, "lumi.curtain")],
ENDPOINTS: {
# <SizePrefixedSimpleDescriptor endpoint=1 profile=260 device_type=514
# device_version=1
# input_clusters=["0x0000", #"0x0001", "0x0003", "0x0004", "0x0005", "0x0006", "0x000a", "0x000d", "0x0013", "0x0102", #"0x0406"]
# output_clusters=["0x0001", "0x0006", "0x000a", "0x000d", "0x0013", "0x0019", "0x0102", "0x0406"]>
1: {
PROFILE_ID: zha.PROFILE_ID,
DEVICE_TYPE: zha.DeviceType.WINDOW_COVERING_DEVICE,
INPUT_CLUSTERS: [
Basic.cluster_id,
#PowerConfiguration.cluster_id,
Identify.cluster_id,
Groups.cluster_id,
Scenes.cluster_id,
OnOff.cluster_id,
Time.cluster_id,
AnalogOutput.cluster_id,
MultistateOutput.cluster_id,
WindowCovering.cluster_id,
#OccupancySensing.cluster_id,
],
OUTPUT_CLUSTERS: [
#PowerConfiguration.cluster_id,
OnOff.cluster_id,
Time.cluster_id,
AnalogOutput.cluster_id,
MultistateOutput.cluster_id,
Ota.cluster_id,
WindowCovering.cluster_id,
#OccupancySensing.cluster_id,
],
},
},
}
replacement = {
ENDPOINTS: {
1: {
PROFILE_ID: zha.PROFILE_ID,
DEVICE_TYPE: zha.DeviceType.WINDOW_COVERING_DEVICE,
INPUT_CLUSTERS: [
Basic.cluster_id,
#PowerConfiguration.cluster_id,
OnOffB1,
AnalogOutputB1,
WindowCoveringB1,
],
OUTPUT_CLUSTERS: [
OnOffB1,
Time.cluster_id,
AnalogOutputB1,
Ota.cluster_id,
WindowCoveringB1,
],
},
},
}AND I changed one parameter in Manage Zigbee device:
max_present_value (id: 0x0041) in Attributes of the selected cluster;poldim
commented
1 month ago AND I changed one parameter in Manage Zigbee device:
- Clusters tab;
- Select AnalogOutputB1 (Endpoint id: 1, Id: 0x000d, Type: in) in Clusters dropdown;
- Select
max_present_value (id: 0x0041)in Attributes of the selected cluster;- Press READ ATTRIBUTE;
- If Value loaded as 0 -> manually change to 100 and press WRITE ATTRIBUTE.
What did doing this fix/solve?
poldim
commented
1 month ago Original comment has been updated with a working custom quirk. I'll need to clean it up at some point and get a pull request sorted.
@hastarin did you do any more cleanup / change to this quirk?
PowerConfiguration & OccupancySensing as others have. Wondering if I should try commenting these out...?My device signature:
{
"node_descriptor": {
"logical_type": 1,
"complex_descriptor_available": 0,
"user_descriptor_available": 0,
"reserved": 0,
"aps_flags": 0,
"frequency_band": 8,
"mac_capability_flags": 142,
"manufacturer_code": 4447,
"maximum_buffer_size": 127,
"maximum_incoming_transfer_size": 100,
"server_mask": 0,
"maximum_outgoing_transfer_size": 100,
"descriptor_capability_field": 0
},
"endpoints": {
"1": {
"profile_id": "0x0104",
"device_type": "0x0202",
"input_clusters": [
"0x0000",
"0x0001",
"0x0006",
"0x000d",
"0x0102"
],
"output_clusters": [
"0x0006",
"0x000a",
"0x000d",
"0x0019",
"0x0102"
]
}
},
"manufacturer": "LUMI",
"model": "lumi.curtain",
"class": "aqara-curtain.CurtainB1"
}Photo:
Device in HA:
hastarin
commented
1 month ago No I haven't as it's been working for me and I barely understood things enough to do that at the time so I'm sticking with the old, if it ain't broke don't fix it.
ZHA used to show a switch which vanished with an update sometime ago now but I just changed anything that was using it and moved on.
You can certainly try commenting those things out it obviously doesn't have an occupancy sensor but I have no idea what power configuration might do.
Seeing that inverted option in your screenshot does make me wonder if perhaps it might work without the quirk and just setting that instead in the current ZHA but again if it ain't broke...
poldim
commented
1 month ago Good stuff...
Also, I'm not sure what identify does, I don't see it doing anything noticeable
Problem description
I'm using ZHA to try controlling this curtain motor and it seems to have a few issues.
Solution description
Either correcting the open/closed state reporting and/or add a configuration option to invert it.
Hide the Occupancy/Opening sensors by default.
Screenshots/Video
Screenshots/Video
[Paste/upload your media here]Device signature
Device signature
```json { "node_descriptor": "NodeDescriptor(logical_type=Diagnostic information
Diagnostic information
I get an error trying to download this.Logs
Logs
``` 2023-07-02 17:25:08.676 DEBUG (MainThread) [zigpy_deconz.uart] Frame received: 0x0e5c000700aa00 2023-07-02 17:25:08.676 DEBUG (MainThread) [zigpy_deconz.api] Received command device_state_changed[Custom quirk
Custom quirk
```python """Aqara Curtain Driver B1 device.""" from __future__ import annotations from typing import Any import logging from zigpy import types as t from zigpy.profiles import zha from zigpy.zcl import foundation from zigpy.zcl.clusters.closures import WindowCovering from zigpy.zcl.clusters.general import ( AnalogOutput, Basic, Groups, Identify, MultistateOutput, OnOff, Ota, PowerConfiguration, Scenes, Time, ) from zigpy.zcl.clusters.measurement import OccupancySensing from zhaquirks import CustomCluster from zhaquirks.const import ( DEVICE_TYPE, ENDPOINTS, INPUT_CLUSTERS, MODELS_INFO, OUTPUT_CLUSTERS, PROFILE_ID, ) from zhaquirks.xiaomi import LUMI, XiaomiCustomDevice _LOGGER = logging.getLogger(__name__) class AnalogOutputB1(CustomCluster, AnalogOutput): """Analog output cluster, used to relay current_value to WindowCovering.""" def __init__(self, *args: Any, **kwargs: Any) -> None: """Init.""" _LOGGER.debug("AnalogOutputB1 Cluster init") super().__init__(*args, **kwargs) self._update_attribute( self.attributes_by_name["max_present_value"].id, 100.0 ) # max_present_value self._update_attribute( self.attributes_by_name["min_present_value"].id, 0.0 ) # min_present_value def _update_attribute(self, attrid: int, value: Any) -> None: _LOGGER.debug("AnalogOutput update attribute %04x to %s... ", attrid, value) super()._update_attribute(attrid, value) if attrid == self.attributes_by_name["present_value"].id: self.endpoint.window_covering._update_attribute( # pylint: disable=protected-access WindowCovering.attributes_by_name[ "current_position_lift_percentage" ].id, (100 - value), ) self.endpoint.on_off._update_attribute( # pylint: disable=protected-access OnOff.attributes_by_name["on_off"].id, value > 0 ) class WindowCoveringB1(CustomCluster, WindowCovering): """Window covering cluster, used to cause commands to update the AnalogOutput present_value.""" stop_called = bool(False) def __init__(self, *args: Any, **kwargs: Any) -> None: """Init.""" _LOGGER.debug("WindowCoveringB1 Cluster init") super().__init__(*args, **kwargs) def _update_attribute(self, attrid: int, value: Any) -> None: _LOGGER.debug( "WindowCovering update attribute %04x to %s and stop_called is %s", attrid, value, self.stop_called, ) if self.stop_called: _LOGGER.debug("Recently stopped, updating AnalogOutput value") self.stop_called = bool(False) self.endpoint.analog_output._update_attribute( # pylint: disable=protected-access AnalogOutput.attributes_by_name["present_value"].id, value ) super()._update_attribute(attrid, value) async def command( self, command_id: foundation.GeneralCommand | int | t.uint8_t, *args: Any, manufacturer: int | t.uint16_t | None = None, expect_reply: bool = True, tsn: int | t.uint8_t | None = None, **kwargs: Any, ) -> Any: """Overwrite the commands. Overwrite analog_output's current_value value to make the curtain work as expected. """ _LOGGER.debug("WindowCovering command %04x", command_id) if command_id == self.commands_by_name["up_open"].id: (res,) = await self.endpoint.analog_output.write_attributes( {"present_value": 100} ) return foundation.GENERAL_COMMANDS[ foundation.GeneralCommand.Default_Response ].schema(command_id=command_id, status=res[0].status) if command_id == self.commands_by_name["down_close"].id: (res,) = await self.endpoint.analog_output.write_attributes( {"present_value": 0} ) return foundation.GENERAL_COMMANDS[ foundation.GeneralCommand.Default_Response ].schema(command_id=command_id, status=res[0].status) if command_id == self.commands_by_name["go_to_lift_percentage"].id: _LOGGER.debug("go_to_lift_percentage %d", args[0]) (res,) = await self.endpoint.analog_output.write_attributes( {"present_value": (100 - args[0])} ) return foundation.GENERAL_COMMANDS[ foundation.GeneralCommand.Default_Response ].schema(command_id=command_id, status=res[0].status) if command_id == self.commands_by_name["stop"].id: self.stop_called = bool(True) return await super().command( command_id, *args, manufacturer=manufacturer, expect_reply=expect_reply, tsn=tsn, **kwargs, ) class OnOffB1(CustomCluster, OnOff): """On Off Cluster, used to update state based on AnalogOutput""" def __init__(self, *args: Any, **kwargs: Any) -> None: """Init.""" _LOGGER.debug("OnOffB1 Cluster init") super().__init__(*args, **kwargs) def _update_attribute(self, attrid: int, value: Any) -> None: _LOGGER.debug("OnOff update attribute %04x to %s", attrid, value) super()._update_attribute(attrid, value) class CurtainB1(XiaomiCustomDevice): """Aqara Curtain Driver B1 device.""" def __init__(self, *args: Any, **kwargs: Any) -> None: """Init.""" super().__init__(*args, **kwargs) # type: ignore _LOGGER.info("CurtainB1 custom quirk loaded") signature = { MODELS_INFO: [(LUMI, "lumi.curtain")], ENDPOINTS: { #Additional information
Now with a working custom quirk attached.