[lvgl] Merge dict-extend chains to speed up schema construction (#16614)

esphome/components/lvgl/__init__.py

@@ -1,3 +1,4 @@
+import functools
 import importlib
 from pathlib import Path
 import pkgutil
@@ -79,7 +80,7 @@ from .schemas import (
     WIDGET_TYPES,
     any_widget_schema,
     container_schema,
-    obj_schema,
+    obj_dict,
 )
 from .styles import styles_to_code, theme_to_code
 from .touchscreens import touchscreen_schema, touchscreens_to_code
@@ -518,16 +519,32 @@ def add_hello_world(config):
     return config
 
 
-def _theme_schema(value):
+@functools.cache
+def _build_theme_schema(
+    widget_types: tuple[tuple[str, widgets.WidgetType], ...],
+) -> cv.Schema:
+    # The theme schema is value-independent: it depends only on the set of
+    # registered widget types. Key the cache on a snapshot of WIDGET_TYPES so
+    # that an external component registering a new widget after the first
+    # validation (legal per any_widget_schema's lazy-evaluation contract)
+    # produces a fresh tuple, a cache miss, and a rebuilt schema -- the cache
+    # self-heals instead of stale-rejecting valid themes. See obj_dict() in
+    # schemas.py for why chained .extend() is avoided here.
     return cv.Schema(
         {
             cv.Optional(df.CONF_DARK_MODE, default=False): cv.boolean,
             **{
-                cv.Optional(name): obj_schema(w).extend(FULL_STYLE_SCHEMA)
-                for name, w in WIDGET_TYPES.items()
+                cv.Optional(name): cv.Schema(
+                    {**obj_dict(w), **FULL_STYLE_SCHEMA.schema}
+                )
+                for name, w in widget_types
             },
         }
-    )(value)
+    )
+
+
+def _theme_schema(value: dict) -> dict:
+    return _build_theme_schema(tuple(WIDGET_TYPES.items()))(value)
 
 
 FINAL_VALIDATE_SCHEMA = final_validation

esphome/components/lvgl/schemas.py

@@ -378,15 +378,33 @@ TRIGGER_EVENT_MAP = {
 }
 
 
-def part_schema(parts):
+def part_dict(parts: tuple[str, ...] | list[str]) -> dict[Any, Any]:
+    """
+    Return the raw mapping used by part_schema, so callers can merge it into a
+    larger dict and avoid chained .extend() calls (each .extend() recompiles the
+    whole mapping, turning the build into O(N^2)).
+
+    Invariant: the source schemas spread here (STATE_SCHEMA, FLAG_SCHEMA, the
+    nested STATE_SCHEMA values) must use the default extra=PREVENT_EXTRA and
+    required=False and must not register any add_extra/prepend_extra
+    validators. Reaching into .schema and rebuilding via cv.Schema(...) keeps
+    only the mapping; non-default extra/required and any _extra_schemas would
+    be silently dropped.
+    """
+    return {
+        **STATE_SCHEMA.schema,
+        **FLAG_SCHEMA.schema,
+        **{cv.Optional(part): STATE_SCHEMA for part in parts},
+    }
+
+
+def part_schema(parts: tuple[str, ...] | list[str]) -> cv.Schema:
     """
     Generate a schema for the various parts (e.g. main:, indicator:) of a widget type
     :param parts:  The parts to include
     :return: The schema
     """
-    return STATE_SCHEMA.extend(FLAG_SCHEMA).extend(
-        {cv.Optional(part): STATE_SCHEMA for part in parts}
-    )
+    return cv.Schema(part_dict(parts))
 
 
 def automation_schema(typ: LvType):
@@ -462,6 +480,43 @@ def base_update_schema(widget_type: WidgetType | LvType, parts):
     return schema
 
 
