core/script/hassfest/zeroconf.py

"""Generate zeroconf file."""
from __future__ import annotations

from collections import OrderedDict, defaultdict
import json

from .model import Config, Integration

BASE = """
\"\"\"Automatically generated by hassfest.

To update, run python3 -m script.hassfest
\"\"\"

# fmt: off

ZEROCONF = {}

HOMEKIT = {}
""".strip()


def generate_and_validate(integrations: dict[str, Integration]):
    """Validate and generate zeroconf data."""
    service_type_dict = defaultdict(list)
    homekit_dict = {}

    for domain in sorted(integrations):
        integration = integrations[domain]

        if not integration.manifest:
            continue

        service_types = integration.manifest.get("zeroconf", [])
        homekit = integration.manifest.get("homekit", {})
        homekit_models = homekit.get("models", [])

        if not (service_types or homekit_models):
            continue

        for entry in service_types:
            data = {"domain": domain}
            if isinstance(entry, dict):
                typ = entry["type"]
                entry_without_type = entry.copy()
                del entry_without_type["type"]
                data.update(entry_without_type)
            else:
                typ = entry

            service_type_dict[typ].append(data)

        for model in homekit_models:
            if model in homekit_dict:
                integration.add_error(
                    "zeroconf",
                    f"Integrations {domain} and {homekit_dict[model]} "
                    "have overlapping HomeKit models",
                )
                break

            homekit_dict[model] = domain

    # HomeKit models are matched on starting string, make sure none overlap.
    warned = set()
    for key in homekit_dict:
        if key in warned:
            continue

        # n^2 yoooo
        for key_2 in homekit_dict:
            if key == key_2 or key_2 in warned:
                continue

            if key.startswith(key_2) or key_2.startswith(key):
                integration.add_error(
                    "zeroconf",
                    f"Integrations {homekit_dict[key]} and {homekit_dict[key_2]} "
                    "have overlapping HomeKit models",
                )
                warned.add(key)
                warned.add(key_2)
                break

    zeroconf = OrderedDict(
        (key, service_type_dict[key]) for key in sorted(service_type_dict)
    )
    homekit = OrderedDict((key, homekit_dict[key]) for key in sorted(homekit_dict))

    return BASE.format(json.dumps(zeroconf, indent=4), json.dumps(homekit, indent=4))


def validate(integrations: dict[str, Integration], config: Config):
    """Validate zeroconf file."""
    zeroconf_path = config.root / "homeassistant/generated/zeroconf.py"
    config.cache["zeroconf"] = content = generate_and_validate(integrations)

    if config.specific_integrations:
        return

    with open(str(zeroconf_path)) as fp:
        current = fp.read().strip()
        if current != content:
            config.add_error(
                "zeroconf",
                "File zeroconf.py is not up to date. Run python3 -m script.hassfest",
                fixable=True,
            )
        return


def generate(integrations: dict[str, Integration], config: Config):
    """Generate zeroconf file."""
    zeroconf_path = config.root / "homeassistant/generated/zeroconf.py"
    with open(str(zeroconf_path), "w") as fp:
        fp.write(f"{config.cache['zeroconf']}\n")