misc_settime: write read values to csv file

custom_components/zha_toolkit/misc.py

@@ -1,4 +1,3 @@
-# import asyncio
 import asyncio
 import logging
 
@@ -280,9 +279,12 @@ async def misc_settime(
         ]  # Time, Timestatus, Timezone, DstStart, DstEnd, DstShift
 
         if params[p.READ_BEFORE_WRITE]:
-            event_data["read_before"] = await cluster.read_attributes(
-                attr_read_list
+            read_resp = await cluster.read_attributes(attr_read_list)
+            event_data["read_before"] = (
+                u.dict_to_jsonable(read_resp[0]),
+                read_resp[1],
             )
+            u.record_read_data(read_resp, cluster, params, listener)
 
         EPOCH2000_TIMESTAMP = 946684800
         utctime_towrite = utcnow().timestamp() - EPOCH2000_TIMESTAMP
@@ -299,6 +301,9 @@ async def misc_settime(
         )
 
         if params[p.READ_AFTER_WRITE]:
-            event_data["read_after"] = await cluster.read_attributes(
-                attr_read_list
+            read_resp = await cluster.read_attributes(attr_read_list)
+            event_data["read_after"] = (
+                u.dict_to_jsonable(read_resp[0]),
+                read_resp[1],
             )
+            u.record_read_data(read_resp, cluster, params, listener)

custom_components/zha_toolkit/services.yaml

@@ -1884,6 +1884,13 @@ misc_settime:
         match.  Defaults to True
       selector:
         boolean:
+    csvout:
+      name: CSV Filename
+      description: >-
+        Filename of CSV to write read data to.  Written to 'csv' directory
+      example: ../web/mycsv.csv
+      selector:
+        text:
 ota_notify:
   name: Trigger Device's Firmware Update
   description:

custom_components/zha_toolkit/translations/en.json

@@ -1073,6 +1073,10 @@
         "read_after_write": {
           "name": "Read After Write",
           "description": "Read attribute after writing.  Can be used to ensure the values match.  Defaults to True"
+        },
+        "csvout": {
+          "name": "CSV Filename",
+          "description": "Filename of CSV to write read data to.  Written to 'csv' directory"
         }
       }
     },

custom_components/zha_toolkit/utils.py

@@ -9,7 +9,9 @@
 import typing
 from enum import Enum
 
+import zigpy
 from homeassistant.components.zha.core.gateway import ZHAGateway
+from homeassistant.util import dt as dt_util
 from pkg_resources import get_distribution, parse_version
 from zigpy import types as t
 from zigpy.exceptions import ControllerException, DeliveryError
@@ -531,6 +533,58 @@ def append_to_csvfile(
         LOGGER.debug(f"Appended {desc} to '{file_name}'")
 
 
+def record_read_data(
+    read_resp, cluster: zigpy.zcl.Cluster, params, listener=None
+):
+    """Record result from attr_write to CSV file if configured"""
+    if params[p.CSV_FILE] is None:
+        return
+
+    date_str = dt_util.utcnow().isoformat()
+
+    for attr_id, read_val in read_resp[0].items():
+        fields = []
+        if params[p.CSV_LABEL] is not None:
+            attr_name = params[p.CSV_LABEL]
+        else:
+            python_type = type(read_resp[0][attr_id])
+            attr_type = f.DATA_TYPES.pytype_to_datatype_id(python_type)
+
+            try:
+                attr_def = cluster.attributes.get(
+                    attr_id, (str(attr_id), None)
+                )
+                if is_zigpy_ge("0.50.0") and isinstance(
+                    attr_def, f.ZCLAttributeDef
+                ):
+                    attr_name = attr_def.name
+                else:
+                    attr_name = attr_def[0]
+            except Exception:
+                attr_name = attr_id
+
+        fields.append(date_str)
+        fields.append(cluster.name)
+        fields.append(attr_name)
+        fields.append(read_val)
+        fields.append(f"0x{attr_id:04X}")
+        fields.append(f"0x{cluster.cluster_id:04X}")
+        fields.append(cluster.endpoint.endpoint_id)
+        fields.append(str(cluster.endpoint.device.ieee))
+        fields.append(
+            f"0x{params[p.MANF]:04X}" if params[p.MANF] is not None else ""
+        )
+        fields.append(f"0x{attr_type:02X}" if attr_type is not None else "")
+
+        append_to_csvfile(
+            fields,
+            "csv",
+            params[p.CSV_FILE],
+            f"{attr_name}={read_val}",
+            listener=listener,
+        )
+
+
 def get_attr_id(cluster, attribute):
     # Try to get attribute id from cluster
     try: