esphome/script/build_language_schema.py

#!/usr/bin/env python3
import argparse
import inspect
import json
import os
from pathlib import Path
import re

import voluptuous as vol

# NOTE: Cannot import other esphome components globally as a modification in vol_schema
# is needed before modules are loaded
import esphome.schema_extractors as ejs

ejs.EnableSchemaExtraction = True

# schema format:
# Schemas are split in several files in json format, one for core stuff, one for each platform (sensor, binary_sensor, etc) and
# one for each component (dallas, sim800l, etc.) component can have schema for root component/hub and also for platform component,
# e.g. dallas has hub component which has pin and then has the sensor platform which has sensor name, index, etc.
# When files are loaded they are merged in a single object.
# The root format is

S_CONFIG_VAR = "config_var"
S_CONFIG_VARS = "config_vars"
S_CONFIG_SCHEMA = "CONFIG_SCHEMA"
S_COMPONENT = "component"
S_COMPONENTS = "components"
S_PLATFORMS = "platforms"
S_SCHEMA = "schema"
S_SCHEMAS = "schemas"
S_EXTENDS = "extends"
S_TYPE = "type"
S_NAME = "name"

parser = argparse.ArgumentParser()
parser.add_argument(
    "--output-path", default=".", help="Output path", type=os.path.abspath
)
parser.add_argument("--check", action="store_true", help="Check only for CI")

# Module-level ``Namespace`` so helper functions can reference ``args``
# without threading it through every call. ``main()`` fills it via
# ``parser.parse_args(namespace=args)``; tests import this module without
# invoking ``main()`` and rely on the defaults below.
args = argparse.Namespace(output_path=".", check=False)

DUMP_RAW = False
DUMP_UNKNOWN = False
DUMP_PATH = False
JSON_DUMP_PRETTY = True

# store here dynamic load of esphome components
components = {}

schema_core = {}

# output is where all is built
output = {"core": schema_core}
# The full generated output is here here
schema_full = {"components": output}

# A string, string map, key is the str(schema) and value is
# a tuple, first element is the schema reference and second is the schema path given, the schema reference is needed to test as different schemas have same key
known_schemas = {}

solve_registry = []


def get_component_names():
    # pylint: disable-next=redefined-outer-name,reimported
    from esphome.loader import CORE_COMPONENTS_PATH

    # return sorted(
    #     ["esphome", "sensor", "esp32", "esp8266", "adc", "touchscreen", "xpt2046"]
    # )
    return sorted(
        [
            d.name
            for d in CORE_COMPONENTS_PATH.iterdir()
            if not d.name.startswith("__") and d.is_dir()
        ]
    )


def load_components():
    from esphome.config import get_component

    for domain in get_component_names():
        components[domain] = get_component(domain, exception=True)
        assert components[domain] is not None


from esphome.const import (  # noqa: E402
    CONF_TYPE,
    KEY_CORE,
    KEY_FRAMEWORK_VERSION,
    KEY_TARGET_FRAMEWORK,
    KEY_TARGET_PLATFORM,
)
from esphome.core import CORE  # noqa: E402

CORE.data[KEY_CORE] = {
    KEY_TARGET_PLATFORM: "esp8266",
    KEY_TARGET_FRAMEWORK: "arduino",
    KEY_FRAMEWORK_VERSION: "0",
}


load_components()

# Import esphome after loading components (so schema is tracked)
# pylint: disable=wrong-import-position
from esphome import automation, pins  # noqa: E402
from esphome.components import remote_base  # noqa: E402
import esphome.config_validation as cv  # noqa: E402
from esphome.helpers import write_file_if_changed  # noqa: E402
from esphome.loader import CORE_COMPONENTS_PATH, get_platform  # noqa: E402
from esphome.util import Registry  # noqa: E402

# pylint: enable=wrong-import-position

# Scalar ``cv.*`` validators the dumper describes by identity. Extending these
# tuples -- rather than decorating each validator with a schema_extractor --
# mirrors how ``cv.boolean`` / ``cv.string`` / ``cv.int_`` are already handled
# in ``convert()`` and keeps runtime validation completely untouched. Grouped by
# the ``type`` emitted into the schema dump so a language server / visual editor
# knows what YAML each field accepts instead of treating it as free-form.
_CV_STRING_VALIDATORS = (
    cv.icon,
    cv.mac_address,
    cv.url,
    cv.publish_topic,
    cv.subscribe_topic,
    cv.mqtt_payload,
    cv.uuid,
    cv.ssid,
    cv.domain,
    cv.domain_name,
    cv.hostname,
    cv.entity_id,
    cv.git_ref,
    cv.string_no_slash,
    cv.version_number,
    cv.validate_esphome_version,
    cv.validate_id_name,
    cv._validate_entity_name,
    cv.validate_source_shorthand,
    cv.ipv4address,
    cv.ipv6network,
    cv.ipv4address_multi_broadcast,
    cv.time_of_day,
    cv.directory,
    cv.file_,
    cv.dimensions,
    cv.none,
)
_CV_INTEGER_VALIDATORS = (cv.hex_int, cv.percentage_int, cv.mqtt_qos)
_CV_FLOAT_VALIDATORS = (
    cv.percentage,
    cv.possibly_negative_percentage,
    cv.temperature,
    cv.temperature_delta,
    cv.color_temperature,
)
_CV_TIME_VALIDATORS = (
    cv.update_interval,
    cv.time_period_str_unit,
    cv.time_period_str_colon,
)
_CV_LAMBDA_VALIDATORS = (cv.lambda_, cv.returning_lambda)


def sort_obj(obj):
    if isinstance(obj, dict):
        is_enum = obj.get(S_TYPE) == "enum"
        result = {}
        for k, v in sorted(obj.items(), key=lambda x: str(x[0])):
            if is_enum and k == "values" and isinstance(v, dict):
                # Preserve source order of enum options
                result[k] = {vk: sort_obj(vv) for vk, vv in v.items()}
            else:
                result[k] = sort_obj(v)
        return result
    if isinstance(obj, list):
        return [sort_obj(item) for item in obj]
    return obj


