Allow validation of pins based on hub config (#5647)

esphome/components/pca6416a/__init__.py

@@ -64,7 +64,7 @@ PCA6416A_PIN_SCHEMA = cv.All(
 )
 
 
-@pins.PIN_SCHEMA_REGISTRY.register("pca6416a", PCA6416A_PIN_SCHEMA)
+@pins.PIN_SCHEMA_REGISTRY.register(CONF_PCA6416A, PCA6416A_PIN_SCHEMA)
 async def pca6416a_pin_to_code(config):
     var = cg.new_Pvariable(config[CONF_ID])
     parent = await cg.get_variable(config[CONF_PCA6416A])

esphome/components/pca9554/__init__.py

@@ -92,7 +92,7 @@ PCA9554_PIN_SCHEMA = cv.All(
 )
 
 
-@pins.PIN_SCHEMA_REGISTRY.register("pca9554", PCA9554_PIN_SCHEMA)
+@pins.PIN_SCHEMA_REGISTRY.register(CONF_PCA9554, PCA9554_PIN_SCHEMA)
 async def pca9554_pin_to_code(config):
     var = cg.new_Pvariable(config[CONF_ID])
     parent = await cg.get_variable(config[CONF_PCA9554])

esphome/components/pcf8574/__init__.py

@@ -65,7 +65,7 @@ PCF8574_PIN_SCHEMA = cv.All(
 )
 
 
-@pins.PIN_SCHEMA_REGISTRY.register("pcf8574", PCF8574_PIN_SCHEMA)
+@pins.PIN_SCHEMA_REGISTRY.register(CONF_PCF8574, PCF8574_PIN_SCHEMA)
 async def pcf8574_pin_to_code(config):
     var = cg.new_Pvariable(config[CONF_ID])
     parent = await cg.get_variable(config[CONF_PCF8574])

esphome/components/sn74hc165/__init__.py

@@ -77,7 +77,15 @@ SN74HC165_PIN_SCHEMA = cv.All(
 )
 
 
-@pins.PIN_SCHEMA_REGISTRY.register(CONF_SN74HC165, SN74HC165_PIN_SCHEMA)
+def sn74hc165_pin_final_validate(pin_config, parent_config):
+    max_pins = parent_config[CONF_SR_COUNT] * 8
+    if pin_config[CONF_NUMBER] >= max_pins:
+        raise cv.Invalid(f"Pin number must be less than {max_pins}")
+
+
+@pins.PIN_SCHEMA_REGISTRY.register(
+    CONF_SN74HC165, SN74HC165_PIN_SCHEMA, sn74hc165_pin_final_validate
+)
 async def sn74hc165_pin_to_code(config):
     var = cg.new_Pvariable(config[CONF_ID])
     await cg.register_parented(var, config[CONF_SN74HC165])

esphome/config.py

@@ -10,7 +10,7 @@ from contextlib import contextmanager
 
 import voluptuous as vol
 
-from esphome import core, yaml_util, loader
+from esphome import core, yaml_util, loader, pins
 import esphome.core.config as core_config
 from esphome.const import (
     CONF_ESPHOME,
@@ -645,14 +645,40 @@ class FinalValidateValidationStep(ConfigValidationStep):
             # If result already has errors, skip this step
             return
 
-        if self.comp.final_validate_schema is None:
-            return
-
         token = fv.full_config.set(result)
 
         conf = result.get_nested_item(self.path)
         with result.catch_error(self.path):
-            self.comp.final_validate_schema(conf)
+            if self.comp.final_validate_schema is not None:
+                self.comp.final_validate_schema(conf)
+
+            fconf = fv.full_config.get()
+
+            def _check_pins(c):
+                for value in c.values():
+                    if not isinstance(value, dict):
+                        continue
+                    for key, (
+                        _,
+                        _,
+                        pin_final_validate,
+                    ) in pins.PIN_SCHEMA_REGISTRY.items():
+                        if (
+                            key != CORE.target_platform
+                            and key in value
+                            and pin_final_validate is not None
+                        ):
+                            pin_final_validate(fconf, value)
+
+            # Check for pin configs and a final_validate schema in the pin registry
+            confs = conf
+            if not isinstance(
+                confs, list
+            ):  # Handle components like SPI that have a list instead of MULTI_CONF
+                confs = [conf]
+            for c in confs:
+                if c:  # Some component have None or empty schemas
+                    _check_pins(c)
 
         fv.full_config.reset(token)
 

esphome/cpp_helpers.py

@@ -35,7 +35,7 @@ async def gpio_pin_expression(conf):
         return None
     from esphome import pins
 
-    for key, (func, _) in pins.PIN_SCHEMA_REGISTRY.items():
+    for key, (func, _, _) in pins.PIN_SCHEMA_REGISTRY.items():
         if key in conf:
             return await coroutine(func)(conf)
     return await coroutine(pins.PIN_SCHEMA_REGISTRY[CORE.target_platform][0])(conf)

esphome/pins.py

@@ -11,10 +11,10 @@ from esphome.const import (
     CONF_PULLUP,
     CONF_IGNORE_STRAPPING_WARNING,
 )
-from esphome.util import SimpleRegistry
+from esphome.util import PinRegistry
 from esphome.core import CORE
 
-PIN_SCHEMA_REGISTRY = SimpleRegistry()
+PIN_SCHEMA_REGISTRY = PinRegistry()
 
 
 def _set_mode(value, default_mode):

esphome/util.py

@@ -57,6 +57,32 @@ class SimpleRegistry(dict):
         return decorator
 
 
+def _final_validate(parent_id_key, fun):
+    def validator(fconf, pin_config):
+        import esphome.config_validation as cv
+
+        parent_path = fconf.get_path_for_id(pin_config[parent_id_key])[:-1]
+        parent_config = fconf.get_config_for_path(parent_path)
+
+        pin_path = fconf.get_path_for_id(pin_config[const.CONF_ID])[:-1]
+        with cv.prepend_path([cv.ROOT_CONFIG_PATH] + pin_path):
+            fun(pin_config, parent_config)
+
+    return validator
+
+
+class PinRegistry(dict):
+    def register(self, name, schema, final_validate=None):
+        if final_validate is not None:
+            final_validate = _final_validate(name, final_validate)
+
+        def decorator(fun):
+            self[name] = (fun, schema, final_validate)
+            return fun
+
+        return decorator
+
+
 def safe_print(message="", end="\n"):
     from esphome.core import CORE