agent/topology.py

from typing import List, Dict, Tuple
import json
import os

TOPOLOGY_PATH = os.getenv(
    "TOPOLOGY_PATH",
    os.path.join(os.path.dirname(__file__), "..", "infra", "topology.json"),
)


def _load_topology() -> Dict:
    try:
        with open(TOPOLOGY_PATH, "r") as f:
            return json.load(f)
    except Exception:
        # If anything goes wrong, we degrade gracefully.
        return {"devices": [], "links": []}


def _index_devices(topology: Dict) -> Dict[str, Dict]:
    return {d["id"]: d for d in topology.get("devices", [])}


def list_devices() -> list:
    """
    Returns device IDs from topology.json to drive dropdowns or validation.
    """
    topo = _load_topology()
    return [d.get("id") for d in topo.get("devices", []) if d.get("id")]


def summarize_topology_impact(steps: List[Dict]) -> str:
    """
    Summarize which devices/roles/pods are in scope for the requested steps.
    """
    if not steps:
        return "_No steps to analyze._"

    topo = _load_topology()
    devices_index = _index_devices(topo)

    touched_devices = {step["json"].get("device", "device") for step in steps}
    found = []
    missing = []

    pods = set()
    roles = set()

    for dev in touched_devices:
        info = devices_index.get(dev)
        if info is None:
            missing.append(dev)
        else:
            found.append(dev)
            if "pod" in info:
                pods.add(info["pod"])
            if "role" in info:
                roles.add(info["role"])

    lines = []

    if found:
        lines.append(f"- Devices with topology metadata: {', '.join(sorted(found))}")
    if missing:
        lines.append(f"- Devices missing from topology.json: {', '.join(sorted(missing))}")

    if pods:
        lines.append(f"- Pods touched: {', '.join(sorted(pods))}")
    if roles:
        lines.append(f"- Roles involved: {', '.join(sorted(roles))}")

    if not lines:
        return "_No useful topology information available._"

    return "\n".join(lines)