scripts/compose_task_profile.py

"""Compose task profiles from Markdown capability cards."""

from __future__ import annotations

import argparse
import json
from pathlib import Path
from typing import Any


REGISTRY_ROOT = Path(__file__).resolve().parents[1] / "docs" / "registry"


def parse_scalar(value: str) -> Any:
    value = value.strip()
    if value in {"true", "True"}:
        return True
    if value in {"false", "False"}:
        return False
    if value in {"null", "None"}:
        return None
    if value.startswith("[") and value.endswith("]"):
        inner = value[1:-1].strip()
        if not inner:
            return []
        return [parse_scalar(part.strip()) for part in inner.split(",")]
    try:
        if "." in value:
            return float(value)
        return int(value)
    except ValueError:
        return value.strip("\"'")


def parse_front_matter(text: str) -> tuple[dict[str, Any], str]:
    lines = text.splitlines()
    if not lines or lines[0].strip() != "---":
        return {}, text
    meta: dict[str, Any] = {}
    end = None
    for idx, line in enumerate(lines[1:], start=1):
        if line.strip() == "---":
            end = idx
            break
        if not line.strip() or line.lstrip().startswith("#"):
            continue
        if ":" not in line:
            continue
        key, value = line.split(":", 1)
        meta[key.strip()] = parse_scalar(value)
    if end is None:
        return meta, text
    return meta, "\n".join(lines[end + 1 :]).strip()


def load_card(kind: str, card_id: str, registry_root: Path) -> dict[str, Any]:
    path = registry_root / kind / f"{card_id}.md"
    if not path.exists():
        available = sorted(p.stem for p in (registry_root / kind).glob("*.md"))
        raise FileNotFoundError(f"Card not found: {path}. Available {kind}: {available}")
    text = path.read_text(encoding="utf-8")
    meta, body = parse_front_matter(text)
    return {"kind": kind, "id": card_id, "path": str(path), "meta": meta, "body": body}


def list_cards(registry_root: Path) -> dict[str, list[str]]:
    result: dict[str, list[str]] = {}
    for kind in ("elements", "satellites", "sensors", "resolutions"):
        folder = registry_root / kind
        result[kind] = sorted(p.stem for p in folder.glob("*.md")) if folder.exists() else []
    return result


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--element")
    parser.add_argument("--satellite")
    parser.add_argument("--sensor")
    parser.add_argument("--resolution")
    parser.add_argument(
        "--fusion",
        help="Optional sensor/product fusion card, for example FUSED_OPTICAL or STREAM_FUSION.",
    )
    parser.add_argument("--registry-root", default=str(REGISTRY_ROOT))
    parser.add_argument("--output", required=False)
    parser.add_argument("--list", action="store_true", help="List available cards and exit.")
    return parser.parse_args()


def main() -> None:
    args = parse_args()
    registry_root = Path(args.registry_root)

    if args.list:
        print(json.dumps(list_cards(registry_root), indent=2, ensure_ascii=False))
        return

    required = {
        "elements": args.element,
        "satellites": args.satellite,
        "sensors": args.sensor,
        "resolutions": args.resolution,
    }
    missing = [key for key, value in required.items() if not value]
    if missing:
        raise SystemExit(f"Missing required cards: {missing}. Use --list to inspect available cards.")

    cards = {kind: load_card(kind, card_id, registry_root) for kind, card_id in required.items() if card_id}
    if args.fusion:
        cards["fusion"] = load_card("sensors", args.fusion, registry_root)

    element_meta = cards["elements"]["meta"]
    sensor_meta = cards["sensors"]["meta"]
    resolution_meta = cards["resolutions"]["meta"]
    fusion_meta = cards.get("fusion", {}).get("meta", {})
    fusion_state = fusion_meta.get("fusion_state", sensor_meta.get("fusion_state", "none"))

    profile = {
        "profile_id": "_".join(
            part for part in [args.element, args.satellite, args.sensor, args.fusion, args.resolution] if part
        ),
        "cards": cards,
        "task": {
            "element": args.element,
            "task_types": element_meta.get("task_types", []),
            "preferred_heads": element_meta.get("preferred_heads", []),
            "label_formats": element_meta.get("label_formats", []),
            "negative_policy": element_meta.get("negative_policy", "unlabeled_is_ignore"),
        },
        "input": {
            "satellite": args.satellite,
            "sensor": args.sensor,
            "resolution_m": resolution_meta.get("resolution_m"),
            "modalities": sensor_meta.get("modalities", []),
            "common_bands": sensor_meta.get("common_bands", []),
            "recommended_patch_sizes": resolution_meta.get("recommended_patch_sizes", []),
        },
        "fusion": {
            "card": args.fusion,
            "state": fusion_state,
            "modalities": fusion_meta.get("modalities", sensor_meta.get("modalities", [])),
            "supports_streaming_fusion": fusion_meta.get(
                "supports_streaming_fusion", sensor_meta.get("supports_streaming_fusion", False)
            ),
            "requires_fusion_metadata": fusion_meta.get(
                "requires_fusion_metadata", sensor_meta.get("requires_fusion_metadata", False)
            ),
            "required_manifest_fields": [
                "state",
                "method",
                "sources",
                "target_resolution_m",
                "native_multispectral_resolution_m",
                "persisted",
                "reproducible",
                "spectral_preservation",
            ],
        },
        "constraints": {
            "do_not_assume_external_coastline_or_land_mask": True,
            "unlabeled_elements_are_ignore_not_negative": True,
        },
    }

    output = json.dumps(profile, indent=2, ensure_ascii=False)
    if args.output:
        output_path = Path(args.output)
        output_path.parent.mkdir(parents=True, exist_ok=True)
        output_path.write_text(output + "\n", encoding="utf-8")
        print(output_path)
    else:
        print(output)


if __name__ == "__main__":
    main()