[core] Select I/O expander pin providers by I2C address (#18939)

Co-authored-by: Claude Sonnet 5 <noreply@anthropic.com>
2026-10-08 03:54:03 +00:00 · 2026-10-07 09:18:40 -10:00
parent 1c0d406cbb
commit 6dec759261
23 changed files with 476 additions and 19 deletions
@@ -79,7 +79,7 @@ MAX6956_PIN_SCHEMA = pins.gpio_base_schema(
    mode_validator=validate_mode,
 ).extend(
    {
-        cv.Required(CONF_MAX6956): cv.use_id(MAX6956),
+        cv.Required(CONF_MAX6956): pins.use_id_or_address(MAX6956),
    }
 )

@@ -60,7 +60,7 @@ MCP23016_PIN_SCHEMA = pins.gpio_base_schema(
    invertible=True,
 ).extend(
    {
-        cv.Required(CONF_MCP23016): cv.use_id(MCP23016),
+        cv.Required(CONF_MCP23016): pins.use_id_or_address(MCP23016),
    }
 )

@@ -74,7 +74,7 @@ MCP23XXX_PIN_SCHEMA = pins.gpio_base_schema(
    invertible=True,
 ).extend(
    {
-        cv.Required(CONF_MCP23XXX): cv.use_id(MCP23XXXBase),
+        cv.Required(CONF_MCP23XXX): pins.use_id_or_address(MCP23XXXBase),
        cv.Optional(CONF_INTERRUPT, default="NO_INTERRUPT"): cv.enum(
            MCP23XXX_INTERRUPT_MODES, upper=True
        ),
@@ -14,6 +14,8 @@ DsiDriverChip(
    pclk_frequency="60MHz",
    lane_bit_rate="730Mbps",
    color_order="RGB",
+    requires={"psram", "pi4ioe5v6408"},
+    reset_pin={"pi4ioe5v6408": {"address": 0x43}, "number": 4},
    initsequence=[
        (0xFF, 0x98, 0x81, 0x01),  # Select Page 1
        (0xB7, 0x03),  # Pad control - 2 lane
@@ -67,6 +69,8 @@ TAB5_ST7123 = DsiDriverChip(
    pclk_frequency="80MHz",
    lane_bit_rate="960Mbps",
    color_order="RGB",
+    requires={"psram", "pi4ioe5v6408"},
+    reset_pin={"pi4ioe5v6408": {"address": 0x43}, "number": 4},
    initsequence=[
        (0x01,),
        (0x60, 0x71, 0x23, 0xa2),
@@ -118,6 +122,8 @@ DsiDriverChip(
    pclk_frequency="70MHz",
    lane_bit_rate="965Mbps",
    color_order="RGB",
+    requires={"psram", "pi4ioe5v6408"},
+    reset_pin={"pi4ioe5v6408": {"address": 0x43}, "number": 4},
    initsequence=[
        (0x01,),
        (0x60, 0x71, 0x21, 0xA2),
@@ -60,7 +60,7 @@ def validate_mode(value: ConfigType) -> ConfigType:
 PCA6416A_PIN_SCHEMA = cv.All(
    {
        cv.GenerateID(): cv.declare_id(PCA6416AGPIOPin),
-        cv.Required(CONF_PCA6416A): cv.use_id(PCA6416AComponent),
+        cv.Required(CONF_PCA6416A): pins.use_id_or_address(PCA6416AComponent),
        cv.Required(CONF_NUMBER): cv.int_range(min=0, max=15),
        cv.Optional(CONF_MODE, default={}): cv.All(
            {
@@ -66,7 +66,7 @@ PCA9554_PIN_SCHEMA = pins.gpio_base_schema(
    mode_validator=validate_mode,
 ).extend(
    {
-        cv.Required(CONF_PCA9554): cv.use_id(PCA9554Component),
+        cv.Required(CONF_PCA9554): pins.use_id_or_address(PCA9554Component),
    }
 )

@@ -63,7 +63,7 @@ PCF8574_PIN_SCHEMA = pins.gpio_base_schema(
    invertible=True,
 ).extend(
    {
-        cv.Required(CONF_PCF8574): cv.use_id(PCF8574Component),
+        cv.Required(CONF_PCF8574): pins.use_id_or_address(PCF8574Component),
    }
 )

@@ -74,7 +74,7 @@ PI4IOE5V6408_PIN_SCHEMA = pins.gpio_base_schema(
    mode_validator=validate_mode,
 ).extend(
    {
-        cv.Required(CONF_PI4IOE5V6408): cv.use_id(PI4IOE5V6408Component),
+        cv.Required(CONF_PI4IOE5V6408): pins.use_id_or_address(PI4IOE5V6408Component),
    }
 )

@@ -133,7 +133,7 @@ CONF_SX1509 = "sx1509"
 SX1509_PIN_SCHEMA = cv.All(
    {
        cv.GenerateID(): cv.declare_id(SX1509GPIOPin),
-        cv.Required(CONF_SX1509): cv.use_id(SX1509Component),
+        cv.Required(CONF_SX1509): pins.use_id_or_address(SX1509Component),
        cv.Required(CONF_NUMBER): cv.int_range(min=0, max=15),
        cv.Optional(CONF_MODE, default={}): cv.All(
            {
@@ -62,7 +62,7 @@ TCA9555_PIN_SCHEMA = pins.gpio_base_schema(
    invertible=True,
 ).extend(
    {
-        cv.Required(CONF_TCA9555): cv.use_id(TCA9555Component),
+        cv.Required(CONF_TCA9555): pins.use_id_or_address(TCA9555Component),
    }
 )

@@ -54,7 +54,7 @@ def validate_pin(pin: int) -> int:
 XL9535_PIN_SCHEMA = cv.All(
    {
        cv.GenerateID(): cv.declare_id(XL9535GPIOPin),
-        cv.Required(CONF_XL9535): cv.use_id(XL9535Component),
+        cv.Required(CONF_XL9535): pins.use_id_or_address(XL9535Component),
        cv.Required(CONF_NUMBER): cv.All(cv.int_range(min=0, max=17), validate_pin),
        cv.Optional(CONF_MODE, default={}): cv.All(
            {
@@ -1129,6 +1129,42 @@ class IDPassValidationStep(ConfigValidationStep):
                    if inherits:
                        matches.append(v[0])

+                if id.match_config:
+                    # Disambiguate among same-type candidates by comparing their own
+                    # declared config against the requested key/value pairs, e.g. an
+                    # I2C address, instead of requiring a single unambiguous candidate.
+                    criteria = ", ".join(f"{k}={v}" for k, v in id.match_config.items())
+                    filtered = [
+                        m
+                        for m in matches
+                        if isinstance(
+                            candidate_conf := result.get_config_for_path(
+                                result.get_path_for_id(m)[:-1]
+                            ),
+                            dict,
+                        )
+                        and all(
+                            candidate_conf.get(k) == v
+                            for k, v in id.match_config.items()
+                        )
+                    ]
+                    if len(filtered) == 1:
+                        id.id = filtered[0].id
+                    elif len(filtered) == 0:
+                        result.add_str_error(
+                            f"Couldn't find a '{id.type}' matching {criteria}. "
+                            "Are you missing a hub declaration, or is the address wrong?",
+                            path,
+                        )
+                    else:
+                        ids = ", ".join(f"'{m.id}'" for m in filtered)
+                        result.add_str_error(
+                            f"Multiple '{id.type}' instances match {criteria}: {ids}. "
+                            "You must assign an explicit ID to the one you want to use.",
+                            path,
+                        )
+                    continue
+
                if len(matches) == 0:
                    result.add_str_error(
                        f"Couldn't find any component that can be used for '{id.type}'. Are you missing a hub declaration?",
@@ -385,7 +385,9 @@ class Lambda:


 class ID:
-    def __init__(self, id, is_declaration=False, type=None, is_manual=None):
+    def __init__(
+        self, id, is_declaration=False, type=None, is_manual=None, match_config=None
+    ):
        self.id = id
        if is_manual is None:
            self.is_manual = id is not None
@@ -393,6 +395,10 @@ class ID:
            self.is_manual = is_manual
        self.is_declaration = is_declaration
        self.type: MockObjClass | None = type
+        # When set, an unnamed (id=None) searching ID is disambiguated among same-type
+        # candidates by matching these key/value pairs against each candidate's own
+        # declared config, instead of requiring exactly one candidate to exist.
+        self.match_config: dict | None = match_config

    def resolve(self, registered_ids):
        from esphome.config_validation import RESERVED_IDS
@@ -431,6 +437,7 @@ class ID:
            is_declaration=self.is_declaration,
            type=self.type,
            is_manual=self.is_manual,
+            match_config=self.match_config,
        )


@@ -1,4 +1,4 @@
-from collections.abc import Callable
+from collections.abc import Callable, Hashable
 from functools import reduce
 from logging import Logger
 import operator
@@ -6,6 +6,7 @@ from typing import Any

 import esphome.config_validation as cv
 from esphome.const import (
+    CONF_ADDRESS,
    CONF_ALLOW_OTHER_USES,
    CONF_IGNORE_STRAPPING_WARNING,
    CONF_INPUT,
@@ -17,8 +18,9 @@ from esphome.const import (
    CONF_PULLDOWN,
    CONF_PULLUP,
 )
-from esphome.core import CORE
+from esphome.core import CORE, ID
 from esphome.cpp_generator import MockObjClass
+from esphome.schema_extractors import schema_extractor


 class PinRegistry(dict):
@@ -69,11 +71,15 @@ class PinRegistry(dict):
        # evaluate here so a validation failure skips the rest
        result = self[key][1](conf)
        if CONF_NUMBER in result:
-            # key maps to the pin schema
-            if key != CORE.target_platform:
-                pin_key = (key, conf[key], result[CONF_NUMBER])
-            else:
-                pin_key = (key, key, result[CONF_NUMBER])
+            # key maps to the pin schema. The bucket used here doesn't need to
+            # identify the provider precisely -- final_validate() regroups pin
+            # usage by the *resolved* provider id once IDPassValidationStep has
+            # run, so this only has to be hashable. A hub selected by
+            # match_config (e.g. address) is given as a dict, which isn't.
+            ref = conf[key] if key != CORE.target_platform else key
+            if not isinstance(ref, Hashable):
+                ref = "match_config"
+            pin_key = (key, ref, result[CONF_NUMBER])
            if pin_key not in self.pins_used:
                self.pins_used[pin_key] = []
            # client_id identifies the instance of the providing component
@@ -107,7 +113,18 @@ class PinRegistry(dict):
        Run the final validation for all pins, and check for reuse
        :param fconf: The full config
        """
-        for (key, _, _), pin_list in self.pins_used.items():
+        # Regroup by (schema key, resolved provider id, pin number) rather than
+        # the validate()-time bucket, so the same physical pin reached via an
+        # explicit id and via {address: ...} (or an omitted id) is recognized as
+        # one reuse group regardless of which syntax was used to reference the hub.
+        grouped: dict[tuple, list] = {}
+        for (key, _, number), pin_list in self.pins_used.items():
+            for entry in pin_list:
+                client_id = entry[1]
+                provider_id = client_id.id if isinstance(client_id, ID) else client_id
+                grouped.setdefault((key, provider_id, number), []).append(entry)
+
+        for (key, _, _), pin_list in grouped.items():
            count = len(pin_list)  # number of places same pin used.
            final_val_fun = self[key][2]  # final validation function
            for pin_path, client_id, pin_config in pin_list:
@@ -131,6 +148,40 @@ class PinRegistry(dict):
 PIN_SCHEMA_REGISTRY = PinRegistry()


+def use_id_or_address(hub_type, address_key=CONF_ADDRESS):
+    """Build a validator for a pin-provider hub reference.
+
+    Accepts everything `cv.use_id(hub_type)` accepts (an explicit id, or omitted to
+    auto-select the sole instance of `hub_type`), plus a mapping like
+    `{address: 0x21}` to instead select the instance of `hub_type` whose own
+    declared config has a matching `address_key`. Use this for pin schemas of
+    I2C-addressed hubs (I/O expanders) so a board with two instances of the
+    same hub type can pick one without assigning it an explicit id.
+    """
+    id_validator = cv.use_id(hub_type)
+
+    @schema_extractor("use_id")
+    def validator(value):
+        if isinstance(value, dict):
+            address = cv.Schema({cv.Required(address_key): cv.i2c_address})(value)[
+                address_key
+            ]
+            return ID(
+                None,
+                is_declaration=False,
+                type=hub_type,
+                # The user explicitly gave selection criteria (unlike the plain
+                # omitted-id auto-pick), so treat the resolved id as manual: it
+                # must survive strip_default_ids() and appear in `esphome config`
+                # output, not be silently dropped as if it were a default.
+                is_manual=True,
+                match_config={address_key: address},
+            )
+        return id_validator(value)
+
+    return validator
+
+
 def _set_mode(value, default_mode):
    if CONF_MODE not in value:
        return {**value, CONF_MODE: default_mode}
@@ -0,0 +1,23 @@
+packages:
+  i2c: !include ../../test_build_components/common/i2c/esp32-p4-idf.yaml
+
+esp_ldo:
+  - id: ldo_id
+    channel: 3
+    voltage: 2.5V
+
+psram:
+
+pi4ioe5v6408:
+  # A second, decoy hub at a different address, so this test only passes if
+  # the model's reset_pin default actually resolves pi4ioe1 by its address
+  # (0x43) -- with only one hub declared, the single-instance auto-pick
+  # fallback would resolve it too, even if address matching regressed.
+  - id: pi4ioe1
+    address: 0x43
+  - id: pi4ioe2
+    address: 0x44
+
+display:
+  - platform: mipi_dsi
+    model: M5STACK-TAB5
@@ -2,6 +2,9 @@ xl9535:
  - id: xl9535_hub
    i2c_id: i2c_bus
    address: 0x20
+  - id: xl9535_hub_2
+    i2c_id: i2c_bus
+    address: 0x21

 binary_sensor:
  - platform: gpio
@@ -20,3 +23,12 @@ binary_sensor:
      mode:
        input: true
      inverted: false
+  - platform: gpio
+    name: XL9535 Pin 1 By Address
+    pin:
+      xl9535:
+        address: 0x21
+      number: 1
+      mode:
+        input: true
+      inverted: false
@@ -447,6 +447,85 @@ def test_device_duplicate_id(
    assert "ID duplicate_device redefined!" in captured.out


+def test_expander_pin_selected_by_address(yaml_file: Callable[[str], str]) -> None:
+    """A pin provider hub with an omitted id can be selected by its address."""
+    result = load_config_from_fixture(
+        yaml_file, "expander_pin_by_address.yaml", FIXTURES_DIR
+    )
+    assert result is not None
+
+    resolved = result["binary_sensor"][0]["pin"]["xl9535"]
+    assert resolved.id == "xl9535_b"
+    # Explicit selection criteria must survive strip_default_ids(), unlike a
+    # plain omitted-id auto-pick.
+    assert resolved.is_manual is True
+
+
+def test_expander_pin_selected_by_address_no_match(
+    yaml_file: Callable[[str], str], capsys: pytest.CaptureFixture[str]
+) -> None:
+    """Selecting an address that matches no hub of that type fails clearly."""
+    result = load_config_from_fixture(
+        yaml_file, "expander_pin_by_address_no_match.yaml", FIXTURES_DIR
+    )
+    assert result is None
+
+    captured = capsys.readouterr()
+    assert (
+        "Couldn't find a 'xl9535::XL9535Component' matching address=0x21. "
+        "Are you missing a hub declaration, or is the address wrong?" in captured.out
+    )
+
+
+def test_expander_pin_selected_by_address_ambiguous(
+    yaml_file: Callable[[str], str], capsys: pytest.CaptureFixture[str]
+) -> None:
+    """Two hubs sharing the same address (e.g. on different buses) still need an id."""
+    result = load_config_from_fixture(
+        yaml_file, "expander_pin_by_address_ambiguous.yaml", FIXTURES_DIR
+    )
+    assert result is None
+
+    captured = capsys.readouterr()
+    assert (
+        "Multiple 'xl9535::XL9535Component' instances match address=0x20: "
+        "'xl9535_a', 'xl9535_b'. You must assign an explicit ID to the one you "
+        "want to use." in captured.out
+    )
+
+
+def test_expander_pin_ambiguous_without_match_config(
+    yaml_file: Callable[[str], str], capsys: pytest.CaptureFixture[str]
+) -> None:
+    """Omitting both id and address with multiple hubs of the same type still
+    falls back to the original "too many candidates" error."""
+    result = load_config_from_fixture(
+        yaml_file, "expander_pin_ambiguous_no_address.yaml", FIXTURES_DIR
+    )
+    assert result is None
+
+    captured = capsys.readouterr()
+    assert (
+        "Too many candidates found for 'xl9535' type 'xl9535::XL9535Component' "
+        "Some are 'xl9535_a', 'xl9535_b'" in captured.out
+    )
+
+
+def test_expander_pin_reuse_detected_across_reference_syntax(
+    yaml_file: Callable[[str], str], capsys: pytest.CaptureFixture[str]
+) -> None:
+    """The same physical pin, reached once by explicit id and once by address,
+    must still be flagged as reused -- reuse detection keys on the resolved
+    provider id, not on which syntax was used to reference it."""
+    result = load_config_from_fixture(
+        yaml_file, "expander_pin_reuse_across_syntax.yaml", FIXTURES_DIR
+    )
+    assert result is None
+
+    captured = capsys.readouterr()
+    assert "Pin 5 is used in multiple places" in captured.out
+
+
 def test_substitution_with_id(
    yaml_file: Callable[[str], str], capsys: pytest.CaptureFixture[str]
 ) -> None:
@@ -0,0 +1,24 @@
+esphome:
+  name: test
+
+esp32:
+  board: esp32dev
+
+i2c:
+  sda: GPIO21
+  scl: GPIO22
+
+xl9535:
+  - id: xl9535_a
+    address: 0x20
+  - id: xl9535_b
+    address: 0x21
+
+binary_sensor:
+  - platform: gpio
+    name: "Expander Pin"
+    pin:
+      xl9535:
+      number: 1
+      mode:
+        input: true
@@ -0,0 +1,25 @@
+esphome:
+  name: test
+
+esp32:
+  board: esp32dev
+
+i2c:
+  sda: GPIO21
+  scl: GPIO22
+
+xl9535:
+  - id: xl9535_a
+    address: 0x20
+  - id: xl9535_b
+    address: 0x21
+
+binary_sensor:
+  - platform: gpio
+    name: "Expander Pin"
+    pin:
+      xl9535:
+        address: 0x21
+      number: 1
+      mode:
+        input: true
@@ -0,0 +1,31 @@
+esphome:
+  name: test
+
+esp32:
+  board: esp32dev
+
+i2c:
+  - id: bus_a
+    sda: GPIO21
+    scl: GPIO22
+  - id: bus_b
+    sda: GPIO18
+    scl: GPIO19
+
+xl9535:
+  - id: xl9535_a
+    i2c_id: bus_a
+    address: 0x20
+  - id: xl9535_b
+    i2c_id: bus_b
+    address: 0x20
+
+binary_sensor:
+  - platform: gpio
+    name: "Expander Pin"
+    pin:
+      xl9535:
+        address: 0x20
+      number: 1
+      mode:
+        input: true
@@ -0,0 +1,23 @@
+esphome:
+  name: test
+
+esp32:
+  board: esp32dev
+
+i2c:
+  sda: GPIO21
+  scl: GPIO22
+
+xl9535:
+  - id: xl9535_a
+    address: 0x20
+
+binary_sensor:
+  - platform: gpio
+    name: "Expander Pin"
+    pin:
+      xl9535:
+        address: 0x21
+      number: 1
+      mode:
+        input: true
@@ -0,0 +1,30 @@
+esphome:
+  name: test
+
+esp32:
+  board: esp32dev
+
+i2c:
+  sda: GPIO21
+  scl: GPIO22
+
+xl9535:
+  - id: xl9535_a
+    address: 0x20
+
+binary_sensor:
+  - platform: gpio
+    name: "Expander Pin By Id"
+    pin:
+      xl9535: xl9535_a
+      number: 5
+      mode:
+        input: true
+  - platform: gpio
+    name: "Expander Pin By Address"
+    pin:
+      xl9535:
+        address: 0x20
+      number: 5
+      mode:
+        input: true
@@ -0,0 +1,110 @@
+"""Unit tests for esphome/pins.py."""
+
+import pytest
+
+from esphome import config_validation as cv, pins
+import esphome.codegen as cg
+from esphome.const import CONF_ADDRESS
+from esphome.core import ID
+from esphome.schema_extractors import SCHEMA_EXTRACT
+
+SomeHub = cg.esphome_ns.class_("SomeHub")
+
+
+def test_use_id_or_address_with_explicit_id() -> None:
+    """An explicit id string still resolves exactly like cv.use_id."""
+    validator = pins.use_id_or_address(SomeHub)
+    result = validator("my_hub")
+
+    assert isinstance(result, ID)
+    assert result.id == "my_hub"
+    assert result.type is SomeHub
+    assert result.is_declaration is False
+    assert result.match_config is None
+
+
+def test_use_id_or_address_with_omitted_id() -> None:
+    """Omitting the id (None) still falls back to the single-instance auto-pick."""
+    validator = pins.use_id_or_address(SomeHub)
+    result = validator(None)
+
+    assert isinstance(result, ID)
+    assert result.id is None
+    assert result.type is SomeHub
+    assert result.match_config is None
+
+
+def test_use_id_or_address_with_address_mapping() -> None:
+    """A {address: ...} mapping produces an unnamed ID with match_config set."""
+    validator = pins.use_id_or_address(SomeHub)
+    result = validator({CONF_ADDRESS: 0x21})
+
+    assert isinstance(result, ID)
+    assert result.id is None
+    assert result.type is SomeHub
+    assert result.is_declaration is False
+    assert result.match_config == {CONF_ADDRESS: 0x21}
+
+
+def test_use_id_or_address_with_string_address() -> None:
+    """A hex string address is normalized the same way cv.i2c_address does."""
+    validator = pins.use_id_or_address(SomeHub)
+    result = validator({CONF_ADDRESS: "0x21"})
+
+    assert result.match_config == {CONF_ADDRESS: 0x21}
+
+
+def test_use_id_or_address_mapping_requires_address_key() -> None:
+    """A mapping without the address key is rejected, not silently ignored."""
+    validator = pins.use_id_or_address(SomeHub)
+
+    with pytest.raises(cv.Invalid):
+        validator({})
+
+
+def test_use_id_or_address_mapping_rejects_invalid_address() -> None:
+    """An out-of-range address is still validated by cv.i2c_address."""
+    validator = pins.use_id_or_address(SomeHub)
+
+    with pytest.raises(cv.Invalid):
+        validator({CONF_ADDRESS: 0x1FF})
+
+
+def test_use_id_or_address_custom_address_key() -> None:
+    """A custom address_key is honored both in the mapping and match_config."""
+    validator = pins.use_id_or_address(SomeHub, address_key="i2c_address")
+    result = validator({"i2c_address": 0x10})
+
+    assert result.match_config == {"i2c_address": 0x10}
+
+
+def test_use_id_or_address_marks_address_match_as_manual() -> None:
+    """The resolved id must survive strip_default_ids(), unlike a plain
+    omitted-id auto-pick, since the user explicitly gave selection criteria."""
+    validator = pins.use_id_or_address(SomeHub)
+    result = validator({CONF_ADDRESS: 0x21})
+
+    assert result.is_manual is True
+
+
+def test_use_id_or_address_schema_extract_passthrough() -> None:
+    """Called with the SCHEMA_EXTRACT sentinel (as build_language_schema.py
+    does), the wrapper must fall through to cv.use_id's own handling and
+    return the hub type, not try to treat it as an address mapping."""
+    validator = pins.use_id_or_address(SomeHub)
+
+    assert validator(SCHEMA_EXTRACT) is SomeHub
+
+
+def test_use_id_or_address_registers_as_use_id_schema(monkeypatch) -> None:
+    """The wrapper must be discoverable by build_language_schema.py as a
+    `use_id` schema (like a bare cv.use_id), or the generated language schema
+    silently drops hub-id completion for every pin schema using it."""
+    from esphome import schema_extractors
+
+    monkeypatch.setattr(schema_extractors, "EnableSchemaExtraction", True)
+    monkeypatch.setattr(schema_extractors, "hidden_schemas", {})
+
+    validator = pins.use_id_or_address(SomeHub)
+
+    assert schema_extractors.hidden_schemas.get(repr(validator)) == "use_id"