def write_file(name, obj):
    full_path = Path(args.output_path) / f"{name}.json"
    sorted_obj = sort_obj(obj)
    if JSON_DUMP_PRETTY:
        json_str = json.dumps(sorted_obj, indent=2)
    else:
        json_str = json.dumps(sorted_obj, separators=(",", ":"))
    write_file_if_changed(full_path, json_str)


def delete_extra_files(keep_names):
    output_path = Path(args.output_path)
    count = 0
    for d in output_path.iterdir():
        if d.suffix == ".json" and d.stem not in keep_names:
            count += 1
            d.unlink()
    return count


def register_module_schemas(key, module, manifest=None):
    count = 0
    for name, schema in module_schemas(module):
        count += 1
        register_known_schema(key, name, schema)

    if (
        manifest
        and manifest.multi_conf
        and key in output
        and S_CONFIG_SCHEMA in output[key][S_SCHEMAS]
    ):
        # Multi conf should allow list of components
        # not sure about 2nd part of the if, might be useless config (e.g. as3935)
        output[key][S_SCHEMAS][S_CONFIG_SCHEMA]["is_list"] = True
    return count


def register_known_schema(module, name, schema):
    if module not in output:
        output[module] = {S_SCHEMAS: {}}
    config = convert_config(schema, f"{module}/{name}")
    if S_TYPE not in config and name != "FINAL_VALIDATE_SCHEMA" and module != "core":
        print(f"Config var without type: {module}.{name}")

    output[module][S_SCHEMAS][name] = config
    repr_schema = repr(schema)
    if repr_schema in known_schemas:
        schema_info = known_schemas[repr_schema]
        schema_info.append((schema, f"{module}.{name}"))
    else:
        known_schemas[repr_schema] = [(schema, f"{module}.{name}")]


def module_schemas(module):
    # This should yield elements in order so extended schemas are resolved properly
    # To do this we check on the source code where the symbol is seen first. Seems to work.
    try:
        module_str = inspect.getsource(module)
    except TypeError:
        # improv
        module_str = ""
    except OSError:
        # some empty __init__ files
        module_str = ""
    schemas = []
    for m_attr_name in dir(module):
        m_attr_obj = getattr(module, m_attr_name)
        if is_convertible_schema(m_attr_obj):
            # Find where the name is assigned in the module source to preserve
            # definition order. Using ^NAME\s*= (multiline) targets assignments
            # at column 0, so "CONFIG_SCHEMA" won't collide with "CONFIG_SCHEMA_BASE".
            match = re.search(
                r"^" + re.escape(m_attr_name) + r"\s*=",
                module_str,
                re.MULTILINE,
            )
            pos = match.start() if match else -1
            schemas.append((pos, m_attr_name, m_attr_obj))

    for _, name, obj in sorted(schemas, key=lambda x: x[0]):
        yield name, obj


found_registries = {}

# Pin validators keys are the functions in pin which validate the pins
pin_validators = {}


def add_pin_validators():
    for m_attr_name in dir(pins):
        if "gpio" in m_attr_name:
            s = pin_validators[repr(getattr(pins, m_attr_name))] = {}
            if "schema" in m_attr_name:
                s["schema"] = True  # else is just number
            if "internal" in m_attr_name:
                s["internal"] = True
            if "input" in m_attr_name:
                s["modes"] = ["input"]
            elif "output" in m_attr_name:
                s["modes"] = ["output"]
            else:
                s["modes"] = []
            if "pullup" in m_attr_name:
                s["modes"].append("pullup")
    from esphome.components.adc import sensor as adc_sensor

    pin_validators[repr(adc_sensor.validate_adc_pin)] = {
        "internal": True,
        "modes": ["input"],
    }


def add_module_registries(domain, module):
    for attr_name in dir(module):
        attr_obj = getattr(module, attr_name)
        if isinstance(attr_obj, Registry):
            if attr_obj == automation.ACTION_REGISTRY:
                reg_type = "action"
                reg_domain = "core"
                found_registries[repr(attr_obj)] = reg_type
            elif attr_obj == automation.CONDITION_REGISTRY:
                reg_type = "condition"
                reg_domain = "core"
                found_registries[repr(attr_obj)] = reg_type
            else:  # attr_name == "FILTER_REGISTRY":
                reg_domain = domain
                reg_type = attr_name.partition("_")[0].lower()
                found_registries[repr(attr_obj)] = f"{domain}.{reg_type}"

            for name in attr_obj:
                if "." not in name:
                    reg_entry_name = name
                else:
                    parts = name.split(".")
                    if len(parts) == 2:
                        reg_domain = parts[0]
                        reg_entry_name = parts[1]
                    elif len(parts) == 3:
                        # is a platform or a component?
                        if parts[0] in schema_core[S_PLATFORMS]:
                            reg_domain = ".".join([parts[1], parts[0]])
                            reg_entry_name = parts[2]
                        elif parts[0] in schema_core[S_COMPONENTS]:
                            reg_domain = parts[0]
                            reg_entry_name = ".".join([parts[1], parts[2]])
                        else:
                            print(f"registry {name} is unknown")

                if reg_domain not in output:
                    output[reg_domain] = {}
                if reg_type not in output[reg_domain]:
                    output[reg_domain][reg_type] = {}
                output[reg_domain][reg_type][reg_entry_name] = convert_config(
                    attr_obj[name].schema, f"{reg_domain}/{reg_type}/{reg_entry_name}"
                )


def do_pins():
    # do pin registries
    pins_providers = schema_core["pins"] = []
    for pin_registry in pins.PIN_SCHEMA_REGISTRY:
        s = convert_config(
            pins.PIN_SCHEMA_REGISTRY[pin_registry][1], f"pins/{pin_registry}"
        )
        if pin_registry not in output:
            output[pin_registry] = {}  # mcp23xxx does not create a component yet
        output[pin_registry]["pin"] = s
        pins_providers.append(pin_registry)


def setBoards(obj, boards):
    obj[S_TYPE] = "enum"
    obj["values"] = {}
    for k, v in boards.items():
        obj["values"][k] = {"docs": v["name"]}


def do_esp32():
    import esphome.components.esp32.boards as esp32_boards

    setBoards(
        output["esp32"]["schemas"]["CONFIG_SCHEMA"]["schema"]["config_vars"]["board"],
        esp32_boards.BOARDS,
    )


