Option flow for custom input names, prevent duplicate devices

.github/workflows/ruff.yaml

@@ -1,4 +1,4 @@
-name: Ruff
+name: Linting
 on:
   push:
     branches: [ main ]
@@ -9,4 +9,4 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v4
-      - uses: chartboost/ruff-action@v1
+      - uses: chartboost/ruff-action@v1

.github/workflows/unittest.yaml

@@ -0,0 +1,22 @@
+name: Tests
+on:
+  push:
+    branches: [ main ]
+  pull_request:
+    branches: [ main ]
+jobs:
+  unittest:
+    runs-on: ubuntu-latest
+    steps:
+    - uses: actions/checkout@v4
+    - name: Setup Python
+      uses: actions/setup-python@v3
+      with:
+        python-version: "3.12"
+    - name: Install dependencies
+      run: |
+        python -m pip install --upgrade pip
+        pip install homeassistant bidict
+    - name: Run tests
+      run: |
+        python -m unittest discover -s tests/ -v

README.md

@@ -27,6 +27,14 @@ Obviously not every single feature can be controlled, only the basics:
 
 The communication is done using Python's `asyncio` and requires no external libraries
 
+## Development
+
+### Tests
+
+```bash
+python3 -m unittest discover -s tests/ -v
+```
+
 ## License
 
 GNU GENERAL PUBLIC LICENSE version 3

custom_components/extron/__init__.py

@@ -1,5 +1,7 @@
 """The Extron integration."""
 
+import logging
+
 from dataclasses import dataclass
 
 from homeassistant.config_entries import ConfigEntry
@@ -8,9 +10,11 @@
 from homeassistant.exceptions import ConfigEntryNotReady
 from homeassistant.helpers.device_registry import DeviceInfo, format_mac
 
+from custom_components.extron.const import OPTION_INPUT_NAMES
 from custom_components.extron.extron import AuthenticationError, ExtronDevice
 
 PLATFORMS: list[Platform] = [Platform.MEDIA_PLAYER, Platform.SENSOR, Platform.BUTTON]
+_LOGGER = logging.getLogger(__name__)
 
 
 @dataclass
@@ -24,13 +28,15 @@ class DeviceInformation:
 class ExtronConfigEntryRuntimeData:
     device: ExtronDevice
     device_information: DeviceInformation
+    input_names: list[str]
 
 
 async def get_device_information(device: ExtronDevice) -> DeviceInformation:
     mac_address = await device.query_mac_address()
     model_name = await device.query_model_name()
     firmware_version = await device.query_firmware_version()
     part_number = await device.query_part_number()
+    ip_address = await device.query_ip_address()
 
     device_info = DeviceInfo(
         identifiers={(DOMAIN, format_mac(mac_address))},
@@ -39,6 +45,7 @@ async def get_device_information(device: ExtronDevice) -> DeviceInformation:
         model=model_name,
         sw_version=firmware_version,
         serial_number=part_number,
+        configuration_url=f"http://{ip_address}/",
     )
 
     return DeviceInformation(mac_address=format_mac(mac_address), model_name=model_name, device_info=device_info)
@@ -55,10 +62,15 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
     except Exception as e:
         raise ConfigEntryNotReady("Unable to connect") from e
 
-    # Store the device and information about as runtime data in the entry
+    # Store runtime information
     device_information = await get_device_information(device)
-    entry.runtime_data = ExtronConfigEntryRuntimeData(device, device_information)
+    input_names = entry.options.get(OPTION_INPUT_NAMES, [])
+    entry.runtime_data = ExtronConfigEntryRuntimeData(device, device_information, input_names)
+
+    # Register a listener for option updates
+    entry.async_on_unload(entry.add_update_listener(entry_update_listener))
 
+    _LOGGER.info(f"Initializing entry with runtime data: {entry.runtime_data}")
     await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
 
     return True
@@ -67,3 +79,8 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
 async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
     """Unload a config entry."""
     return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
