Add support for device configuration to config_flow

Author: monty68

.gitignore

@@ -136,3 +136,6 @@ custom_components/uniled/lib/zng/__pycache__
 custom_components/uniled/lib/__pycache__
 custom_components/uniled/__pycache__
 custom_components/zengge_mesh
+custom_components/uniled/significant_change.py
+custom_components/uniled/reproduce_state.py
+.gitignore

custom_components/uniled/config_flow.py

@@ -28,6 +28,7 @@
     CONF_PORT,
     CONF_PROTOCOL,
     CONF_USERNAME,
+    Platform,
 )
 from .lib.net.device import UNILED_TRANSPORT_NET, UniledNetDevice, UniledNetModel
 from .lib.ble.device import UNILED_TRANSPORT_BLE, UniledBleDevice, UniledBleModel
@@ -43,6 +44,8 @@
 )
 from .const import (
     DOMAIN,
+    ATTR_UL_CHIP_TYPE,
+    ATTR_UL_LIGHT_TYPE,
     CONF_UL_RETRY_COUNT as CONF_RETRY_COUNT,
     CONF_UL_TRANSPORT as CONF_TRANSPORT,
     CONF_UL_UPDATE_INTERVAL as CONF_UPDATE_INTERVAL,
@@ -52,7 +55,11 @@
     UNILED_MIN_UPDATE_INTERVAL as MIN_UPDATE_INTERVAL,
     UNILED_DEF_UPDATE_INTERVAL as DEFAULT_UPDATE_INTERVAL,
     UNILED_MAX_UPDATE_INTERVAL as MAX_UPDATE_INTERVAL,
+    UNILED_OPTIONS_ATTRIBUTES,
 )
+from .coordinator import UniledUpdateCoordinator
+
+import homeassistant.helpers.config_validation as cv
 import voluptuous as vol
 import functools
 import operator
@@ -103,6 +110,21 @@ def _mesh_get_context(self) -> dict:
             "options": options,
         }
 
