cuibinge/typical-marine-ecological-feature-recognition-code

1"""Compose task profiles from Markdown capability cards."""2 3from __future__ import annotations4 5import argparse6import json7from pathlib import Path8from typing import Any9 10 11REGISTRY_ROOT = Path(__file__).resolve().parents[1] / "docs" / "registry"12 13 14def parse_scalar(value: str) -> Any:15    value = value.strip()16    if value in {"true", "True"}:17        return True18    if value in {"false", "False"}:19        return False20    if value in {"null", "None"}:21        return None22    if value.startswith("[") and value.endswith("]"):23        inner = value[1:-1].strip()24        if not inner:25            return []26        return [parse_scalar(part.strip()) for part in inner.split(",")]27    try:28        if "." in value:29            return float(value)30        return int(value)31    except ValueError:32        return value.strip("\"'")33 34 35def parse_front_matter(text: str) -> tuple[dict[str, Any], str]:36    lines = text.splitlines()37    if not lines or lines[0].strip() != "---":38        return {}, text39    meta: dict[str, Any] = {}40    end = None41    for idx, line in enumerate(lines[1:], start=1):42        if line.strip() == "---":43            end = idx44            break45        if not line.strip() or line.lstrip().startswith("#"):46            continue47        if ":" not in line:48            continue49        key, value = line.split(":", 1)50        meta[key.strip()] = parse_scalar(value)51    if end is None:52        return meta, text53    return meta, "\n".join(lines[end + 1 :]).strip()54 55 56def load_card(kind: str, card_id: str, registry_root: Path) -> dict[str, Any]:57    path = registry_root / kind / f"{card_id}.md"58    if not path.exists():59        available = sorted(p.stem for p in (registry_root / kind).glob("*.md"))60        raise FileNotFoundError(f"Card not found: {path}. Available {kind}: {available}")61    text = path.read_text(encoding="utf-8")62    meta, body = parse_front_matter(text)63    return {"kind": kind, "id": card_id, "path": str(path), "meta": meta, "body": body}64 65 66def list_cards(registry_root: Path) -> dict[str, list[str]]:67    result: dict[str, list[str]] = {}68    for kind in ("elements", "satellites", "sensors", "resolutions"):69        folder = registry_root / kind70        result[kind] = sorted(p.stem for p in folder.glob("*.md")) if folder.exists() else []71    return result72 73 74def parse_args() -> argparse.Namespace:75    parser = argparse.ArgumentParser(description=__doc__)76    parser.add_argument("--element")77    parser.add_argument("--satellite")78    parser.add_argument("--sensor")79    parser.add_argument("--resolution")80    parser.add_argument(81        "--fusion",82        help="Optional sensor/product fusion card, for example FUSED_OPTICAL or STREAM_FUSION.",83    )84    parser.add_argument("--registry-root", default=str(REGISTRY_ROOT))85    parser.add_argument("--output", required=False)86    parser.add_argument("--list", action="store_true", help="List available cards and exit.")87    return parser.parse_args()88 89 90def main() -> None:91    args = parse_args()92    registry_root = Path(args.registry_root)93 94    if args.list:95        print(json.dumps(list_cards(registry_root), indent=2, ensure_ascii=False))96        return97 98    required = {99        "elements": args.element,100        "satellites": args.satellite,101        "sensors": args.sensor,102        "resolutions": args.resolution,103    }104    missing = [key for key, value in required.items() if not value]105    if missing:106        raise SystemExit(f"Missing required cards: {missing}. Use --list to inspect available cards.")107 108    cards = {kind: load_card(kind, card_id, registry_root) for kind, card_id in required.items() if card_id}109    if args.fusion:110        cards["fusion"] = load_card("sensors", args.fusion, registry_root)111 112    element_meta = cards["elements"]["meta"]113    sensor_meta = cards["sensors"]["meta"]114    resolution_meta = cards["resolutions"]["meta"]115    fusion_meta = cards.get("fusion", {}).get("meta", {})116    fusion_state = fusion_meta.get("fusion_state", sensor_meta.get("fusion_state", "none"))117 118    profile = {119        "profile_id": "_".join(120            part for part in [args.element, args.satellite, args.sensor, args.fusion, args.resolution] if part121        ),122        "cards": cards,123        "task": {124            "element": args.element,125            "task_types": element_meta.get("task_types", []),126            "preferred_heads": element_meta.get("preferred_heads", []),127            "label_formats": element_meta.get("label_formats", []),128            "negative_policy": element_meta.get("negative_policy", "unlabeled_is_ignore"),129        },130        "input": {131            "satellite": args.satellite,132            "sensor": args.sensor,133            "resolution_m": resolution_meta.get("resolution_m"),134            "modalities": sensor_meta.get("modalities", []),135            "common_bands": sensor_meta.get("common_bands", []),136            "recommended_patch_sizes": resolution_meta.get("recommended_patch_sizes", []),137        },138        "fusion": {139            "card": args.fusion,140            "state": fusion_state,141            "modalities": fusion_meta.get("modalities", sensor_meta.get("modalities", [])),142            "supports_streaming_fusion": fusion_meta.get(143                "supports_streaming_fusion", sensor_meta.get("supports_streaming_fusion", False)144            ),145            "requires_fusion_metadata": fusion_meta.get(146                "requires_fusion_metadata", sensor_meta.get("requires_fusion_metadata", False)147            ),148            "required_manifest_fields": [149                "state",150                "method",151                "sources",152                "target_resolution_m",153                "native_multispectral_resolution_m",154                "persisted",155                "reproducible",156                "spectral_preservation",157            ],158        },159        "constraints": {160            "do_not_assume_external_coastline_or_land_mask": True,161            "unlabeled_elements_are_ignore_not_negative": True,162        },163    }164 165    output = json.dumps(profile, indent=2, ensure_ascii=False)166    if args.output:167        output_path = Path(args.output)168        output_path.parent.mkdir(parents=True, exist_ok=True)169        output_path.write_text(output + "\n", encoding="utf-8")170        print(output_path)171    else:172        print(output)173 174 175if __name__ == "__main__":176    main()177