+
+
+async def entry_update_listener(hass: HomeAssistant, config_entry: ConfigEntry):
+    # Reload the entry when options have been changed
+    await hass.config_entries.async_reload(config_entry.entry_id)

custom_components/extron/button.py

@@ -14,7 +14,6 @@ async def async_setup_entry(hass, entry: ConfigEntry, async_add_entities):
     runtime_data: ExtronConfigEntryRuntimeData = entry.runtime_data
     device = runtime_data.device
     device_information = runtime_data.device_information
-    logger.info(device_information)
 
     # Add entities
     async_add_entities([ExtronRebootButton(device, device_information)])

custom_components/extron/config_flow.py

@@ -6,10 +6,11 @@
 
 import voluptuous as vol
 
-from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
+from homeassistant.config_entries import ConfigEntry, ConfigFlow, OptionsFlow
+from homeassistant.helpers.device_registry import format_mac
 from homeassistant.helpers.selector import selector
 
-from .const import CONF_DEVICE_TYPE, CONF_HOST, CONF_PASSWORD, CONF_PORT, DOMAIN
+from .const import CONF_DEVICE_TYPE, CONF_HOST, CONF_PASSWORD, CONF_PORT, DOMAIN, OPTION_INPUT_NAMES
 from .extron import AuthenticationError, DeviceType, ExtronDevice
 
 _LOGGER = logging.getLogger(__name__)
@@ -38,7 +39,7 @@ class ExtronConfigFlow(ConfigFlow, domain=DOMAIN):
 
     VERSION = 1
 
-    async def async_step_user(self, user_input: dict[str, Any] | None = None) -> ConfigFlowResult:
+    async def async_step_user(self, user_input: dict[str, Any] | None = None):
         """Handle the initial step."""
         errors: dict[str, str] = {}
         if user_input is not None:
@@ -51,13 +52,45 @@ async def async_step_user(self, user_input: dict[str, Any] | None = None) -> Con
                 model_name = await extron_device.query_model_name()
                 title = f"Extron {model_name}"
 
+                # Make a unique ID for the entry, prevent adding the same device twice
+                unique_id = format_mac(await extron_device.query_mac_address())
+                await self.async_set_unique_id(unique_id)
+                self._abort_if_unique_id_configured()
+
                 # Disconnect, we'll connect again later, this was just for validation
                 await extron_device.disconnect()
             except AuthenticationError:
                 errors["base"] = "invalid_auth"
-            except Exception:
+            except (BrokenPipeError, ConnectionError, OSError):  # all technically OSError
                 errors["base"] = "cannot_connect"
             else:
                 return self.async_create_entry(title=title, data=user_input)
 
         return self.async_show_form(step_id="user", data_schema=STEP_USER_DATA_SCHEMA, errors=errors)
+
+    @staticmethod
+    def async_get_options_flow(config_entry: ConfigEntry) -> OptionsFlow:
+        """Create the options flow"""
+        return ExtronOptionsFlowHandler(config_entry)
+
+
+class ExtronOptionsFlowHandler(OptionsFlow):
+    def __init__(self, config_entry: ConfigEntry) -> None:
+        """Initialize options flow."""
+        self.config_entry = config_entry
+
+    async def async_step_init(self, user_input: dict[str, Any] | None = None):
+        """Manage optional settings for the entry."""
+        if user_input is not None:
+            return self.async_create_entry(title="", data=user_input)
+
+        return self.async_show_form(
+            step_id="init",
+            data_schema=vol.Schema(
+                {
+                    vol.Optional(
+                        OPTION_INPUT_NAMES, default=self.config_entry.options.get(OPTION_INPUT_NAMES)
+                    ): selector({"text": {"multiple": True}}),
+                }
+            ),
+        )

custom_components/extron/const.py

@@ -6,3 +6,5 @@
 CONF_PORT = "port"
 CONF_PASSWORD = "password"
 CONF_DEVICE_TYPE = "device_type"
