From a84b10f905237a05596607868e34aa685c6d04a5 Mon Sep 17 00:00:00 2001 From: epenet <6771947+epenet@users.noreply.github.com> Date: Wed, 1 Oct 2025 12:00:38 +0000 Subject: [PATCH] Update quirks loader --- homeassistant/components/tuya/__init__.py | 5 ++ .../components/tuya/quirks/__init__.py | 49 +++++++++++++++++++ .../components/tuya/quirks/device_quirk.py | 12 ++++- .../components/tuya/quirks/registry.py | 15 ++++++ 4 files changed, 80 insertions(+), 1 deletion(-) diff --git a/homeassistant/components/tuya/__init__.py b/homeassistant/components/tuya/__init__.py index a01f3da1ba58..d67ecafa7e92 100644 --- a/homeassistant/components/tuya/__init__.py +++ b/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 diff --git a/homeassistant/components/tuya/quirks/__init__.py b/homeassistant/components/tuya/quirks/__init__.py index 7625bf3dad90..635704f6c423 100644 --- a/homeassistant/components/tuya/quirks/__init__.py +++ b/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" + ) diff --git a/homeassistant/components/tuya/quirks/device_quirk.py b/homeassistant/components/tuya/quirks/device_quirk.py index 7bc55b75a1b2..1dc876d1892e 100644 --- a/homeassistant/components/tuya/quirks/device_quirk.py +++ b/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)) diff --git a/homeassistant/components/tuya/quirks/registry.py b/homeassistant/components/tuya/quirks/registry.py index 0e23f48f021d..9a552614d84d 100644 --- a/homeassistant/components/tuya/quirks/registry.py +++ b/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)