def do_esp8266():
    import esphome.components.esp8266.boards as esp8266_boards

    setBoards(
        output["esp8266"]["schemas"]["CONFIG_SCHEMA"]["schema"]["config_vars"]["board"],
        esp8266_boards.BOARDS,
    )


def fix_remote_receiver():
    if "remote_receiver.binary_sensor" not in output:
        return
    remote_receiver_schema = output["remote_receiver.binary_sensor"]["schemas"]
    remote_receiver_schema["CONFIG_SCHEMA"] = {
        "type": "schema",
        "schema": {
            "extends": ["binary_sensor._BINARY_SENSOR_SCHEMA", "core.COMPONENT_SCHEMA"],
            "config_vars": output["remote_base"].pop("binary"),
        },
    }
    remote_receiver_schema["CONFIG_SCHEMA"]["schema"]["config_vars"]["receiver_id"] = {
        "key": "GeneratedID",
        "use_id_type": "remote_base::RemoteReceiverBase",
        "type": "use_id",
    }


def fix_script():
    if "script" not in output:
        return
    output["script"][S_SCHEMAS][S_CONFIG_SCHEMA][S_TYPE] = S_SCHEMA
    config_schema = output["script"][S_SCHEMAS][S_CONFIG_SCHEMA]
    config_schema[S_SCHEMA][S_CONFIG_VARS]["id"]["id_type"] = {
        "class": "script::Script"
    }
    config_schema["is_list"] = True


def fix_font():
    if "font" not in output:
        return
    output["font"][S_SCHEMAS]["FILE_SCHEMA"] = output["font"][S_SCHEMAS].pop(
        "TYPED_FILE_SCHEMA"
    )


def fix_globals():
    if "globals" not in output:
        return
    from esphome.components.globals import _NON_RESTORING_SCHEMA

    config = convert_config(_NON_RESTORING_SCHEMA, "globals/CONFIG_SCHEMA")
    config["is_list"] = True
    output["globals"][S_SCHEMAS][S_CONFIG_SCHEMA] = config


def fix_mapping():
    if "mapping" not in output:
        return
    from esphome.components.mapping import BASE_SCHEMA

    config = convert_config(BASE_SCHEMA, "mapping/CONFIG_SCHEMA")
    output["mapping"][S_SCHEMAS][S_CONFIG_SCHEMA] = config


def fix_menu():
    if "display_menu_base" not in output:
        return
    # # Menu has a recursive schema which is not kept properly
    schemas = output["display_menu_base"][S_SCHEMAS]
    # 1. Move items to a new schema
    schemas["MENU_TYPES"] = {
        S_TYPE: S_SCHEMA,
        S_SCHEMA: {
            S_CONFIG_VARS: {
                "items": schemas["DISPLAY_MENU_BASE_SCHEMA"][S_SCHEMA][S_CONFIG_VARS][
                    "items"
                ]
            }
        },
    }
    # 2. Remove items from the base schema
    schemas["DISPLAY_MENU_BASE_SCHEMA"][S_SCHEMA][S_CONFIG_VARS].pop("items")
    # 3. Add extends to this
    schemas["DISPLAY_MENU_BASE_SCHEMA"][S_SCHEMA][S_EXTENDS].append(
        "display_menu_base.MENU_TYPES"
    )
    # 4. Configure menu items inside as recursive
    menu = schemas["MENU_TYPES"][S_SCHEMA][S_CONFIG_VARS]["items"]["types"]["menu"]
    menu[S_CONFIG_VARS].pop("items")
    menu[S_EXTENDS].append("display_menu_base.MENU_TYPES")


def fix_lvgl_widgets():
    # lvgl's `widgets:` is a recursive tree (a widget can contain widgets). The
    # dumper has no cycle detection, so — like fix_menu — hoist the inlined
    # widget-type enumeration into a named schema and reference it for both the
    # top-level list and each widget's own children, instead of expanding it.
    if "lvgl" not in output:
        return
    schemas = output["lvgl"][S_SCHEMAS]
    config_vars = schemas["CONFIG_SCHEMA"][S_SCHEMA][S_CONFIG_VARS]
    widgets = config_vars.get("widgets")
    if not widgets or S_SCHEMA not in widgets or S_CONFIG_VARS not in widgets[S_SCHEMA]:
        return
    # 1. Hoist the (one-level) widget enumeration into a named schema.
    schemas["WIDGET_TYPES"] = {S_TYPE: S_SCHEMA, S_SCHEMA: widgets[S_SCHEMA]}
    # 2. Reference it from the top-level widgets: list instead of inlining.
    widgets[S_SCHEMA] = {S_EXTENDS: ["lvgl.WIDGET_TYPES"]}
    # 3. Let every widget contain child widgets, via the same named ref.
    for widget in schemas["WIDGET_TYPES"][S_SCHEMA][S_CONFIG_VARS].values():
        if widget.get(S_TYPE) == S_SCHEMA and S_SCHEMA in widget:
            widget[S_SCHEMA].setdefault(S_CONFIG_VARS, {})["widgets"] = {
                S_TYPE: S_SCHEMA,
                "is_list": True,
                "key": "Optional",
                S_SCHEMA: {S_EXTENDS: ["lvgl.WIDGET_TYPES"]},
            }


def get_logger_tags():
    pattern = re.compile(r'^static const char \*const TAG = "(\w.*)";', re.MULTILINE)
    # tags not in components dir
    tags = [
        "app",
        "component",
        "entity_base",
        "scheduler",
        "api.service",
    ]
    for file in CORE_COMPONENTS_PATH.rglob("*.cpp"):
        data = file.read_text(encoding="utf-8")
        match = pattern.search(data)
        if match:
            tags.append(match.group(1))
    return tags


def add_logger_tags():
    if "logger" not in output or "schemas" not in output["logger"]:
        return
    tags = get_logger_tags()
    logs = output["logger"]["schemas"]["CONFIG_SCHEMA"]["schema"]["config_vars"][
        "logs"
    ]["schema"]["config_vars"]
    for t in tags:
        logs[t] = logs["string"].copy()
    logs.pop("string")