+# Memoize obj_dict() the same way _OBJ_SCHEMA_CACHE memoizes obj_schema().
+# automation_schema(w.w_type) builds fresh Trigger.template(...) objects on
+# every call, so without this cache _theme_schema pays that cost per widget
+# per validation. Callers must treat the returned dict as immutable. The
+# _theme_schema caller spreads it into a fresh dict, which is safe; the
+# obj_schema caller passes it directly to cv.Schema(...) -- voluptuous stores
+# the mapping by reference but never mutates it (.extend() copies first), so
+# the alias is also safe today. Adding in-place mutation of obj_schema(w).schema
+# would corrupt this cache.
+_OBJ_DICT_CACHE: dict[int, tuple[WidgetType, dict[Any, Any]]] = {}
+
+
+def obj_dict(widget_type: WidgetType) -> dict[Any, Any]:
+    """
+    Return the raw mapping used by obj_schema, so callers can merge it into a
+    larger dict and avoid chained .extend() calls.
+
+    Inherits the same source-schema invariant documented on part_dict: any
+    schema spread into this mapping must use the default extra=PREVENT_EXTRA
+    and required=False and must carry no add_extra/prepend_extra validators.
+
+    The returned mapping is cached and must be treated as immutable by callers.
+    """
+    cached = _OBJ_DICT_CACHE.get(id(widget_type))
+    if cached is not None and cached[0] is widget_type:
+        return cached[1]
+    built = {
+        **part_dict(widget_type.parts),
+        **ALIGN_TO_SCHEMA,
+        **automation_schema(widget_type.w_type),
+        cv.Optional(CONF_STATE): SET_STATE_SCHEMA,
+        cv.Optional(CONF_GROUP): cv.use_id(lv_group_t),
+    }
+    _OBJ_DICT_CACHE[id(widget_type)] = (widget_type, built)
+    return built
+
+
 # Widget types are module-level singletons populated at import time, so we
 # can cache compiled obj_schemas by widget_type identity for the lifetime of
 # the process. The strong reference in the value keeps the key (an id()
@@ -469,7 +524,7 @@ def base_update_schema(widget_type: WidgetType | LvType, parts):
 _OBJ_SCHEMA_CACHE: dict[int, tuple[WidgetType, cv.Schema]] = {}
 
 
