Implement support for power sockets

custom_components/dingz/api.py

@@ -251,7 +251,7 @@ class OutputConfigLight(TypedDict, total=False):
 class OutputConfig(TypedDict, total=False):
     active: bool
     name: str
-    type: Literal["light"] | Literal["fan"] | str
+    type: Literal["light"] | Literal["fan"] | Literal["power_socket"] | str
     groups: str
     feedback: OutputConfigFeedback
     light: OutputConfigLight

custom_components/dingz/switch.py

@@ -13,8 +13,13 @@
 
 from . import api
 from .const import DOMAIN
-from .helpers import compile_json_path, json_path_lookup
-from .shared import ConfigCoordinator, Shared
+from .helpers import (
+    DelayedCoordinatorRefreshMixin,
+    UserAssignedNameMixin,
+    compile_json_path,
+    json_path_lookup,
+)
+from .shared import ConfigCoordinator, Shared, StateCoordinator
 
 
 async def async_setup_entry(
@@ -36,6 +41,13 @@ async def async_setup_entry(
         ),
     ]
 
+    for index, dingz_output in enumerate(shared.config.data.outputs):
+        if (
+            dingz_output.get("active", False)
+            and dingz_output.get("type") == "power_socket"
+        ):
+            entities.append(PowerSocket(shared, index))
+
     async_add_entities(entities)
 
 
@@ -70,3 +82,54 @@ async def _set(self, value: bool) -> None:
         config[self.entity_description.key] = value
         await self.coordinator.shared.client.update_mqtt_service_config(config)
         await self.coordinator.async_request_refresh()
+
+
+class PowerSocket(
+    CoordinatorEntity[StateCoordinator],
+    SwitchEntity,
+    UserAssignedNameMixin,
+    DelayedCoordinatorRefreshMixin,
+):
+    def __init__(self, shared: Shared, index: int) -> None:
+        super().__init__(shared.state)
+        self.__index = index
+
+        self._attr_unique_id = (
+            f"{self.coordinator.shared.mac_addr}-power_socket-{index}"
+        )
+        self._attr_device_info = shared.device_info
+        self._attr_translation_key = "power_socket"
+
+    @property
+    def dingz_output_config(self) -> api.OutputConfig:
+        try:
+            return self.coordinator.shared.config.data.outputs[self.__index]
+        except LookupError:
+            return api.OutputConfig()
+
+    @property
+    def dingz_dimmer_state(self) -> api.StateDimmer:
+        try:
+            return self.coordinator.data["dimmers"][self.__index]
+        except LookupError:
+            return api.StateDimmer()
+
+    @property
+    def comp_index(self) -> int:
+        return self.__index
+
+    @property
+    def user_given_name(self) -> str | None:
+        return self.dingz_output_config.get("name")
+
+    @property
+    def is_on(self) -> bool | None:
+        return self.dingz_dimmer_state.get("on")
+
+    async def async_turn_on(self, **kwargs: Any) -> None:
+        await self.coordinator.shared.client.set_dimmer(self.__index, "on")
+        await self.delayed_request_refresh()
+
+    async def async_turn_off(self, **kwargs: Any) -> None:
+        await self.coordinator.shared.client.set_dimmer(self.__index, "off")
+        await self.delayed_request_refresh()

custom_components/dingz/translations/en.json

@@ -158,6 +158,9 @@
         "switch": {
             "mqtt_enable": {
                 "name": "MQTT Enable"
+            },
+            "power_socket": {
+                "name": "Power Socket {position}"
             }
         },
         "text": {