+    async def async_step_mesh_menu(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
+        """Menu step"""
+        mesh_uuid = self.context.get(CONF_MESH_UUID, 0)
+
+        return self.async_show_menu(
+            step_id="mesh_menu",
+            menu_options=["mesh_cloud", "tune_comms"],
+            description_placeholders={
+                "mesh_title": self._mesh_title(mesh_uuid),
+                "mesh_uuid": hex(mesh_uuid),
+            },
+        )
+
     async def async_step_mesh_cloud(
         self, user_input: dict[str, Any] | None = None
     ) -> FlowResult:
@@ -149,13 +171,13 @@ async def async_step_mesh_cloud(
                         if has_changed:
                             # _LOGGER.info(f"Updating configuration: {info['data']}")
                             self.hass.config_entries.async_update_entry(
-                                self.config_entry, data=info['data']
+                                self.config_entry, data=info["data"]
                             )
-                        return self.async_create_entry(title="", data=info['options'])
+                        return self.async_create_entry(title="", data=info["options"])
                     return self.async_create_entry(
                         title=self.context["title_placeholders"]["name"],
-                        data=info['data'],
-                        options=info['options']
+                        data=info["data"],
+                        options=info["options"],
                     )
                 else:
                     errors[CONF_COUNTRY] = "mesh_no_devices"
@@ -197,26 +219,206 @@ async def async_step_init(
         self, user_input: dict[str, Any] | None = None
     ) -> FlowResult:
         """Manage the options."""
-        self._mesh_set_context()
+        self.coordinator: UniledUpdateCoordinator = self.hass.data[DOMAIN][
+            self.config_entry.entry_id
+        ]
+
         if self.config_entry.data.get(CONF_TRANSPORT) == UNILED_TRANSPORT_ZNG:
+            self._mesh_set_context()
             return await self.async_step_mesh_menu()
+
+        config_options = 0
+        for channel in self.coordinator.device.channel_list:
+            if not channel.features:
+                continue
+            for feature in channel.features:
+                if feature.attr in UNILED_OPTIONS_ATTRIBUTES:
+                    config_options += 1
+
+        if config_options:
+            return await self.async_step_conf_menu()
         return await self.async_step_tune_comms()
 
-    async def async_step_mesh_menu(
+    async def async_step_conf_menu(
         self, user_input: dict[str, Any] | None = None
     ) -> FlowResult:
-        """Menu step"""
-        mesh_uuid = self.context.get(CONF_MESH_UUID, 0)
-
+        """Configuration menu step"""
+        if not self.coordinator.device.available:
+            return self.async_abort(reason="not_available")
         return self.async_show_menu(
-            step_id="mesh_menu",
-            menu_options=["tune_comms", "mesh_cloud"],
+            step_id="conf_menu",
+            menu_options=["channels", "tune_comms"],
+        )
+
+    async def async_step_channels(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
+        """Channels Menu"""
+        if self.coordinator.device.channels > 1:
+            if user_input is not None:
+                channel_id = int(user_input.get("channel", 0))
+            else:
+                channels = {}
+                for channel in self.coordinator.device.channel_list:
+                    if not channel.features:
+                        continue
+                    for feature in channel.features:
+                        if feature.attr in UNILED_OPTIONS_ATTRIBUTES:
+                            channels[channel.number] = channel.name
+                            break
+
+                if len(channels) > 1:
+                    data_schema = vol.Schema(
+                        {
+                            vol.Required("channel"): vol.In(
+                                {number: name for number, name in channels.items()}
+                            ),
+                        }
+                    )
+                    return self.async_show_form(
+                        step_id="channels",
+                        data_schema=data_schema,
+                    )
+                elif len(channels) == 1:
+                    channel_id = next(iter(channels))
+                else:
+                    return self.async_abort(reason="no_configurable")
+        else:
+            channel_id = 0
+        channel = self.coordinator.device.channel(channel_id)
+        self.context["channel"] = channel
+        if channel.has(ATTR_UL_LIGHT_TYPE) or channel.has(ATTR_UL_CHIP_TYPE):
+            return await self.async_step_conf_type()
+        return await self.async_step_conf_channel()
+
+    async def async_step_conf_type(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
+        """Configure type"""
+        errors: dict[str, str] = {}
+        channel = self.context.get("channel", None)
+
+        if (conf_value := channel.get(ATTR_UL_LIGHT_TYPE, None)) is not None:
+            conf_attr = ATTR_UL_LIGHT_TYPE
+        elif (conf_value := channel.get(ATTR_UL_CHIP_TYPE, None)) is not None:
+            conf_attr = ATTR_UL_CHIP_TYPE
+        else:
+            return await self.async_step_conf_channel()
+
+        if not self.coordinator.device.available:
+            errors[conf_attr] = "not_available"
+
+        if user_input is not None and not errors:
+            conf_value = user_input.get(conf_attr, conf_value)
+            if conf_value != channel.get(conf_attr, conf_value):
+                if await self.coordinator.device.async_set_state(
+                    channel, conf_attr, conf_value
+                ):
+                    self.options[conf_attr] = conf_value
+                    return self.async_create_entry(title="", data=self.options)
+                else:
+                    errors[conf_attr] = "unknown"
+            else:
+                return await self.async_step_conf_channel()
+
+        data_schema = vol.Schema(
+            {
+                vol.Required(conf_attr, default=conf_value): SelectSelector(
+                    SelectSelectorConfig(
+                        mode=SelectSelectorMode.DROPDOWN,
+                        options=self.coordinator.device.get_list(channel, conf_attr),
+                    )
+                ),
+            }
+        )
+        return self.async_show_form(
+            step_id="conf_type",
+            data_schema=data_schema,
             description_placeholders={
-                "mesh_title": self._mesh_title(mesh_uuid),
-                "mesh_uuid": hex(mesh_uuid),
+                "channel_name": channel.name,
             },
+            errors=errors,
         )
 
+    async def async_step_conf_channel(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
+        """Configure a channel"""
+        errors: dict[str, str] = {}
+        channel = self.context.get("channel", None)
+
+        if user_input is not None and not errors:
+            for conf_attr, conf_value in user_input.items():
+                _LOGGER.warn("%s = %s", conf_attr, conf_value)
+                if not self.coordinator.device.available:
+                    errors[conf_attr] = "not_available"
+                elif conf_value != channel.get(conf_attr, conf_value):
+                    if await self.coordinator.device.async_set_state(
+                        channel, conf_attr, conf_value
+                    ):
+                        self.options[conf_attr] = conf_value
+                    else:
+                        errors[conf_attr] = "unknown"
+
+        if user_input is None or errors:
+            schema = None
+            for conf_attr in UNILED_OPTIONS_ATTRIBUTES:
+                if conf_attr == ATTR_UL_LIGHT_TYPE or conf_attr == ATTR_UL_CHIP_TYPE:
+                    continue
+                for feature in channel.features:
+                    if feature.attr != conf_attr:
+                        continue
+                    if (conf_value := channel.get(conf_attr, None)) is None:
+                        break
+                    if feature.platform == Platform.NUMBER:
+                        option = {
+                            vol.Required(conf_attr, default=conf_value): vol.All(
+                                vol.Coerce(int),
+                                vol.Range(min=feature.min_value, max=feature.max_value),
+                            ),
+                        }
+                    elif feature.platform == Platform.SELECT:
+                        option = {
+                            vol.Required(conf_attr, default=conf_value): SelectSelector(
+                                SelectSelectorConfig(
+                                    mode=SelectSelectorMode.DROPDOWN,
+                                    options=self.coordinator.device.get_list(
+                                        channel, conf_attr
+                                    ),
+                                )
+                            ),
+                        }
+                    elif feature.platform == Platform.SWITCH:
+                        option = {vol.Required(conf_attr, default=conf_value): cv.boolean}
+                    else:
+                        _LOGGER.warning(
+                            "Unsupported feature platform: '%s' for '%s'.",
+                            feature.platform,
+                            feature.attr,
+                        )
+                        break
+                    if option and schema is None:
+                        schema = vol.Schema(option)
+                    elif option and schema:
+                        schema = schema.extend(option)
+                    break
+
+            if schema is not None:
+                return self.async_show_form(
+                    step_id="conf_channel",
+                    data_schema=schema,
+                    description_placeholders={
+                        "channel_name": channel.name,
+                        "light_type": channel.get(ATTR_UL_LIGHT_TYPE, None),
+                        "chip_type": channel.get(ATTR_UL_CHIP_TYPE, None),
+                    },
+                    errors=errors,
+                )
+
+        if not self.coordinator.device.available:
+            return self.async_abort(reason="not_available")
+        return self.async_create_entry(title="", data=self.options)
+
     async def async_step_tune_comms(
         self, user_input: dict[str, Any] | None = None
     ) -> FlowResult:

custom_components/uniled/entity.py

@@ -49,6 +49,8 @@
     ATTR_UL_STATUS,
     ATTR_UL_SUGGESTED_AREA,
     ATTR_UL_TOTAL_PIXELS,
+    UNILED_ENTITY_ATTRIBUTES,
+    UNILED_OPTIONS_ATTRIBUTES,
     UNILED_STATE_CHANGE_LATENCY,
 )
 
@@ -132,6 +134,7 @@ def async_uniled_entity_update(
             if (
                 not feature.platform.startswith(platform)
                 or feature.group == UniledGroup.OPTION
+                or feature.attr in UNILED_OPTIONS_ATTRIBUTES
             ):
                 continue
             if entity := async_add_entity(coordinator, channel, feature):
@@ -181,19 +184,18 @@ def __init__(
         self._device: UniledDevice = coordinator.device
         self._channel: UniledChannel = channel
         self._feature: UniledAttribute = feature
+        # TODO: This is buggy: self.platform is not set, throwing errors?
+        # self._attr_translation_key = feature.attr
         self._attr_has_entity_name = True
 
         if channel.name:
             if feature.name.lower() not in channel.name.lower():
                 self._attr_name = f"{channel.name} {feature.name}"
             else:
                 self._attr_name = f"{channel.name}"
-        else:
+        elif feature.name and feature.name.lower() != feature.platform:
             self._attr_name = f"{feature.name}"
 
-        if self._attr_name.lower() == feature.platform:
-            self._attr_name = None
-
         base_unique_id = coordinator.entry.unique_id or coordinator.entry.entry_id
         self._attr_unique_id = f"_{base_unique_id}"
 
@@ -205,6 +207,7 @@ def __init__(
             self._attr_unique_id = f"{self._attr_unique_id}_{key}"
 
         # _LOGGER.debug("%s: %s (%s)", self._attr_unique_id, self._attr_name, channel.name)
+        # _LOGGER.warn("%s: %s %s %s", self._attr_unique_id, self.name, self.feature.platform, self.platform)
 
         self._attr_entity_registry_enabled_default = feature.enabled
         self._attr_entity_category = None
@@ -291,6 +294,8 @@ def extra_state_attributes(self):
         extra = {}
         if self.feature and self.feature.extra:
             for x in self.feature.extra:
+                if x in UNILED_ENTITY_ATTRIBUTES:
+                    continue
                 if (value := self.device.get_state(self.channel, x)) is not None:
                     extra[x] = value
         return extra

custom_components/uniled/lib/ble/banlanx2.py

@@ -404,12 +404,7 @@ def parse_notifications(
 
         if not device.master.features:
             features = [
-                LightStripFeature(extra=(
-                    ATTR_UL_LIGHT_MODE,
-                    ATTR_UL_LIGHT_MODE_NUMBER,
-                    ATTR_UL_EFFECT_NUMBER,
-                    ATTR_UL_EFFECT_SPEED,
-                )),
+                LightStripFeature(extra=UNILED_CONTROL_ATTRIBUTES),
                 EffectTypeFeature(),
                 EffectSpeedFeature(BANLANX2_MAX_EFFECT_SPEED),
                 EffectLengthFeature(BANLANX2_MAX_EFFECT_LENGTH),

custom_components/uniled/lib/ble/banlanx3.py

@@ -256,14 +256,7 @@ def parse_notifications(
 
         if not device.master.features:
             features = [
-                LightStripFeature(
-                    extra=(
-                        ATTR_UL_LIGHT_MODE,
-                        ATTR_UL_LIGHT_MODE_NUMBER,
-                        ATTR_UL_EFFECT_NUMBER,
-                        ATTR_UL_EFFECT_SPEED,
-                    )
-                ),
+                LightStripFeature(extra=UNILED_CONTROL_ATTRIBUTES),
                 EffectTypeFeature(),
                 EffectSpeedFeature(BANLANX3_MAX_EFFECT_SPEED),
                 ChipOrderFeature(),