def add_referenced_recursive(referenced_schemas, config_var, path, eat_schema=False):
    assert (
        S_CONFIG_VARS not in config_var and S_EXTENDS not in config_var
    )  # S_TYPE in cv or "key" in cv or len(cv) == 0
    if (
        config_var.get(S_TYPE) in ["schema", "trigger", "maybe"]
        and S_SCHEMA in config_var
    ):
        schema = config_var[S_SCHEMA]
        for k, v in schema.get(S_CONFIG_VARS, {}).items():
            if eat_schema:
                new_path = path + [S_CONFIG_VARS, k]
            else:
                new_path = path + ["schema", S_CONFIG_VARS, k]
            add_referenced_recursive(referenced_schemas, v, new_path)
        for k in schema.get(S_EXTENDS, []):
            if k not in referenced_schemas:
                referenced_schemas[k] = [path]
            elif path not in referenced_schemas[k]:
                referenced_schemas[k].append(path)

            s1 = get_str_path_schema(k)
            p = k.split(".")
            if len(p) == 3:
                if path[0] == f"{p[0]}.{p[1]}":
                    # special case for schema inside platforms
                    add_referenced_recursive(
                        referenced_schemas, s1, [path[0], "schemas", p[2]]
                    )
                else:
                    add_referenced_recursive(
                        referenced_schemas, s1, [f"{p[0]}.{p[1]}", "schemas", p[2]]
                    )
            else:
                add_referenced_recursive(
                    referenced_schemas, s1, [p[0], "schemas", p[1]]
                )
    elif config_var.get(S_TYPE) == "typed":
        for tk, tv in config_var.get("types").items():
            add_referenced_recursive(
                referenced_schemas,
                {
                    S_TYPE: S_SCHEMA,
                    S_SCHEMA: tv,
                },
                path + ["types", tk],
                eat_schema=True,
            )


def get_str_path_schema(strPath):
    parts = strPath.split(".")
    if len(parts) > 2:
        parts[0] += "." + parts[1]
        parts[1] = parts[2]
    return output.get(parts[0], {}).get(S_SCHEMAS, {}).get(parts[1], {})


def pop_str_path_schema(strPath):
    parts = strPath.split(".")
    if len(parts) > 2:
        parts[0] += "." + parts[1]
        parts[1] = parts[2]
    output.get(parts[0], {}).get(S_SCHEMAS, {}).pop(parts[1])


def get_arr_path_schema(path):
    s = output
    for x in path:
        s = s[x]
    return s


def merge(source, destination):
    """
    run me with nosetests --with-doctest file.py

    >>> a = { 'first' : { 'all_rows' : { 'pass' : 'dog', 'number' : '1' } } }
    >>> b = { 'first' : { 'all_rows' : { 'fail' : 'cat', 'number' : '5' } } }
    >>> merge(b, a) == { 'first' : { 'all_rows' : { 'pass' : 'dog', 'fail' : 'cat', 'number' : '5' } } }
    True
    """
    for key, value in source.items():
        if isinstance(value, dict):
            # get node or create one
            node = destination.setdefault(key, {})
            merge(value, node)
        else:
            destination[key] = value

    return destination


def is_platform_schema(schema_name):
    # added mostly because of schema_name == "microphone.MICROPHONE_SCHEMA"
    # and "alarm_control_panel"
    # which is shrunk because there is only one component of the schema (i2s_audio)
    component = schema_name.split(".")[0]
    return component in components and components[component].is_platform_component


def shrink():
    """Shrink the extending schemas which has just an end type, e.g. at this point
    ota / port is type schema with extended pointing to core.port, this should instead be
    type number. core.port is number

    This also fixes enums, as they are another schema and they are instead put in the same cv
    """

    # referenced_schemas contains a dict, keys are all that are shown in extends: [] arrays, values are lists of paths that are pointing to that extend
    # e.g. key: core.COMPONENT_SCHEMA has a lot of paths of config vars which extends this schema

    pass_again = True

    while pass_again:
        pass_again = False

        referenced_schemas = {}

        for k, v in output.items():
            for kv, vv in v.items():
                if kv != "pin" and isinstance(vv, dict):
                    for kvv, vvv in vv.items():
                        add_referenced_recursive(referenced_schemas, vvv, [k, kv, kvv])

        for x, paths in referenced_schemas.items():
            if len(paths) == 1 and not is_platform_schema(x):
                key_s = get_str_path_schema(x)
                arr_s = get_arr_path_schema(paths[0])
                # key_s |= arr_s
                # key_s.pop(S_EXTENDS)
                pass_again = True
                if S_SCHEMA in arr_s:
                    if S_EXTENDS in arr_s[S_SCHEMA]:
                        arr_s[S_SCHEMA].pop(S_EXTENDS)
                    else:
                        print("expected extends here!" + x)
                    arr_s = merge(key_s, arr_s)
                    if arr_s[S_TYPE] in ["enum", "typed"]:
                        arr_s.pop(S_SCHEMA)
                else:
                    arr_s.pop(S_EXTENDS)
                    arr_s |= key_s[S_SCHEMA]

    # simple types should be spread on each component,
    # for enums so far these are logger.is_log_level, cover.validate_cover_state and pulse_counter.sensor.COUNT_MODE_SCHEMA
    # then for some reasons sensor filter registry falls here
    # then are all simple types, integer and strings
    for x, paths in referenced_schemas.items():
        key_s = get_str_path_schema(x)
        # Spread scalar leaf schemas (a single ``type`` with no nested schema or
        # config vars) onto each referencing field so the type is inline. This
        # covers enum/registry/integer/string plus float_with_unit quantities
        # (e.g. a ``core.frequency`` schema typed ``frequency``), time and
        # lambda -- but never structural schemas, which stay as references.
        key_type = key_s.get(S_TYPE) if key_s else None
        if (
            key_type is not None
            and key_type not in ("schema", "typed", "trigger", "pin", "use_id")
            and S_SCHEMA not in key_s
            and S_CONFIG_VARS not in key_s
        ):
            if key_type == "registry":
                print("Spreading registry: " + x)
            for target in paths:
                target_s = get_arr_path_schema(target)
                if S_SCHEMA not in target_s:
                    print("skipping simple spread for " + ".".join(target))
                    continue
                extends = target_s[S_SCHEMA][S_EXTENDS]
                if x not in extends:
                    # Already handled on an earlier visit (a field can list the
                    # same schema reference more than once).
                    continue
                if len(extends) > 1:
                    # The field references several schemas at once (e.g. a value
                    # that extends both hex_uint8_t and uint8_t). Drop this
                    # reference and let the remaining one(s) describe the type,
                    # rather than forcing a single-extends spread here.
                    extends.remove(x)
                    continue
                assert extends == [x]
                target_s.pop(S_SCHEMA)
                target_s |= key_s
                if key_s[S_TYPE] in ["integer", "string"]:
                    target_s["data_type"] = x.split(".")[1]
            # remove this dangling again
            pop_str_path_schema(x)
        elif not key_s or set(key_s) <= {"min", "max"}:
            # An untyped named schema, or one carrying only range bounds (e.g.
            # positive_float = All(float_, Range(min=0)) has no scalar type but a
            # min). Spread its data_type name and any bounds onto each field.
            for target in paths:
                target_s = get_arr_path_schema(target)
                if S_SCHEMA not in target_s:
                    # an empty schema like speaker.SPEAKER_SCHEMA
                    target_s[S_EXTENDS].remove(x)
                    continue
                assert x in target_s[S_SCHEMA][S_EXTENDS]
                target_s.pop(S_SCHEMA)
                target_s.pop(S_TYPE)  # undefined
                target_s["data_type"] = x.split(".")[1]
                target_s.update(key_s)  # carry min/max bounds, if any
            # remove this dangling again
            pop_str_path_schema(x)

    # remove unreachable schemas
    for domain, domain_schemas in output.items():
        for schema_name in list(domain_schemas.get(S_SCHEMAS, {}).keys()):
            s = f"{domain}.{schema_name}"
            if (
                not s.endswith("." + S_CONFIG_SCHEMA)
                and s not in referenced_schemas
                and not is_platform_schema(s)
            ):
                domain_schemas[S_SCHEMAS].pop(schema_name)


