homelab-codex-ws/scripts/ha/lib/split.py

#!/usr/bin/env python3
"""Split HA list/dict config files into one-object-per-file layouts.

See kb/decisions/ha-configs-as-code.md ("Split + normalization"):
  automations.yaml -> automations/<id>.yaml
  scenes.yaml       -> scenes/<id>.yaml
  scripts.yaml      -> scripts/<key>.yaml
"""
import os
import sys

import yaml

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from normalize import dump_canonical_yaml  # noqa: E402


def split_list_by_id(items, id_field="id"):
    """items: list of dicts, each expected to carry an `id_field` key.

    Returns {id: obj}. Raises ValueError on a missing/duplicate id so a bad
    split can never silently drop or overwrite an object.
    """
    result = {}
    for obj in items:
        if not isinstance(obj, dict) or id_field not in obj:
            raise ValueError(f"object missing '{id_field}' field: {obj!r}")
        key = str(obj[id_field])
        if key in result:
            raise ValueError(f"duplicate {id_field} '{key}'")
        result[key] = obj
    return result


def split_dict(data):
    """scripts.yaml is already a dict keyed by script id — pass through."""
    return dict(data or {})


SPLIT_SPECS = {
    "automations.yaml": ("automations", lambda data: split_list_by_id(data or [], "id")),
    "scenes.yaml": ("scenes", lambda data: split_list_by_id(data or [], "id")),
    "scripts.yaml": ("scripts", split_dict),
}


def write_split(objects_by_key, out_dir):
    """Write out_dir/<key>.yaml (canonical form) for each key/obj pair.

    Removes stale *.yaml files left over from a previous run for keys that
    no longer exist, so the split output always mirrors the source exactly.
    """
    os.makedirs(out_dir, exist_ok=True)
    expected = {f"{key}.yaml" for key in objects_by_key}
    for existing in os.listdir(out_dir):
        if existing.endswith(".yaml") and existing not in expected:
            os.remove(os.path.join(out_dir, existing))
    for key, obj in objects_by_key.items():
        path = os.path.join(out_dir, f"{key}.yaml")
        with open(path, "w", encoding="utf-8") as f:
            f.write(dump_canonical_yaml(obj))


def split_file(source_path, out_root):
    """source_path: path to e.g. .../automations.yaml. out_root: config/<instance>/.

    Returns the output subdirectory written, or None if source_path's
    basename isn't one of the known split targets.
    """
    basename = os.path.basename(source_path)
    if basename not in SPLIT_SPECS:
        return None
    subdir_name, splitter = SPLIT_SPECS[basename]
    with open(source_path, "r", encoding="utf-8") as f:
        data = yaml.safe_load(f)
    objects_by_key = splitter(data)
    out_dir = os.path.join(out_root, subdir_name)
    write_split(objects_by_key, out_dir)
    return out_dir


def main(argv):
    if len(argv) != 3:
        print("usage: split.py <source-yaml-path> <config-out-root>", file=sys.stderr)
        return 2
    source_path, out_root = argv[1], argv[2]
    result_dir = split_file(source_path, out_root)
    if result_dir is None:
        print(
            f"split.py: '{os.path.basename(source_path)}' is not a known split target",
            file=sys.stderr,
        )
        return 1
    print(result_dir)
    return 0


if __name__ == "__main__":
    raise SystemExit(main(sys.argv))