-def obj_schema(widget_type: WidgetType):
+def obj_schema(widget_type: WidgetType) -> cv.Schema:
     """
     Create a schema for a widget type itself i.e. no allowance for children
     :param widget_type:
@@ -478,17 +533,7 @@ def obj_schema(widget_type: WidgetType):
     cached = _OBJ_SCHEMA_CACHE.get(id(widget_type))
     if cached is not None and cached[0] is widget_type:
         return cached[1]
-    schema = (
-        part_schema(widget_type.parts)
-        .extend(ALIGN_TO_SCHEMA)
-        .extend(automation_schema(widget_type.w_type))
-        .extend(
-            {
-                cv.Optional(CONF_STATE): SET_STATE_SCHEMA,
-                cv.Optional(CONF_GROUP): cv.use_id(lv_group_t),
-            }
-        )
-    )
+    schema = cv.Schema(obj_dict(widget_type))
     _OBJ_SCHEMA_CACHE[id(widget_type)] = (widget_type, schema)
     return schema
 

tests/component_tests/lvgl/test_schema_dict_helpers.py

@@ -0,0 +1,236 @@
+"""Tests for part_dict / obj_dict / part_schema / obj_schema mapping contracts.
+
+These guard the dict-merge refactor: the dict helpers must keep returning the
+same logical mapping as the chained-extend version produced, and the
+corresponding Schema(...) wrappers must accept and reject the same configs.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Generator
+
+import pytest
+import voluptuous as vol
+
+from esphome import config_validation as cv
+import esphome.components.lvgl
+from esphome.components.lvgl import (
+    _theme_schema,
+    defines as df,
+    schemas as lvgl_schemas,
+)
+from esphome.components.lvgl.schemas import (
+    ALIGN_TO_SCHEMA,
+    FLAG_SCHEMA,
+    FULL_STYLE_SCHEMA,
+    STATE_SCHEMA,
+    STYLE_SCHEMA,
+    WIDGET_TYPES,
+    automation_schema,
+    obj_dict,
+    obj_schema,
+    part_dict,
+    part_schema,
+)
+from esphome.components.lvgl.types import LvType
+from esphome.components.lvgl.widgets import WidgetType
+
+
+@pytest.fixture(autouse=True)
+def _clear_obj_dict_cache() -> Generator[None]:
+    cache = getattr(lvgl_schemas, "_OBJ_DICT_CACHE", None)
+    if cache is not None:
+        cache.clear()
+    # The lazily-built theme schema is cached on _build_theme_schema; clear it
+    # too so each test starts from a clean slate.
+    build_theme = getattr(esphome.components.lvgl, "_build_theme_schema", None)
+    if build_theme is not None and hasattr(build_theme, "cache_clear"):
+        build_theme.cache_clear()
+    yield
+    if cache is not None:
+        cache.clear()
+    if build_theme is not None and hasattr(build_theme, "cache_clear"):
+        build_theme.cache_clear()
+
+
+def _marker_names(mapping) -> set[str]:
+    """Return the underlying string names of every voluptuous Marker key."""
+    names: set[str] = set()
+    for key in mapping:
+        if isinstance(key, vol.Marker):
+            schema = key.schema
+            if isinstance(schema, str):
+                names.add(schema)
+    return names
+
+
+def _widget_type(name: str = "obj"):
+    wt = WIDGET_TYPES.get(name)
+    assert wt is not None, f"widget type {name!r} not registered"
+    return wt
+
+
+def test_part_dict_includes_state_flag_and_part_keys() -> None:
+    parts = ("indicator", "knob")
+    keys = _marker_names(part_dict(parts))
+
+    assert {"indicator", "knob"} <= keys
+    assert _marker_names(STATE_SCHEMA.schema) <= keys
+    assert _marker_names(FLAG_SCHEMA.schema) <= keys
+
+
+def test_obj_dict_extends_part_dict_with_align_automation_state_group() -> None:
+    wt = _widget_type("obj")
+    part_keys = _marker_names(part_dict(wt.parts))
+    obj_keys = _marker_names(obj_dict(wt))
+
+    assert part_keys <= obj_keys
+    assert _marker_names(ALIGN_TO_SCHEMA) <= obj_keys
+    assert _marker_names(automation_schema(wt.w_type)) <= obj_keys
+    assert {"state", "group"} <= obj_keys
+
+
+def test_obj_dict_is_memoized_by_widget_type() -> None:
+    wt = _widget_type("obj")
+    first = obj_dict(wt)
+    second = obj_dict(wt)
+    assert first is second
+    # Different widget type, different dict.
+    assert obj_dict(_widget_type("label")) is not first
+
+
+def test_part_schema_round_trips_known_state_and_part_settings() -> None:
+    schema = part_schema(("indicator",))
+    out = schema(
+        {
+            "bg_color": 0x112233,
+            "checked": {"bg_color": 0x445566},
+            "indicator": {"bg_color": 0x778899},
+        }
+    )
+    assert out["bg_color"] == 0x112233
+    assert out["checked"]["bg_color"] == 0x445566
+    assert out["indicator"]["bg_color"] == 0x778899
+
+
+def test_part_schema_rejects_unknown_part() -> None:
+    schema = part_schema(("indicator",))
+    with pytest.raises(vol.Invalid):
+        schema({"definitely_not_a_part": {}})
+
+
+@pytest.mark.parametrize("name", sorted(WIDGET_TYPES))
+def test_obj_schema_accepts_empty_config_for_every_widget_type(name: str) -> None:
+    obj_schema(_widget_type(name))({})
+
+
+def test_obj_schema_accepts_align_to_and_state_group() -> None:
+    schema = obj_schema(_widget_type("obj"))
+    out = schema(
+        {
+            df.CONF_ALIGN_TO: {
+                "id": "some_other_widget",
+                df.CONF_ALIGN: "TOP_LEFT",
+            },
+            "state": {"checked": True},
+        }
+    )
+    assert out[df.CONF_ALIGN_TO][df.CONF_ALIGN] == "LV_ALIGN_TOP_LEFT"
+    assert out["state"]["checked"] is True
+
+
+def test_obj_schema_rejects_unknown_top_level_key() -> None:
+    with pytest.raises(vol.Invalid):
+        obj_schema(_widget_type("obj"))({"definitely_not_a_real_key": 1})
+
+
+def test_part_schema_returns_cv_schema_for_extend_callers() -> None:
+    schema = part_schema(("indicator",))
+    extended = schema.extend({cv.Optional("extra_key"): cv.string})
+    out = extended({"extra_key": "value", "bg_color": 0xAABBCC})
+    assert out["extra_key"] == "value"
+    assert out["bg_color"] == 0xAABBCC
+
+
+def test_obj_schema_returns_cv_schema_for_extend_callers() -> None:
+    schema = obj_schema(_widget_type("obj"))
+    extended = schema.extend({cv.Optional("extra_key"): cv.string})
+    extended({"extra_key": "value"})
+
+
+@pytest.mark.parametrize(
+    "schema",
+    [STATE_SCHEMA, FLAG_SCHEMA, STYLE_SCHEMA, FULL_STYLE_SCHEMA],
+)
+def test_spread_sources_carry_no_extra_schemas(schema: cv.Schema) -> None:
+    # part_dict / obj_dict reach into .schema and rebuild via cv.Schema(...),
+    # which silently drops _extra_schemas and any non-default extra/required.
+    # Lock the invariant so a future add_extra() on these sources fails CI
+    # instead of quietly removing validation from part/obj/theme schemas.
+    assert not schema._extra_schemas
+    assert schema.extra is vol.PREVENT_EXTRA
+    assert schema.required is False
+
+
+def test_theme_schema_merges_obj_dict_and_full_style_props() -> None:
+    # _theme_schema is the riskiest merge: obj_dict(w) and FULL_STYLE_SCHEMA.schema
+    # share many STYLE_SCHEMA marker instances. Exercise the merged schema
+    # end-to-end with one key from each side (a STATE_SCHEMA part from obj_dict
+    # and a FULL_STYLE-only property) to lock the behaviour against future
+    # regressions in either source.
+    out = _theme_schema(
+        {
+            df.CONF_DARK_MODE: True,
+            "obj": {
+                "bg_color": 0x112233,
+                "checked": {"bg_color": 0x445566},
+                df.CONF_PAD_ROW: 4,
+                df.CONF_GRID_CELL_X_ALIGN: "CENTER",
+            },
+        }
+    )
+    assert out[df.CONF_DARK_MODE] is True
+    obj_out = out["obj"]
+    assert obj_out["bg_color"] == 0x112233
+    assert obj_out["checked"]["bg_color"] == 0x445566
+    assert obj_out[df.CONF_PAD_ROW] == 4
+    assert obj_out[df.CONF_GRID_CELL_X_ALIGN] == "LV_GRID_ALIGN_CENTER"
+
+
+def test_theme_schema_self_heals_when_a_widget_type_is_registered_later() -> None:
+    # _build_theme_schema is functools.cached on a snapshot of WIDGET_TYPES.
+    # any_widget_schema explicitly supports external components registering
+    # widgets lazily, and the device builder revalidates in-process, so a
+    # widget registered after first use must invalidate the cached snapshot.
+    _theme_schema({df.CONF_DARK_MODE: True})  # populate the cache
+
+    name = "test_self_heal_widget"
+    assert name not in WIDGET_TYPES
+    # is_mock=True skips registration side-effects; insert into WIDGET_TYPES
+    # manually so the next theme call sees the new entry.
+    WIDGET_TYPES[name] = WidgetType(name, LvType("test_fake_t"), (), is_mock=True)
+    try:
+        out = _theme_schema({df.CONF_DARK_MODE: False, name: {"bg_color": 0x010203}})
+        assert out[name]["bg_color"] == 0x010203
+    finally:
+        WIDGET_TYPES.pop(name, None)
+
+
+@pytest.mark.parametrize(
+    "schema",
+    [STATE_SCHEMA, FLAG_SCHEMA, STYLE_SCHEMA, FULL_STYLE_SCHEMA],
+)
+def test_spread_sources_have_no_top_level_marker_defaults(schema: cv.Schema) -> None:
+    # _theme_schema merges obj_dict(w) with FULL_STYLE_SCHEMA.schema; on a key
+    # collision, dict-spread keeps the first source's marker (and its default)
+    # but the last source's value, whereas .extend() would take both from the
+    # later source. The two are equivalent today because the overlapping
+    # markers are the same instances (both derive from STYLE_SCHEMA) and none
+    # carry a top-level default. Lock that so a future divergent default would
+    # fail CI rather than silently drift the merged validation.
+    offenders = [
+        marker.schema
+        for marker in schema.schema
+        if isinstance(marker, vol.Optional) and marker.default is not vol.UNDEFINED
+    ]
+    assert not offenders, f"top-level Optional with default: {offenders}"