def is_cv_invalid(schema):
    return repr(schema).startswith("<function invalid.<locals>.validator")


def build_schema():
    print("Building schema")

    # check esphome was not loaded globally (IDE auto imports)
    if len(ejs.extended_schemas) == 0:
        raise LookupError(
            "no data collected. Did you globally import an ESPHome component?"
        )

    # Core schema
    schema_core[S_SCHEMAS] = {}
    register_module_schemas("core", cv)

    platforms = {}
    schema_core[S_PLATFORMS] = platforms
    core_components = {}
    schema_core[S_COMPONENTS] = core_components

    add_pin_validators()

    # Load a preview of each component
    for domain, manifest in components.items():
        if manifest.is_platform_component:
            # e.g. sensor, binary sensor, add S_COMPONENTS
            # note: S_COMPONENTS is not filled until loaded, e.g.
            # if lock: is not used, then we don't need to know about their
            # platforms yet.
            output[domain] = {S_COMPONENTS: {}, S_SCHEMAS: {}}
            platforms[domain] = {}
        elif manifest.config_schema is not None:
            if is_cv_invalid(manifest.config_schema):
                continue
            output[domain] = {S_SCHEMAS: {S_CONFIG_SCHEMA: {}}}

    # Generate platforms (e.g. sensor, binary_sensor, climate )
    for domain in platforms:
        c = components[domain]
        register_module_schemas(domain, c.module)

    # Generate components
    for domain, manifest in components.items():
        if domain not in platforms:
            if manifest.config_schema is not None and not is_cv_invalid(
                manifest.config_schema
            ):
                core_components[domain] = {}
                if len(manifest.dependencies) > 0:
                    core_components[domain]["dependencies"] = manifest.dependencies
            register_module_schemas(domain, manifest.module, manifest)

        for platform in platforms:
            platform_manifest = get_platform(domain=platform, platform=domain)
            if platform_manifest is not None:
                count = register_module_schemas(
                    f"{domain}.{platform}", platform_manifest.module, platform_manifest
                )
                if count > 0:
                    output[platform][S_COMPONENTS].setdefault(domain, {})
                    if len(platform_manifest.dependencies) > 0:
                        output[platform][S_COMPONENTS][domain]["dependencies"] = (
                            platform_manifest.dependencies
                        )

    # Do registries
    add_module_registries("core", automation)
    for domain, manifest in components.items():
        add_module_registries(domain, manifest.module)
    add_module_registries("remote_base", remote_base)

    # update props pointing to registries
    for reg_config_var in solve_registry:
        (registry, config_var) = reg_config_var
        config_var[S_TYPE] = "registry"
        config_var["registry"] = found_registries[repr(registry)]

    do_pins()
    do_esp8266()
    do_esp32()
    fix_remote_receiver()
    fix_script()
    fix_font()
    fix_globals()
    fix_mapping()
    add_logger_tags()
    shrink()
    fix_menu()
    fix_lvgl_widgets()

    # aggregate components, so all component info is in same file, otherwise we have dallas.json, dallas.sensor.json, etc.
    data = {}
    for component, component_schemas in output.items():
        if "." in component:
            key = component.partition(".")[0]
            if key not in data:
                data[key] = {}
            data[key][component] = component_schemas
        else:
            if component not in data:
                data[component] = {}
            data[component] |= {component: component_schemas}

    # bundle core inside esphome
    data["esphome"]["core"] = data.pop("core")["core"]

    # Surface deprecated component aliases (declared via ``ALIASES = [...]``
    # on the canonical component) so language servers / dashboard
    # autocomplete still accept legacy top-level keys instead of flagging
    # them as unknown. Each alias gets its own bundle that mirrors the
    # canonical schema; ``alias_of`` and the optional ``removal_version``
    # metadata let consumers render a deprecation hint and point users at
    # the canonical name. Without this, configs migrated only at runtime
    # (via the ``_resolve_component_aliases`` pre-pass) would still light
    # up as errors in the editor.
    for domain, manifest in components.items():
        aliases = manifest.aliases
        if not aliases or domain not in data:
            continue
        canonical_bundle = data[domain].get(domain)
        if canonical_bundle is None:
            continue
        for alias in aliases:
            alias_entry = dict(canonical_bundle)
            alias_entry["alias_of"] = domain
            if manifest.alias_removal_version is not None:
                alias_entry["removal_version"] = manifest.alias_removal_version
            data[alias] = {alias: alias_entry}

    if GENERATED_ID_TYPES:
        print(
            "Unconsumed id_type matchers:",
            [id_type for _, id_type in GENERATED_ID_TYPES],
        )

    if args.check:  # do not gen files
        return

    for c, s in data.items():
        write_file(c, s)
    deleted = delete_extra_files(data.keys())
    print(f"Written {len(data.items())} deleted {deleted} files.")


