Update quirks loader

homeassistant/components/tuya/__init__.py

@@ -31,6 +31,7 @@ from .const import (
     TUYA_DISCOVERY_NEW,
     TUYA_HA_SIGNAL_UPDATE_ENTITY,
 )
+from .quirks import register_tuya_quirks
 
 # Suppress logs from the library, it logs unneeded on error
 logging.getLogger("tuya_sharing").setLevel(logging.CRITICAL)
@@ -103,6 +104,10 @@ async def async_setup_entry(hass: HomeAssistant, entry: TuyaConfigEntry) -> bool
             model_id=device.product_id,
         )
 
+    # Should be loaded from configuration.yaml
+    # but for now, we can use a hardcoded path for testing
+    quirks_path = "/config/tuya_quirks/"
+    await hass.async_add_executor_job(register_tuya_quirks, quirks_path)
     await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
     # If the device does not register any entities, the device does not need to subscribe
     # So the subscription is here

homeassistant/components/tuya/quirks/__init__.py

@@ -2,6 +2,12 @@
 
 from __future__ import annotations
 
+import importlib
+import logging
+import pathlib
+import pkgutil
+import sys
+
 from .device_quirk import TuyaCoverDefinition, TuyaCoverDeviceClass, TuyaDeviceQuirk
 from .homeassistant import parse_enum
 from .registry import QuirksRegistry
@@ -14,5 +20,48 @@ __all__ = [
     "TuyaDeviceQuirk",
     "parse_enum",
 ]
+_LOGGER = logging.getLogger(__name__)
 
 TUYA_QUIRKS_REGISTRY = QuirksRegistry()
+
+
+def register_tuya_quirks(custom_quirks_path: str | None = None) -> None:
+    """Register all quirks with zigpy, including optional custom quirks."""
+
+    if custom_quirks_path is not None:
+        TUYA_QUIRKS_REGISTRY.purge_custom_quirks(custom_quirks_path)
+
+    # Import all quirks in the `zhaquirks` package first
+    for _importer, modname, _ispkg in pkgutil.walk_packages(
+        path=__path__,
+        prefix=__name__ + ".",
+    ):
+        _LOGGER.debug("Loading quirks module %r", modname)
+        importlib.import_module(modname)
+
+    if custom_quirks_path is None:
+        return
+
+    path = pathlib.Path(custom_quirks_path)
+    _LOGGER.debug("Loading custom quirks from %r", path)
+
+    loaded = False
+
+    # Treat the custom quirk path (e.g. `/config/tuya_quirks/`) itself as a module
+    for importer, modname, _ispkg in pkgutil.walk_packages(path=[str(path)]):
+        _LOGGER.debug("Loading custom quirk module %r", modname)
+
+        try:
+            spec = importer.find_spec(modname)  # type: ignore[call-arg]
+            module = importlib.util.module_from_spec(spec)  # type: ignore[arg-type]
+            sys.modules[modname] = module
+            spec.loader.exec_module(module)  # type: ignore[union-attr]
+        except Exception:
+            _LOGGER.exception("Unexpected exception importing custom quirk %r", modname)
+        else:
+            loaded = True
+
+    if loaded:
+        _LOGGER.warning(
+            "Loaded custom quirks. Please contribute them to https://github.com/TBD"
+        )

homeassistant/components/tuya/quirks/device_quirk.py

@@ -3,6 +3,8 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
+import inspect
+import pathlib
 from typing import TYPE_CHECKING, Self
 
 from .homeassistant import TuyaCoverDeviceClass
@@ -31,7 +33,6 @@ class TuyaCoverDefinition(BaseTuyaDefinition):
     set_position_dp_code: str | None = None
 
 
-@dataclass
 class TuyaDeviceQuirk:
     """Quirk for Tuya device."""
 
@@ -43,6 +44,15 @@ class TuyaDeviceQuirk:
         self._applies_to = []
         self.cover_definitions = []
 
+        current_frame = inspect.currentframe()
+        if TYPE_CHECKING:
+            assert current_frame is not None
+        caller = current_frame.f_back
+        if TYPE_CHECKING:
+            assert caller is not None
+        self.quirk_file = pathlib.Path(caller.f_code.co_filename)
+        self.quirk_file_line = caller.f_lineno
+
     def applies_to(self, *, category: str, product_id: str) -> Self:
         """Set the device type the quirk applies to."""
         self._applies_to.append((category, product_id))

homeassistant/components/tuya/quirks/registry.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import logging
 from typing import TYPE_CHECKING, Self
 
 from tuya_sharing import CustomerDevice
@@ -9,6 +10,8 @@ from tuya_sharing import CustomerDevice
 if TYPE_CHECKING:
     from .device_quirk import TuyaDeviceQuirk
 
+_LOGGER = logging.getLogger(__name__)
+
 
 class QuirksRegistry:
     """Registry for Tuya quirks."""
@@ -34,3 +37,15 @@ class QuirksRegistry:
     def get_quirk_for_device(self, device: CustomerDevice) -> TuyaDeviceQuirk | None:
         """Get the quirk for a specific device."""
         return self._quirks.get(device.category, {}).get(device.product_id)
+
+    def purge_custom_quirks(self, custom_quirks_root: str) -> None:
+        """Purge custom quirks from the registry."""
+        for category_quirks in self._quirks.values():
+            to_remove = []
+            for product_id, quirk in category_quirks.items():
+                if quirk.quirk_file.is_relative_to(custom_quirks_root):
+                    to_remove.append(product_id)
+
+            for product_id in to_remove:
+                _LOGGER.debug("Removing stale custom quirk: %s", product_id)
+                category_quirks.pop(product_id)