+
+OPTION_INPUT_NAMES = "input_names"

custom_components/extron/extron.py

@@ -102,6 +102,9 @@ async def query_part_number(self):
     async def query_mac_address(self):
         return await self.run_command("\x1b" + "CH")
 
+    async def query_ip_address(self):
+        return await self.run_command("\x1b" + "CI")
+
     async def reboot(self):
         await self.run_command("\x1b" + "1BOOT")
 
@@ -113,8 +116,8 @@ def __init__(self, device: ExtronDevice) -> None:
     def get_device(self) -> ExtronDevice:
         return self._device
 
-    async def view_input(self):
-        return await self._device.run_command("$")
+    async def view_input(self) -> int:
+        return int(await self._device.run_command("$"))
 
     async def select_input(self, input: int):
         await self._device.run_command(f"{str(input)}$")
@@ -154,8 +157,8 @@ def __init__(self, device: ExtronDevice) -> None:
     def get_device(self) -> ExtronDevice:
         return self._device
 
-    async def view_input(self):
-        return await self._device.run_command("!")
+    async def view_input(self) -> int:
+        return int(await self._device.run_command("!"))
 
     async def select_input(self, input: int):
         await self._device.run_command(f"{str(input)}!")

custom_components/extron/manifest.json

@@ -11,7 +11,7 @@
   "documentation": "https://www.home-assistant.io/integrations/extron",
   "homekit": {},
   "iot_class": "local_polling",
-  "requirements": [],
+  "requirements": ["bidict"],
   "ssdp": [],
   "zeroconf": []
 }

custom_components/extron/media_player.py

@@ -1,5 +1,6 @@
 import logging
 
+from bidict import bidict
 from homeassistant.components.media_player import MediaPlayerEntity, MediaPlayerEntityFeature, MediaPlayerState
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.helpers.entity import DeviceInfo
@@ -11,26 +12,32 @@
 logger = logging.getLogger(__name__)
 
 
+def make_source_bidict(num_sources: int, input_names: list[str]) -> bidict:
+    # Use user-defined input name for the source when available
+    return bidict({i + 1: input_names[i] if i < len(input_names) else str(i + 1) for i in range(num_sources)})
+
+
 async def async_setup_entry(hass, entry: ConfigEntry, async_add_entities):
     # Extract stored runtime data from the entry
     runtime_data: ExtronConfigEntryRuntimeData = entry.runtime_data
     device = runtime_data.device
     device_information = runtime_data.device_information
-    logger.info(device_information)
+    input_names = runtime_data.input_names
 
     # Add entities
     if entry.data[CONF_DEVICE_TYPE] == DeviceType.SURROUND_SOUND_PROCESSOR.value:
         ssp = SurroundSoundProcessor(device)
-        async_add_entities([ExtronSurroundSoundProcessor(ssp, device_information)])
+        async_add_entities([ExtronSurroundSoundProcessor(ssp, device_information, input_names)])
     elif entry.data[CONF_DEVICE_TYPE] == DeviceType.HDMI_SWITCHER.value:
         hdmi_switcher = HDMISwitcher(device)
-        async_add_entities([ExtronHDMISwitcher(hdmi_switcher, device_information)])
+        async_add_entities([ExtronHDMISwitcher(hdmi_switcher, device_information, input_names)])
 
 
 class AbstractExtronMediaPlayerEntity(MediaPlayerEntity):
-    def __init__(self, device: ExtronDevice, device_information: DeviceInformation) -> None:
+    def __init__(self, device: ExtronDevice, device_information: DeviceInformation, input_names: list[str]) -> None:
         self._device = device
         self._device_information = device_information
+        self._input_names = input_names
         self._device_class = "receiver"
         self._state = MediaPlayerState.PLAYING
 
@@ -66,12 +73,12 @@ def name(self):
 
 
 class ExtronSurroundSoundProcessor(AbstractExtronMediaPlayerEntity):