def is_convertible_schema(schema):
    if schema is None:
        return False
    if isinstance(schema, (cv.Schema, cv.All, cv.Any)):
        return True
    if repr(schema) in ejs.hidden_schemas:
        return True
    if repr(schema) in ejs.typed_schemas:
        return True
    if repr(schema) in ejs.list_schemas:
        return True
    if repr(schema) in ejs.registry_schemas:
        return True
    if isinstance(schema, dict):
        for k in schema:
            if isinstance(k, (cv.Required, cv.Optional)):
                return True
    return False


def convert_config(schema, path):
    converted = {}
    convert(schema, converted, path)
    return converted


GENERATED_ID_TYPES = [
    (
        lambda p: p.startswith("i2c/CONFIG_SCHEMA/") and p.endswith("/id"),
        {"class": "i2c::I2CBus", "parents": ["Component"]},
    ),
    (
        lambda p: p == "uart/CONFIG_SCHEMA/val 1/ext0/all/id",
        {"class": "uart::UARTComponent", "parents": ["Component"]},
    ),
    (
        lambda p: p == "http_request/CONFIG_SCHEMA/val 1/ext0/all/id",
        {"class": "http_request::HttpRequestComponent", "parents": ["Component"]},
    ),
    (
        lambda p: (
            p
            == "uptime.sensor/CONFIG_SCHEMA/type_timestamp/ext0/ext1/all/time_id/val 1"
        ),
        {},
    ),
    (lambda p: p == "esp_ldo/action/voltage.adjust/all/all/id", {}),
]


