Model command arguments as struct parameters

majordom_zigbee/controller.py

@@ -24,7 +24,7 @@
 from zigpy.types import EUI64
 from zigpy.zcl.clusters.general import Identify
 from zigpy.zcl.foundation import Status as ZCLStatus
-from zigpy.zcl.foundation import ZCLAttributeAccess, ZCLAttributeDef
+from zigpy.zcl.foundation import ZCLAttributeAccess, ZCLAttributeDef, ZCLCommandDef
 
 from majordom_zigbee._serial import port_holder
 
@@ -352,7 +352,7 @@ async def stop(self):
     # Hub -> device operations
     # -------------------------------------------------------------------------
 
-    async def start_pairing_window(self, duration_sec: int) -> None:
+    async def start_pairing_window(self, duration_sec: int, credentials: ProvidedCredentials | None = None) -> None:
         if not self._application:
             raise ZBConnectionError("ZigBee application is not started")
         log.debug("[PERMIT-JOIN] opening for %ds", duration_sec)
@@ -442,7 +442,7 @@ async def pair_device(
                             )
 
                         if issubclass(attribute.type, Enum) and data_type != ParameterDataType.bool:
-                            valid_values = {member.name: str(member.value) for member in attribute.type}
+                            valid_values = {member.value: member.name for member in attribute.type}
 
                         parameters.append(
                             ZBParameterState(
@@ -490,7 +490,7 @@ async def pair_device(
                             if hasattr(field.type, "max_value"):
                                 max_value = field.type.max_value
                             if isinstance(field.type, type) and issubclass(field.type, Enum):
-                                valid_values = {member.name: str(member.value) for member in field.type}
+                                valid_values = {member.value: member.name for member in field.type}
 
                             fields.append(
                                 Parameter(
@@ -513,7 +513,7 @@ async def pair_device(
                                     device.id, endpoint.endpoint_id, cluster.cluster_id, command.id
                                 ),
                                 name=command.name,
-                                data_type=ParameterDataType.none,
+                                data_type=ParameterDataType.struct if fields else ParameterDataType.none,
                                 role=ParameterRole.control,
                                 fields=json.loads(json.dumps([f.model_dump(mode="json") for f in fields]))
                                 if fields
@@ -545,7 +545,18 @@ async def pair_device(
                     # relay already derives, so there's nothing to set here.
                     main_parameter.default_value = set(main_spec.cycle)
                 elif main_spec.default_arguments is not None:
-                    main_parameter.integration_data.default_arguments = main_spec.default_arguments
+                    data = main_parameter.integration_data
+                    data.default_arguments = main_spec.default_arguments
+                    # The same arguments keyed by sub-parameter id (the SDK struct value) as a one-value
+                    # button: a `struct` command then passes can_be_main_parameter and the relay sends
+                    # them on a tap. default_arguments stays for devices stored before this.
+                    endpoint = cast(zigpy.endpoint.Endpoint, zbdevice.endpoints[data.endpoint_id])
+                    zbcommand = endpoint.in_clusters[data.cluster_id].server_commands[main_spec.target_id]
+                    names = self._command_field_names(device.id, data.endpoint_id, data.cluster_id, zbcommand)
+                    ids = {name: field_id for field_id, name in names.items()}
+                    main_parameter.default_value = {
+                        ids.get(name, name): value for name, value in main_spec.default_arguments.items()
+                    }
             log.debug(
                 f"[PAIR] mapped schema {_zb_path(device, zbdevice)}\n\t"
                 + "\n\t".join(
@@ -682,6 +693,13 @@ async def send_command(self, command: DeviceCommand, device: ZBDevice, parameter
             # A value-less send (e.g. tapping the main parameter) falls back to the arguments this
             # command was set up with as a main parameter — see integration_data.default_arguments.
             arguments = command.value if command.value is not None else parameter.integration_data.default_arguments
+            if isinstance(arguments, dict):
+                # A struct value is keyed by sub-parameter id (SDK), zigpy takes field names. Name
+                # keys (e.g. default_arguments) pass through as-is.
+                names = self._command_field_names(
+                    device.id, parameter.integration_data.endpoint_id, parameter.integration_data.cluster_id, zbcommand
+                )
+                arguments = {names.get(str(key), str(key)): value for key, value in arguments.items()}
             try:
                 if isinstance(arguments, dict):
                     result = await cluster.command(zbcommand.id, **arguments)
@@ -854,6 +872,16 @@ async def _disconnect_unpaired_discovery(self, discovery_id: UUID, ieee: EUI64):
         self._majordom_discoveries.pop(discovery_id)
         self._awaiting_zb_discoveries.pop(discovery_id)
 
+    def _command_field_names(
+        self, device_id: UUID, endpoint_id: int, cluster_id: int, zbcommand: ZCLCommandDef
+    ) -> dict[str, str]:
+        """Sub-parameter id (as str) -> zigpy field name, for a command parameter's fields."""
+        fields = enumerate(zbcommand.schema.fields)
+        return {
+            str(self._mapper.command_field_uuid(device_id, endpoint_id, cluster_id, zbcommand.id, i)): field.name
+            for i, field in fields
+        }
+
     def _get_device_main_parameter(
         self, device_id: UUID, zbdevice: ZPDevice
     ) -> tuple[UUID | None, MainParameterSpec | None]:

majordom_zigbee/model.py

@@ -21,10 +21,10 @@ class ZBParameterIntegrationData(BaseModel):
     command_id: int | None = None
     type: ZBParameterType
     # Args to send when this command is the device's one-tap main parameter and needs them (e.g. a
-    # brightness level). A command parameter's data_type is `none`, which already satisfies
-    # ParameterState.can_be_main_parameter, so no `default_value` is needed for the flag — this only
-    # carries *what to send*. send_command applies it when a command arrives with no value (i.e. the
-    # user tapped the main parameter). Mirrors Matter's default_arguments.
+    # brightness level), keyed by field name; a Hub-sent `struct` value is keyed by sub-parameter id,
+    # and send_command accepts both. This only carries *what to send*. send_command applies it when a
+    # command arrives with no value (i.e. the user tapped the main parameter). Mirrors Matter's
+    # default_arguments.
     default_arguments: dict | None = None