100 lines
3.2 KiB
Python
100 lines
3.2 KiB
Python
|
#!/usr/bin/env python3
|
||
|
|
"""Split HA list/dict config files into one-object-per-file layouts.
|
||
|
|
|
||
|
|
See services/home-assistant/DESIGN.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))
|