def convert(schema, config_var, path):
    """config_var can be a config_var or a schema: both are dicts
    config_var has a S_TYPE property, if this is S_SCHEMA, then it has a S_SCHEMA property
    schema does not have a type property, schema can have optionally both S_CONFIG_VARS and S_EXTENDS
    """
    repr_schema = repr(schema)

    if repr_schema in known_schemas:
        schema_info = known_schemas[(repr_schema)]
        for schema_instance, name in schema_info:
            if schema_instance is schema:
                assert S_CONFIG_VARS not in config_var
                assert S_EXTENDS not in config_var
                if S_TYPE not in config_var:
                    config_var[S_TYPE] = S_SCHEMA
                # assert config_var[S_TYPE] == S_SCHEMA

                if S_SCHEMA not in config_var:
                    config_var[S_SCHEMA] = {}
                if S_EXTENDS not in config_var[S_SCHEMA]:
                    config_var[S_SCHEMA][S_EXTENDS] = [name]
                elif name not in config_var[S_SCHEMA][S_EXTENDS]:
                    config_var[S_SCHEMA][S_EXTENDS].append(name)
                return

    # Extended schemas are tracked when the .extend() is used in a schema
    if repr_schema in ejs.extended_schemas:
        extended = ejs.extended_schemas.get(repr_schema)
        for idx, ext in enumerate(extended):
            convert(ext, config_var, f"{path}/ext{idx}")
        return

    if isinstance(schema, cv.SensitiveValidator):
        config_var["sensitive"] = True
        config_var["sensitive_source"] = "explicit"
        if isinstance(schema, cv.BindKeyValidator):
            # Its inner is the bound ``_validate`` method (a hex-key string);
            # walking it yields no type, so describe it directly.
            config_var[S_TYPE] = "string"
        else:
            convert(schema.inner, config_var, f"{path}/sensitive")
        return

    if isinstance(schema, cv.All):
        i = 0
        for inner in schema.validators:
            i = i + 1
            convert(inner, config_var, path + f"/val {i}")
        return

    if hasattr(schema, "validators"):
        i = 0
        for inner in schema.validators:
            i = i + 1
            convert(inner, config_var, path + f"/val {i}")

    if isinstance(schema, cv.Schema):
        convert(schema.schema, config_var, path + "/all")
        return

    if isinstance(schema, dict):
        convert_keys(config_var, schema, path)
        return

    if repr_schema in ejs.list_schemas:
        config_var["is_list"] = True
        items_schema = ejs.list_schemas[repr_schema][0]
        convert(items_schema, config_var, path + "/list")
        return

    if DUMP_RAW:
        config_var["raw"] = repr_schema

    # A numeric range constraint (from cv.int_range / cv.float_range / a bare
    # vol.Range in an All) contributes bounds, not a type. Attach them at the
    # config var level, next to ``type``, so editors can validate the range.
    if isinstance(schema, vol.Range):
        # min/max may be non-numeric (e.g. a TimePeriod for a time-period range);
        # keep numbers as-is and stringify anything else so the dump stays JSON
        # serializable.
        if schema.min is not None:
            config_var["min"] = (
                schema.min if isinstance(schema.min, (int, float)) else str(schema.min)
            )
        if schema.max is not None:
            config_var["max"] = (
                schema.max if isinstance(schema.max, (int, float)) else str(schema.max)
            )
        return

    # pylint: disable=comparison-with-callable
    if schema == cv.boolean:
        config_var[S_TYPE] = "boolean"
    elif schema == automation.validate_potentially_and_condition:
        config_var[S_TYPE] = "registry"
        config_var["registry"] = "condition"
    elif schema in (cv.int_, cv.int_range) or schema in _CV_INTEGER_VALIDATORS:
        config_var[S_TYPE] = "integer"
    elif schema in (cv.string, cv.string_strict, cv.valid_name) or (
        schema in _CV_STRING_VALIDATORS
    ):
        config_var[S_TYPE] = "string"
    elif schema in _CV_FLOAT_VALIDATORS:
        config_var[S_TYPE] = "float"
    elif schema in _CV_TIME_VALIDATORS:
        config_var[S_TYPE] = "time"
    elif schema in _CV_LAMBDA_VALIDATORS:
        config_var[S_TYPE] = "lambda"
    elif schema == cv.entity_category:
        config_var[S_TYPE] = "enum"
        config_var["values"] = dict.fromkeys(cv.ENTITY_CATEGORIES)

    elif isinstance(schema, vol.Schema):
        # test: esphome/project
        config_var[S_TYPE] = "schema"
        config_var["schema"] = convert_config(schema.schema, path + "/s")["schema"]

    elif repr_schema in pin_validators:
        config_var |= pin_validators[repr_schema]
        config_var[S_TYPE] = "pin"

    # pylint: disable-next=too-many-nested-blocks
    elif repr_schema in ejs.hidden_schemas:
        schema_type = ejs.hidden_schemas[repr_schema]

        data = schema(ejs.SCHEMA_EXTRACT)

        # enums, e.g. esp32/variant
        if schema_type == "one_of":
            config_var[S_TYPE] = "enum"
            config_var["values"] = dict.fromkeys(list(data))
        elif schema_type == "enum":
            config_var[S_TYPE] = "enum"
            config_var["values"] = dict.fromkeys(list(data.keys()))
        elif schema_type == "variant_enum":
            # Per-variant enum (e.g. psram mode/speed): each value carries the
            # list of variants that accept it so clients can filter to the
            # user's selected variant. Additive to the plain enum format —
            # consumers that ignore the metadata still see every option.
            config_var[S_TYPE] = "enum"
            config_var["values"] = {
                value: {"variants": variants} for value, variants in data.items()
            }
        elif schema_type == "maybe":
            # maybe_simple_value: either a scalar shorthand (mapped to the key in
            # data[1]) or the full wrapped schema. The wrapped schema is usually a
            # plain Schema (converts to a "schema" config var), but may be something
            # else, e.g. a typed_schema (converts to a "typed" config var with
            # "types" and no top-level "schema" key). Merge whatever it produced
            # rather than assuming a "schema" key is present.
            config_var["maybe"] = data[1]
            config_var.update(convert_config(data[0], path + "/maybe"))
        # esphome/on_boot
        elif schema_type == "automation":
            extra_schema = None
            config_var[S_TYPE] = "trigger"
            if ejs.extended_schemas[repr(data)][0] == automation.AUTOMATION_SCHEMA:
                extra_schema = ejs.extended_schemas[repr(data)][1]
            if (
                extra_schema is not None and len(extra_schema) > 1
            ):  # usually only trigger_id here
                config = convert_config(extra_schema, path + "/extra")
                if "schema" in config:
                    automation_schema = config["schema"]
                    if not (
                        len(automation_schema["config_vars"]) == 1
                        and "trigger_id" in automation_schema["config_vars"]
                    ):
                        automation_schema["config_vars"]["then"] = {S_TYPE: "trigger"}
                        if "trigger_id" in automation_schema["config_vars"]:
                            automation_schema["config_vars"].pop("trigger_id")

                        config_var[S_TYPE] = "trigger"
                        config_var["schema"] = automation_schema
                        # some triggers can have a list of actions directly, while others needs to have some other configuration,
                        # e.g. sensor.on_value_rang, and the list of actions is only accepted under "then" property.
                        try:
                            schema({"delay": "1s"})
                        except cv.Invalid:
                            config_var["has_required_var"] = True
        elif schema_type == "effects":
            config_var[S_TYPE] = "registry"
            config_var["registry"] = "light.effects"
            config_var["filter"] = data[0]
        elif schema_type == "float":
            config_var[S_TYPE] = "float"
            # cv.float_with_unit reports its canonical unit (e.g. "Hz", "V") for
            # SCHEMA_EXTRACT; surface it next to the type so editors can label
            # the field. Other float sources return None (no unit).
            if isinstance(data, str):
                config_var["unit"] = data
        elif schema_type in ("string", "integer", "time", "lambda"):
            # Scalar validators (e.g. cv.date_time) that declare their result
            # type via schema_extractor. ``data`` is unused (the decorated
            # validator returns None for SCHEMA_EXTRACT).
            config_var[S_TYPE] = schema_type
        elif schema_type == "templatable":
            config_var["templatable"] = True
            convert(data, config_var, path + "/templat")
        elif schema_type == "triggers":
            # remote base
            convert(data, config_var, path + "/trigger")
        elif schema_type == "sensor":
            schema = data
            convert(data, config_var, path + "/trigger")
        elif schema_type == "declare_id":
            # pylint: disable=protected-access
            parents = data._parents

            config_var["id_type"] = {
                "class": str(data.base),
                "parents": (
                    [str(x.base) for x in parents]
                    if isinstance(parents, list)
                    else None
                ),
            }
        elif schema_type == "use_id":
            if inspect.ismodule(data):
                m_attr_obj = data.CONFIG_SCHEMA
                use_schema = known_schemas.get(repr(m_attr_obj))
                if use_schema:
                    [output_module, output_name] = use_schema[0][1].split(".")
                    use_id_config = output[output_module][S_SCHEMAS][output_name]
                    config_var["use_id_type"] = use_id_config["schema"]["config_vars"][
                        "id"
                    ]["id_type"]["class"]
                    config_var[S_TYPE] = "use_id"
            elif isinstance(data, str):
                # TODO: Figure out why pipsolar does this
                config_var["use_id_type"] = data
            else:
                config_var["use_id_type"] = str(data.base)
                config_var[S_TYPE] = "use_id"
        elif schema_type == "schema":
            # A callable CONFIG_SCHEMA that returned a representative schema
            # for extraction (model-driven components); walk it as usual.
            convert(data, config_var, path)
        else:
            raise TypeError("Unknown extracted schema type")
    elif config_var.get("key") == "GeneratedID":
        for i, (matcher, id_type) in enumerate(GENERATED_ID_TYPES):
            if matcher(path):
                config_var["id_type"] = id_type
                GENERATED_ID_TYPES.pop(i)
                break
        else:
            print("Cannot determine id_type for " + path)

            # raise TypeError("Cannot determine id_type for " + path)

    elif repr_schema in ejs.registry_schemas:
        solve_registry.append((ejs.registry_schemas[repr_schema], config_var))

    elif repr_schema in ejs.typed_schemas:
        config_var[S_TYPE] = "typed"
        types = config_var["types"] = {}
        typed_schema = ejs.typed_schemas[repr_schema]
        if len(typed_schema) > 1:
            config_var["typed_key"] = typed_schema[1].get("key", CONF_TYPE)
        for schema_key, schema_type in typed_schema[0][0].items():
            config = convert_config(schema_type, path + "/type_" + schema_key)
            types[schema_key] = config["schema"]

    elif DUMP_UNKNOWN and S_TYPE not in config_var:
        config_var["unknown"] = repr_schema

    if DUMP_PATH:
        config_var["path"] = path
    if S_TYPE not in config_var:
        pass
        # print(path)