-    def __init__(self, ssp: SurroundSoundProcessor, device_information: DeviceInformation):
-        super().__init__(ssp.get_device(), device_information)
+    def __init__(self, ssp: SurroundSoundProcessor, device_information: DeviceInformation, input_names: list[str]):
+        super().__init__(ssp.get_device(), device_information, input_names)
         self._ssp = ssp
 
         self._source = None
-        self._source_list = ["1", "2", "3", "4", "5"]
+        self._source_bidict = self.create_source_bidict()
         self._volume = None
         self._muted = False
 
@@ -86,7 +93,7 @@ def get_device_type(self):
         return DeviceType.SURROUND_SOUND_PROCESSOR
 
     async def async_update(self):
-        self._source = await self._ssp.view_input()
+        self._source = self._source_bidict.get(await self._ssp.view_input())
         self._muted = await self._ssp.is_muted()
         volume = await self._ssp.get_volume_level()
         self._volume = volume / 100
@@ -109,10 +116,13 @@ def source(self):
 
     @property
     def source_list(self):
-        return self._source_list
+        return list(self._source_bidict.values())
+
+    def create_source_bidict(self) -> bidict:
+        return make_source_bidict(5, self._input_names)
 
     async def async_select_source(self, source):
-        await self._ssp.select_input(int(source))
+        await self._ssp.select_input(self._source_bidict.inverse.get(source))
         self._source = source
 
     async def async_mute_volume(self, mute: bool) -> None:
@@ -129,38 +139,46 @@ async def async_volume_down(self) -> None:
 
 
 class ExtronHDMISwitcher(AbstractExtronMediaPlayerEntity):
-    def __init__(self, hdmi_switcher: HDMISwitcher, device_information: DeviceInformation) -> None:
-        super().__init__(hdmi_switcher.get_device(), device_information)
+    def __init__(
+        self, hdmi_switcher: HDMISwitcher, device_information: DeviceInformation, input_names: list[str]
+    ) -> None:
+        super().__init__(hdmi_switcher.get_device(), device_information, input_names)
         self._hdmi_switcher = hdmi_switcher
 
         self._state = MediaPlayerState.PLAYING
         self._source = None
+        self._source_bidict = self.create_source_bidict()
 
     _attr_supported_features = MediaPlayerEntityFeature.SELECT_SOURCE
 
     def get_device_type(self):
         return DeviceType.HDMI_SWITCHER
 
     async def async_update(self):
-        self._source = await self._hdmi_switcher.view_input()
+        self._source = self._source_bidict.get(await self._hdmi_switcher.view_input())
 
     @property
     def source(self):
         return self._source
 
     @property
     def source_list(self):
+        return list(self._source_bidict.values())
+
+    def create_source_bidict(self) -> bidict:
         model_name = self._device_information.model_name
         sw = model_name.split(" ")[0]
 
         if sw == "SW2":
-            return ["1", "2"]
+            num_sources = 2
         elif sw == "SW4":
-            return ["1", "2", "3", "4"]
+            num_sources = 4
         elif sw == "SW6":
-            return ["1", "2", "3", "4", "5", "6"]
+            num_sources = 6
         else:
-            return ["1", "2", "3", "4", "5", "6", "7", "8"]
+            num_sources = 8
+
+        return make_source_bidict(num_sources, self._input_names)
 
     async def async_select_source(self, source: str):
         await self._hdmi_switcher.select_input(int(source))

custom_components/extron/sensor.py

@@ -20,7 +20,6 @@ async def async_setup_entry(hass, entry: ConfigEntry, async_add_entities):
     runtime_data: ExtronConfigEntryRuntimeData = entry.runtime_data
     device = runtime_data.device
     device_information = runtime_data.device_information
-    logger.info(device_information)
 
     # Add entities
     if entry.data[CONF_DEVICE_TYPE] == DeviceType.SURROUND_SOUND_PROCESSOR.value: