numos/export_state.py

back to source

"""Build numos.state from the ops registry plus seed data.

Units are derived from the ops registry rather than authored, so the unit set
changes when the registry changes. The content hash is what makes that drift
detectable instead of silent.
"""

import json
import os

from numos import seed
from numos.state import render
from numos.validate import validate

OPS_URL = "http://127.0.0.1:8080/api/ops"


def unit_name_for_op(op_name):
    """op_SwarmExecutor_v1 style name -> stable unit name."""
    return "op-" + op_name.lower().replace("_", "-")


def build_state(ops):
    """Assemble a validated state dict from op names plus seed data."""
    units = list(seed.INFRA_UNITS)
    for op_name in sorted(set(ops)):
        units.append({
            "name": unit_name_for_op(op_name),
            "kind": "longrun",
            "restart": "on-failure",
            "backoff_ms": 250,
            "backoff_max_ms": 30000,
            "arch_mask": [],
            "health_probe": None,
            "requires": ["join"],
            "after": [],
            "exec": "numctl run-op %s" % op_name,
        })
    state = {
        "version": 1,
        "arch_targets": seed.ARCH_TARGETS,
        "boot_phases": seed.BOOT_PHASES,
        "units": sorted(units, key=lambda u: u["name"]),
        "health": [],
    }
    validate(state)
    return state


def fetch_ops(url=OPS_URL):
    """Read op names from a live IntrikataTopology instance."""
    import urllib.request
    with urllib.request.urlopen(url, timeout=10) as response:
        payload = json.loads(response.read().decode("utf-8"))
    ops = payload.get("ops", payload) if isinstance(payload, dict) else payload
    return [op["name"] if isinstance(op, dict) else op for op in ops]


def export(ops, out_dir):
    """Write numos.state and its JSON web view. Returns the state path."""
    state = build_state(ops)
    if not os.path.isdir(out_dir):
        os.makedirs(out_dir)

    state_path = os.path.join(out_dir, "numos.state")
    with open(state_path, "w", newline="\n") as handle:
        handle.write(render(state))

    json_path = os.path.join(out_dir, "numos.state.json")
    with open(json_path, "w", newline="\n") as handle:
        json.dump(state, handle, sort_keys=True, indent=2)
        handle.write("\n")

    return state_path


def main():
    import argparse
    parser = argparse.ArgumentParser(description="Export numos.state")
    parser.add_argument("--ops-url", default=OPS_URL)
    parser.add_argument("--out", default="dist")
    args = parser.parse_args()
    path = export(fetch_ops(args.ops_url), args.out)
    print("wrote %s" % path)


if __name__ == "__main__":
    main()