def get_overridden_config(key, converted):
    # check if the key is in any extended schema in this converted schema, i.e.
    # if we see a on_value_range in a dallas sensor, then this is overridden because
    # it is already defined in sensor
    assert S_CONFIG_VARS not in converted and S_EXTENDS not in converted
    config = converted.get(S_SCHEMA, {})

    return get_overridden_key_inner(key, config, {})


def get_overridden_key_inner(key, config, ret):
    if S_EXTENDS not in config:
        return ret
    for s in config[S_EXTENDS]:
        p = s.partition(".")
        s1 = output.get(p[0], {}).get(S_SCHEMAS, {}).get(p[2], {}).get(S_SCHEMA)
        if s1:
            if key in s1.get(S_CONFIG_VARS, {}):
                for k, v in s1.get(S_CONFIG_VARS)[key].items():
                    if k not in ret:  # keep most overridden
                        ret[k] = v
            get_overridden_key_inner(key, s1, ret)

    return ret


def convert_keys(converted, schema, path):
    for k, v in schema.items():
        # deprecated stuff
        if repr(v).startswith("<function invalid"):
            continue

        result = {}

        if isinstance(k, cv.GenerateID):
            result["key"] = "GeneratedID"
        elif isinstance(k, cv.Required):
            result["key"] = "Required"
        elif isinstance(k, (cv.Optional, cv.Inclusive, cv.Exclusive)):
            result["key"] = "Optional"
        else:
            converted["key"] = "String"
            key_string_match = re.search(
                r"<function (\w*) at \w*>", str(k), re.IGNORECASE
            )
            if key_string_match:
                converted["key_type"] = key_string_match.group(1)
            else:
                converted["key_type"] = str(k)

        # ``cv.OnlyWith`` / ``cv.OnlyWithout`` expose ``default`` as
        # a property that returns ``vol.UNDEFINED`` when the gating
        # component isn't loaded — and at schema-generation time
        # ``CORE.loaded_integrations`` is always empty, so the
        # property never resolves. The unconditional default lives
        # on ``_default``; expose it under a *new* per-class field
        # (``default_with`` for ``OnlyWith``, ``default_without`` for
        # ``OnlyWithout``) that bundles the value with the gating
        # component(s). Pure addition to the bundle — old consumers
        # that read only ``default`` see these fields as
        # default-less (same as today, no regression where they used
        # to fall back to a hard-coded UI default); new consumers
        # opt-in to the gated fields and apply the default
        # *conditionally* on which integrations the user has
        # loaded. Without the gate info, an ethernet-only config on
        # ``cv.OnlyWith(K, "wifi", default=True)`` would otherwise
        # render ``True`` even though ESPHome itself wouldn't apply
        # the default for that config.
        if isinstance(k, (cv.OnlyWith, cv.OnlyWithout)):
            default_value = k._default()
            if default_value is not None:
                components = (
                    list(k._component)
                    if isinstance(k._component, list)
                    else [k._component]
                )
                gate_field = (
                    "default_with" if isinstance(k, cv.OnlyWith) else "default_without"
                )
                result[gate_field] = {
                    "value": str(default_value),
                    "components": components,
                }
        elif hasattr(k, "default") and str(k.default) != "...":
            default_value = k.default()
            if default_value is not None:
                result["default"] = str(default_value)

        # UI hint from ``cv.Optional`` / ``cv.Required`` — surfaced
        # for schema consumers (visual editors) that want to render
        # advanced / yaml-only fields differently. ESPHome itself
        # ignores it at runtime; emitting only when set keeps the
        # dump compact and backwards-compatible with markers that
        # don't carry the attribute. The value is the str form of
        # ``cv.Visibility`` (e.g. ``"advanced"`` / ``"yaml_only"``)
        # so consumers don't need an enum import to read it.
        visibility = getattr(k, "visibility", None)
        if visibility is not None:
            result["visibility"] = str(visibility)

        # Do value
        convert(v, result, path + f"/{str(k)}")

        # Heuristic fallback when the field's validator wasn't explicitly
        # wrapped in ``cv.sensitive``. Only applies to string-typed leaves so
        # we don't mark unrelated nested schemas. ``sensitive_source`` lets
        # consumers distinguish explicit markers from heuristic matches. Pull
        # the field name from ``k.schema`` (voluptuous's stored key) rather
        # than ``str(k)`` so we don't depend on the marker's ``__str__``
        # representation.
        if (
            "sensitive" not in result
            and result.get(S_TYPE) == "string"
            and isinstance(k, (cv.Required, cv.Optional, cv.Inclusive, cv.Exclusive))
            and isinstance(k.schema, str)
        ):
            key_lower = k.schema.lower()
            if any(frag in key_lower for frag in cv.SENSITIVE_KEY_FRAGMENTS):
                result["sensitive"] = True
                result["sensitive_source"] = "heuristic"

        if "schema" not in converted:
            converted[S_TYPE] = "schema"
            converted["schema"] = {S_CONFIG_VARS: {}}
        if S_CONFIG_VARS not in converted["schema"]:
            converted["schema"][S_CONFIG_VARS] = {}
        for base_k, base_v in get_overridden_config(k, converted).items():
            if base_k in result and base_v == result[base_k]:
                result.pop(base_k)
        converted["schema"][S_CONFIG_VARS][str(k)] = result
        if "key" in converted and converted["key"] == "String":
            config_vars = converted["schema"]["config_vars"]
            assert len(config_vars) == 1
            key = list(config_vars.keys())[0]
            assert key.startswith("<")
            config_vars["string"] = config_vars.pop(key)


def main() -> None:
    parser.parse_args(namespace=args)
    build_schema()


if __name__ == "__main__":
    main()