Merge pull request #43 from mdeweerd/dev

Add Tuya magic spell

README.md

@@ -72,6 +72,9 @@
     - [`zha_devices`: Device List Information to Event or CSV](#zha_devices-device-list-information-to-event-or-csv)
     - [`register_services`: Reregister ZHA-Toolkit services](#register_services-reregister-zha-toolkit-services)
     - [User method](#user-method)
+  - [Manufacturers](#manufacturers)
+    - [Tuya](#tuya)
+      - [`tuya_magic` - Tuya Magic spell](#tuya_magic---tuya-magic-spell)
 - [Credits/Motivation](#creditsmotivation)
 - [License](#license)
 - [Contributing](#contributing)
@@ -1391,6 +1394,33 @@ This is a powerful tool to develop your own custom tool, and propose it for
 inclusion in the zha-toolkit when it's ready and of potential use to
 others.
 
+## Manufacturers
+
+### Tuya
+
+Shame on Tuya to be a
+[member of the Zigbee Alliance](https://zigbeealliance.org/member/tuya-global-inc/)
+and deviate from the Zigbee Specifications.
+
+#### `tuya_magic` - Tuya Magic spell
+
+This was labeled the
+[standard tuya "magic spell"](https://github.com/Koenkk/zigbee2mqtt/issues/9564#issuecomment-1051123722)
+as it makes most Tuya devices work normally.
+
+currently only the "read" part is implemented - if needed a
+super_magic_spell can be added to also execute the write procedure.
+
+It has to be done using the "execute" command it's not implemented as a
+searchable service.
+
+```yaml
+service: zha_toolkit.execute
+data:
+  command: tuya_magic
+  ieee: light.tz3000_dbou1ap4_ts0505a_level_light_color_on_off
+```
+
 # Credits/Motivation
 
 This project was forked from

STATS.md

@@ -1,6 +1,9 @@
 # Badges showing number of downloads per version
 
 - ![badge latest](https://img.shields.io/github/downloads/mdeweerd/zha-toolkit/latest/total.svg)
+- ![badge v0.7.25](https://img.shields.io/github/downloads/mdeweerd/zha-toolkit/v0.7.25/total.svg)
+- ![badge v0.7.23](https://img.shields.io/github/downloads/mdeweerd/zha-toolkit/v0.7.23/total.svg)
+- ![badge v0.7.24](https://img.shields.io/github/downloads/mdeweerd/zha-toolkit/v0.7.24/total.svg)
 - ![badge v0.7.22](https://img.shields.io/github/downloads/mdeweerd/zha-toolkit/v0.7.22/total.svg)
 - ![badge v0.7.21](https://img.shields.io/github/downloads/mdeweerd/zha-toolkit/v0.7.21/total.svg)
 - ![badge v0.7.20](https://img.shields.io/github/downloads/mdeweerd/zha-toolkit/v0.7.20/total.svg)

custom_components/zha_toolkit/manifest.json

@@ -9,7 +9,7 @@
     "domain": "zha_toolkit",
     "iot_class": "local_polling",
     "issue_tracker": "https://github.com/mdeweerd/zha-toolkit/issues",
-    "name": "🧰 ZHA Toolkit Service",
+    "name": "\ud83e\uddf0 ZHA Toolkit Service",
     "requirements": [],
     "version": "0.0.0"
 }

custom_components/zha_toolkit/services.yaml

@@ -50,6 +50,7 @@ execute:
             - scan_device
             - unbind_coordinator
             - unbind_group
+            - tuya_magic
             - zcl_cmd
             - zdo_flood_parent_annce
             - zdo_join_with_code
@@ -1521,6 +1522,52 @@ unbind_coordinator:
       description: Wait for/expect a reply (not used yet)
       selector:
         boolean:
+tuya_magic:
+  description: >-
+    Do Tuya magic spell (= make most Tuya devices work normally)
+  fields:
+    ieee:
+      description: >-
+        Entity name,\ndevice name, or\nIEEE address of the node to execute
+        command
+      example: 00:0d:6f:00:05:7d:2d:34
+      required: true
+      selector:
+        entity:
+          integration: zha
+    tries:
+      description: Number of times the zigbee packet should be attempted
+      selector:
+        number:
+          min: 0
+          max: 255
+          mode: box
+    event_success:
+      description: Event name in case of success
+      example: my_read_success_trigger_event
+      selector:
+        text:
+    event_fail:
+      description: Event name in case of failure
+      example: my_read_fail_trigger_event
+      selector:
+        text:
+    event_done:
+      description: >-
+        Event name when the service call did all its work (either success
+        or failure).  Has event data with relevant attributes.
+      example: my_read_done_trigger_event
+      selector:
+        text:
+    fail_exception:
+      description: >-
+        Throw exception when success==False, useful to stop scripts, automations
+      selector:
+        boolean:
+    expect_reply:
+      description: Wait for/expect a reply (not used yet)
+      selector:
+        boolean:
 zcl_cmd:
   description: Send cluster command
   fields:

custom_components/zha_toolkit/tuya.py

@@ -0,0 +1,26 @@
+from custom_components.zha_toolkit import utils as u
+from custom_components.zha_toolkit.params import INTERNAL_PARAMS as p
+
+
+async def tuya_magic(
+    app, listener, ieee, cmd, data, service, params, event_data
+):
+
+    dev = app.get_device(ieee)
+    """Initialize device so that all endpoints become available."""
+    basic_cluster = dev.endpoints[1].in_clusters[0]
+
+    # The magic spell is needed only once.
+    # TODO: Improve by doing this only once (successfully).
+
+    # Magic spell - part 1
+    attr_to_read = [4, 0, 1, 5, 7, 0xFFFE]
+    res = await u.cluster_read_attributes(
+        basic_cluster, attr_to_read, tries=params[p.TRIES]
+    )
+
+    event_data["result"] = res
+
+    # Magic spell - part 2 (skipped - does not seem to be needed)
+    # attr_to_write={0xffde:13}
+    # basic_cluster.write_attributes(attr_to_write, tries=3)

custom_components/zha_toolkit/utils.py

@@ -1,12 +1,15 @@
 from __future__ import annotations
 
+import asyncio
 import json
 import logging
 import os
 from enum import Enum
 
 from homeassistant.util.json import save_json
 from zigpy import types as t
+from zigpy.exceptions import DeliveryError
+from zigpy.util import retryable
 from zigpy.zcl import foundation as f
 
 from .params import INTERNAL_PARAMS as p
@@ -687,3 +690,27 @@ def extractParams(  # noqa: C901
         params[p.CSV_LABEL] = rawParams[P.CSVLABEL]
 
     return params
+
+
+# zigpy wrappers
+
+# The zigpy library does not offer retryable on read_attributes.
+# Add it ourselves
+@retryable(
+    (DeliveryError, asyncio.CancelledError, asyncio.TimeoutError), tries=1
+)
+async def cluster_read_attributes(
+    cluster, attrs, manufacturer=None
+) -> tuple[list, list]:
+    """Read attributes from cluster, retryable"""
+    return await cluster.read_attributes(attrs, manufacturer=manufacturer)
+
+
+# The zigpy library does not offer retryable on read_attributes.
+# Add it ourselves
+@retryable(
+    (DeliveryError, asyncio.CancelledError, asyncio.TimeoutError), tries=1
+)
+async def cluster__write_attributes(cluster, attrs, manufacturer=None):
+    """Write cluster attributes from cluster, retryable"""
+    return await cluster._write_attributes(attrs, manufacturer=manufacturer)

custom_components/zha_toolkit/zcl_attr.py

@@ -7,7 +7,6 @@
 from homeassistant.util import dt as dt_util
 from zigpy import types as t
 from zigpy.exceptions import DeliveryError
-from zigpy.util import retryable
 from zigpy.zcl import Cluster
 from zigpy.zcl import foundation as f
 
@@ -239,28 +238,6 @@ async def conf_report(
             )
 
 
-# The zigpy library does not offer retryable on read_attributes.
-# Add it ourselves
-@retryable(
-    (DeliveryError, asyncio.CancelledError, asyncio.TimeoutError), tries=1
-)
-async def cluster_read_attributes(
-    cluster, attrs, manufacturer=None
-) -> tuple[list, list]:
-    """Read attributes from cluster, retryable"""
-    return await cluster.read_attributes(attrs, manufacturer=manufacturer)
-
-
-# The zigpy library does not offer retryable on read_attributes.
-# Add it ourselves
-@retryable(
-    (DeliveryError, asyncio.CancelledError, asyncio.TimeoutError), tries=1
-)
-async def cluster__write_attributes(cluster, attrs, manufacturer=None):
-    """Write cluster attributes from cluster, retryable"""
-    return await cluster._write_attributes(attrs, manufacturer=manufacturer)
-
-
 async def attr_read(*args, **kwargs):
     # Delegate to attr_write which also handles the read command.
     await attr_write(*args, **kwargs)
@@ -304,7 +281,7 @@ async def attr_write(  # noqa: C901
         or (cmd != S.ATTR_WRITE)
     ):
         LOGGER.debug("Request attr read %s", attr_read_list)
-        result_read = await cluster_read_attributes(
+        result_read = await u.cluster_read_attributes(
             cluster,
             attr_read_list,
             manufacturer=params[p.MANF],
@@ -384,7 +361,7 @@ async def attr_write(  # noqa: C901
             result_read = None
 
         LOGGER.debug("Request attr write %s", attr_write_list)
-        result_write = await cluster__write_attributes(
+        result_write = await u.cluster__write_attributes(
             cluster,
             attr_write_list,
             manufacturer=params[p.MANF],
@@ -405,7 +382,7 @@ async def attr_write(  # noqa: C901
 
         if params[p.READ_AFTER_WRITE]:
             LOGGER.debug(f"Request attr read {attr_read_list!r}")
-            result_read = await cluster_read_attributes(
+            result_read = await u.cluster_read_attributes(
                 cluster,
                 attr_read_list,
                 manufacturer=params[p.MANF],