Update: Add mission pipeline registry

app.py

@@ -129,7 +129,7 @@ def _resolve_mission_plan(
     if not normalized_prompt:
         raise HTTPException(status_code=400, detail="Mission prompt is required.")
     _require_coordinates(latitude, longitude)
-    plan = get_mission_plan(normalized_prompt)
+    plan = get_mission_plan(normalized_prompt, latitude=latitude, longitude=longitude)
     return plan, normalized_prompt
 
 
@@ -150,10 +150,15 @@ def _validate_inputs(
 
 
 def _location_only_prompt(latitude: float, longitude: float) -> str:
+    lat_dir = "N" if latitude >= 0 else "S"
+    lon_dir = "E" if longitude >= 0 else "W"
+    lat_deg = abs(int(latitude))
+    lon_deg = abs(int(longitude))
+    grid_hint = f"{lat_deg}°{lat_dir}, {lon_deg}°{lon_dir}"
     return (
         "Threat reconnaissance mission. "
-        f"Identify and prioritize potential hostile or suspicious object classes around latitude {latitude:.4f}, "
-        f"longitude {longitude:.4f}. Consider common threats for this environment when selecting classes."
+        f"Identify and prioritize potential hostile or suspicious object classes near grid {grid_hint}. "
+        "Consider common threats for this environment when selecting classes."
     )
 
 
@@ -164,7 +169,7 @@ async def location_context(
 ):
     prompt = _location_only_prompt(latitude, longitude)
     try:
-        plan = get_mission_plan(prompt)
+        plan = get_mission_plan(prompt, latitude=latitude, longitude=longitude)
     except Exception as exc:
         logging.exception("Location-only planning failed.")
         raise HTTPException(status_code=500, detail=str(exc))
@@ -186,7 +191,7 @@ async def mission_plan_endpoint(
     if not normalized_prompt:
         raise HTTPException(status_code=400, detail="Prompt is required.")
     try:
-        plan = get_mission_plan(normalized_prompt)
+        plan = get_mission_plan(normalized_prompt, latitude=latitude, longitude=longitude)
     except Exception as exc:
         logging.exception("Mission planning failed.")
         raise HTTPException(status_code=500, detail=str(exc))

coco_classes.py

@@ -0,0 +1,163 @@
+from __future__ import annotations
+
+import difflib
+import re
+from typing import Dict, Tuple
+
+COCO_CLASSES: Tuple[str, ...] = (
+    "person",
+    "bicycle",
+    "car",
+    "motorcycle",
+    "airplane",
+    "bus",
+    "train",
+    "truck",
+    "boat",
+    "traffic light",
+    "fire hydrant",
+    "stop sign",
+    "parking meter",
+    "bench",
+    "bird",
+    "cat",
+    "dog",
+    "horse",
+    "sheep",
+    "cow",
+    "elephant",
+    "bear",
+    "zebra",
+    "giraffe",
+    "backpack",
+    "umbrella",
+    "handbag",
+    "tie",
+    "suitcase",
+    "frisbee",
+    "skis",
+    "snowboard",
+    "sports ball",
+    "kite",
+    "baseball bat",
+    "baseball glove",
+    "skateboard",
+    "surfboard",
+    "tennis racket",
+    "bottle",
+    "wine glass",
+    "cup",
+    "fork",
+    "knife",
+    "spoon",
+    "bowl",
+    "banana",
+    "apple",
+    "sandwich",
+    "orange",
+    "broccoli",
+    "carrot",
+    "hot dog",
+    "pizza",
+    "donut",
+    "cake",
+    "chair",
+    "couch",
+    "potted plant",
+    "bed",
+    "dining table",
+    "toilet",
+    "tv",
+    "laptop",
+    "mouse",
+    "remote",
+    "keyboard",
+    "cell phone",
+    "microwave",
+    "oven",
+    "toaster",
+    "sink",
+    "refrigerator",
+    "book",
+    "clock",
+    "vase",
+    "scissors",
+    "teddy bear",
+    "hair drier",
+    "toothbrush",
+)
+
+
+def coco_class_catalog() -> str:
+    """Return the COCO classes in a comma-separated catalog for prompts."""
+
+    return ", ".join(COCO_CLASSES)
+
+
+def _normalize(label: str) -> str:
+    return re.sub(r"[^a-z0-9]+", " ", label.lower()).strip()
+
+
+_CANONICAL_LOOKUP: Dict[str, str] = {_normalize(name): name for name in COCO_CLASSES}
+_COCO_SYNONYMS: Dict[str, str] = {
+    "people": "person",
+    "man": "person",
+    "woman": "person",
+    "men": "person",
+    "women": "person",
+    "motorbike": "motorcycle",
+    "motor bike": "motorcycle",
+    "bike": "bicycle",
+    "aircraft": "airplane",
+    "plane": "airplane",
+    "jet": "airplane",
+    "aeroplane": "airplane",
+    "pickup": "truck",
+    "pickup truck": "truck",
+    "semi": "truck",
+    "lorry": "truck",
+    "tractor trailer": "truck",
+    "coach": "bus",
+    "television": "tv",
+    "tv monitor": "tv",
+    "mobile phone": "cell phone",
+    "smartphone": "cell phone",
+    "cellphone": "cell phone",
+    "dinner table": "dining table",
+    "sofa": "couch",
+    "cooker": "oven",
+}
+_ALIAS_LOOKUP: Dict[str, str] = {_normalize(alias): canonical for alias, canonical in _COCO_SYNONYMS.items()}
+
+
+def canonicalize_coco_name(value: str | None) -> str | None:
+    """Map an arbitrary string to the closest COCO class name if possible."""
+
+    if not value:
+        return None
+    normalized = _normalize(value)
+    if not normalized:
+        return None
+    if normalized in _CANONICAL_LOOKUP:
+        return _CANONICAL_LOOKUP[normalized]
+    if normalized in _ALIAS_LOOKUP:
+        return _ALIAS_LOOKUP[normalized]
+
+    for alias_norm, canonical in _ALIAS_LOOKUP.items():
+        if alias_norm and alias_norm in normalized:
+            return canonical
+    for canonical_norm, canonical in _CANONICAL_LOOKUP.items():
+        if canonical_norm and canonical_norm in normalized:
+            return canonical
+
+    tokens = normalized.split()
+    for token in tokens:
+        if token in _CANONICAL_LOOKUP:
+            return _CANONICAL_LOOKUP[token]
+        if token in _ALIAS_LOOKUP:
+            return _ALIAS_LOOKUP[token]
+
+    close = difflib.get_close_matches(normalized, list(_CANONICAL_LOOKUP.keys()), n=1, cutoff=0.82)
+    if close:
+        return _CANONICAL_LOOKUP[close[0]]
+    return None

inference.py

@@ -6,6 +6,7 @@ import numpy as np
 from models.model_loader import load_detector
 from mission_planner import MissionPlan, get_mission_plan
 from mission_summarizer import summarize_results
+from pipeline_registry import get_pipeline_spec
 from utils.video import extract_frames, write_video
 
 
@@ -86,6 +87,18 @@ def run_inference(
     resolved_plan = mission_plan or get_mission_plan(mission_prompt_clean)
     logging.info("Mission plan: %s", resolved_plan.to_json())
     queries = resolved_plan.queries()
+    plan_detector = None
+    if resolved_plan.pipeline and resolved_plan.pipeline.primary_id:
+        spec = get_pipeline_spec(resolved_plan.pipeline.primary_id)
+        if spec:
+            hf_bindings = spec.get("huggingface") or {}
+            detections = hf_bindings.get("detection") or []
+            for entry in detections:
+                detector_key = entry.get("detector_key")
+                if detector_key:
+                    plan_detector = detector_key
+                    break
+    active_detector = detector_name or plan_detector
 
     processed_frames: List[np.ndarray] = []
     detection_log: List[Dict[str, Any]] = []
@@ -93,7 +106,7 @@ def run_inference(
         if max_frames is not None and idx >= max_frames:
             break
         logging.debug("Processing frame %d", idx)
-        processed_frame, detections = infer_frame(frame, queries, detector_name=detector_name)
+        processed_frame, detections = infer_frame(frame, queries, detector_name=active_detector)
         detection_log.append({"frame_index": idx, "detections": detections})
         processed_frames.append(processed_frame)
 

mission_context.py

@@ -0,0 +1,277 @@
+from __future__ import annotations
+
+import json
+import re
+from dataclasses import asdict, dataclass
+from datetime import datetime
+from pathlib import Path
+from typing import Any, Dict, List, Literal, Optional, Tuple
+
+try:
+    import reverse_geocoder as rg  # type: ignore
+except ImportError:  # pragma: no cover
+    rg = None
+
+try:
+    from timezonefinder import TimezoneFinder  # type: ignore
+except ImportError:  # pragma: no cover
+    TimezoneFinder = None  # type: ignore
+
+try:
+    from zoneinfo import ZoneInfo
+except ImportError:  # pragma: no cover
+    ZoneInfo = None  # type: ignore
+
+
+MISSION_TYPE_OPTIONS: Tuple[str, ...] = (
+    "surveillance",
+    "tracking",
+    "threat_detection",
+    "safety_monitoring",
+)
+
+LOCATION_TYPE_OPTIONS: Tuple[str, ...] = (
+    "urban",
+    "suburban",
+    "rural",
+    "industrial",
+    "coastal",
+    "harbor",
+    "bridge",
+    "roadway",
+    "indoor",
+    "unknown",
+)
+
+TIME_OF_DAY_OPTIONS: Tuple[str, ...] = ("day", "night")
+
+PRIORITY_LEVEL_OPTIONS: Tuple[str, ...] = ("routine", "elevated", "high")
+
+INTEL_FLAG_OPTIONS: Tuple[str, ...] = ("none", "recent_incident", "elevated_activity")
+
+SENSOR_TYPE_OPTIONS: Tuple[str, ...] = ("rgb_camera", "thermal_camera")
+
+COMPUTE_MODE_OPTIONS: Tuple[str, ...] = ("edge", "hybrid", "cloud")
+
+_DEFAULT_FOCUS_TERMS: Tuple[str, ...] = (
+    "drone",
+    "uav",
+    "vehicle",
+    "truck",
+    "convoy",
+    "ship",
+    "boat",
+    "aircraft",
+    "personnel",
+    "crowd",
+    "wildfire",
+    "perimeter",
+    "checkpoint",
+    "bridge",
+    "chemical",
+    "radiation",
+    "pipeline",
+)
+
+_KEYWORD_CONFIG_PATH = Path(__file__).with_name("mission_keywords.json")
+
+
+def _load_focus_terms() -> Tuple[str, ...]:
+    focus_terms = list(_DEFAULT_FOCUS_TERMS)
+    if not _KEYWORD_CONFIG_PATH.exists():
+        return tuple(focus_terms)
+    try:
+        config = json.loads(_KEYWORD_CONFIG_PATH.read_text())
+    except Exception:
+        return tuple(focus_terms)
+    focus_overrides = config.get("focus_terms")
+    if isinstance(focus_overrides, (list, tuple)):
+        normalized_focus = [str(item).strip() for item in focus_overrides if str(item).strip()]
+        if normalized_focus:
+            focus_terms = normalized_focus
+    return tuple(focus_terms)
+
+
+MISSION_FOCUS_KEYWORDS = _load_focus_terms()
+_TIMEZONE_FINDER = TimezoneFinder() if TimezoneFinder is not None else None
+
+
+@dataclass
+class MissionContext:
+    mission_type: Literal["surveillance", "tracking", "threat_detection", "safety_monitoring"] = "surveillance"
+    location_type: Literal[
+        "urban",
+        "suburban",
+        "rural",
+        "industrial",
+        "coastal",
+        "harbor",
+        "bridge",
+        "roadway",
+        "indoor",
+        "unknown",
+    ] = "unknown"
+    time_of_day: Literal["day", "night"] | None = None
+    priority_level: Literal["routine", "elevated", "high"] | None = None
+    intel_flag: Literal["none", "recent_incident", "elevated_activity"] = "none"
+    sensor_type: Literal["rgb_camera", "thermal_camera"] = "rgb_camera"
+    compute_mode: Literal["edge", "hybrid", "cloud"] = "edge"
+
+    def __post_init__(self) -> None:
+        self._validate()
+
+    def _validate(self) -> None:
+        self._ensure_choice("mission_type", self.mission_type, MISSION_TYPE_OPTIONS)
+        self._ensure_choice("location_type", self.location_type, LOCATION_TYPE_OPTIONS)
+        if self.time_of_day is not None:
+            self._ensure_choice("time_of_day", self.time_of_day, TIME_OF_DAY_OPTIONS)
+        if self.priority_level is not None:
+            self._ensure_choice("priority_level", self.priority_level, PRIORITY_LEVEL_OPTIONS)
+        self._ensure_choice("intel_flag", self.intel_flag, INTEL_FLAG_OPTIONS)
+        self._ensure_choice("sensor_type", self.sensor_type, SENSOR_TYPE_OPTIONS)
+        self._ensure_choice("compute_mode", self.compute_mode, COMPUTE_MODE_OPTIONS)
+
+    @staticmethod
+    def _ensure_choice(field: str, value: str, choices: Tuple[str, ...]) -> None:
+        if value not in choices:
+            allowed = ", ".join(choices)
+            raise ValueError(f"Invalid value '{value}' for {field}. Allowed values: {allowed}.")
+
+    def to_prompt_payload(self) -> Dict[str, Any]:
+        payload: Dict[str, Any] = {
+            "mission_type": self.mission_type,
+            "location_type": self.location_type,
+        }
+        if self.time_of_day:
+            payload["time_of_day"] = self.time_of_day
+        return payload
+
+
+@dataclass
+class MissionClass:
+    name: str
+    score: float
+    rationale: str
+
+
+@dataclass
+class PipelineRecommendation:
+    primary_id: str | None = None
+    primary_reason: str | None = None
+    fallback_id: str | None = None
+    fallback_reason: str | None = None
+
+    def to_dict(self) -> Dict[str, Any]:
+        return {
+            "primary": self._entry_dict(self.primary_id, self.primary_reason),
+            "fallback": self._entry_dict(self.fallback_id, self.fallback_reason),
+        }
+
+    @staticmethod
+    def _entry_dict(pipeline_id: str | None, reason: str | None) -> Dict[str, Any] | None:
+        if not pipeline_id:
+            return None
+        from pipeline_registry import get_pipeline_spec
+
+        spec = get_pipeline_spec(pipeline_id)
+        return {
+            "id": pipeline_id,
+            "name": spec["id"] if spec else pipeline_id,
+            "reason": reason or "",
+            "modalities": spec.get("modalities") if spec else None,
+            "location_types": spec.get("location_types") if spec else None,
+            "time_of_day": spec.get("time_of_day") if spec else None,
+            "huggingface": spec.get("huggingface") if spec else None,
+        }
+
+
+@dataclass
+class MissionPlan:
+    mission: str
+    relevant_classes: List[MissionClass]
+    context: MissionContext | None = None
+    pipeline: PipelineRecommendation | None = None
+
+    def queries(self) -> List[str]:
+        return [entry.name for entry in self.relevant_classes]
+
+    def to_dict(self) -> Dict[str, Any]:
+        context_payload = self.context.to_prompt_payload() if self.context else None
+        return {
+            "mission": self.mission,
+            "context": context_payload,
+            "entities": [asdict(entry) for entry in self.relevant_classes],
+            "pipelines": self.pipeline.to_dict() if self.pipeline else None,
+        }
+
+    def to_json(self) -> str:
+        import json
+
+        return json.dumps(self.to_dict())
+
+
+def build_prompt_hints(
+    mission: str,
+    latitude: Optional[float],
+    longitude: Optional[float],
+) -> Dict[str, Any]:
+    hints: Dict[str, Any] = {}
+    focus_terms = _extract_focus_terms(mission)
+    if focus_terms:
+        hints["mission_focus_terms"] = focus_terms
+    timezone_id = _lookup_timezone(latitude, longitude)
+    if timezone_id:
+        hints["timezone"] = timezone_id
+    local_time = _local_time_from_timezone(timezone_id)
+    if local_time:
+        hints["local_time"] = local_time
+    locality = _nearest_locality(latitude, longitude)
+    if locality:
+        hints["nearest_locality"] = locality
+    return hints
+
+
+def _normalize(text: str) -> str:
+    return re.sub(r"\s+", " ", text.lower()).strip()
+
+
+def _extract_focus_terms(mission: str) -> List[str]:
+    text = _normalize(mission)
+    matched = [term for term in MISSION_FOCUS_KEYWORDS if term in text]
+    return matched[:5]
+
+
+def _lookup_timezone(latitude: Optional[float], longitude: Optional[float]) -> Optional[str]:
+    if latitude is None or longitude is None or _TIMEZONE_FINDER is None:
+        return None
+    try:
+        return _TIMEZONE_FINDER.timezone_at(lat=latitude, lng=longitude)
+    except Exception:
+        return None
+
+
+def _local_time_from_timezone(timezone_id: Optional[str]) -> Optional[str]:
+    if timezone_id is None or ZoneInfo is None:
+        return None
+    try:
+        tz = ZoneInfo(timezone_id)
+    except Exception:
+        return None
+    return datetime.now(tz).isoformat(timespec="seconds")
+
+
+def _nearest_locality(latitude: Optional[float], longitude: Optional[float]) -> Optional[str]:
+    if latitude is None or longitude is None or rg is None:
+        return None
+    try:
+        results = rg.search((latitude, longitude), mode=1)  # type: ignore[arg-type]
+    except Exception:
+        return None
+    if not results:
+        return None
+    match = results[0]
+    city = (match.get("name") or "").strip()
+    admin = (match.get("admin1") or "").strip()
+    country = (match.get("cc") or "").strip()
+    components = [component for component in (city, admin, country) if component]
+    return ", ".join(components) if components else None

mission_planner.py

@@ -2,125 +2,35 @@ from __future__ import annotations
 
 import json
 import logging
-from dataclasses import asdict, dataclass
-from typing import Dict, List, Tuple
+from dataclasses import asdict, dataclass, replace
+from datetime import datetime
+from typing import Any, Dict, List, Mapping, Tuple
 
+from coco_classes import canonicalize_coco_name, coco_class_catalog
+from mission_context import (
+    MissionClass,
+    MissionContext,
+    MissionPlan,
+    MISSION_TYPE_OPTIONS,
+    LOCATION_TYPE_OPTIONS,
+    TIME_OF_DAY_OPTIONS,
+    PRIORITY_LEVEL_OPTIONS,
+    PipelineRecommendation,
+    build_prompt_hints,
+)
+from pipeline_registry import (
+    PIPELINE_SPECS,
+    fallback_pipeline_for_context,
+    filter_pipelines_for_context,
+    get_pipeline_spec,
+)
 from prompt import mission_planner_system_prompt, mission_planner_user_prompt
 from utils.openai_client import get_openai_client
 
 
-YOLO_CLASSES: Tuple[str, ...] = (
-    "person",
-    "bicycle",
-    "car",
-    "motorcycle",
-    "airplane",
-    "bus",
-    "train",
-    "truck",
-    "boat",
-    "traffic light",
-    "fire hydrant",
-    "stop sign",
-    "parking meter",
-    "bench",
-    "bird",
-    "cat",
-    "dog",
-    "horse",
-    "sheep",
-    "cow",
-    "elephant",
-    "bear",
-    "zebra",
-    "giraffe",
-    "backpack",
-    "umbrella",
-    "handbag",
-    "tie",
-    "suitcase",
-    "frisbee",
-    "skis",
-    "snowboard",
-    "sports ball",
-    "kite",
-    "baseball bat",
-    "baseball glove",
-    "skateboard",
-    "surfboard",
-    "tennis racket",
-    "bottle",
-    "wine glass",
-    "cup",
-    "fork",
-    "knife",
-    "spoon",
-    "bowl",
-    "banana",
-    "apple",
-    "sandwich",
-    "orange",
-    "broccoli",
-    "carrot",
-    "hot dog",
-    "pizza",
-    "donut",
-    "cake",
-    "chair",
-    "couch",
-    "potted plant",
-    "bed",
-    "dining table",
-    "toilet",
-    "tv",
-    "laptop",
-    "mouse",
-    "remote",
-    "keyboard",
-    "cell phone",
-    "microwave",
-    "oven",
-    "toaster",
-    "sink",
-    "refrigerator",
-    "book",
-    "clock",
-    "vase",
-    "scissors",
-    "teddy bear",
-    "hair drier",
-    "toothbrush",
-)
-
-
 DEFAULT_OPENAI_MODEL = "gpt-4o-mini"
 
 
-@dataclass
-class MissionClass:
-    name: str
-    score: float
-    rationale: str
-
-
-@dataclass
-class MissionPlan:
-    mission: str
-    relevant_classes: List[MissionClass]
-
-    def queries(self) -> List[str]:
-        return [entry.name for entry in self.relevant_classes]
-
-    def to_dict(self) -> dict:
-        return {
-            "mission": self.mission,
-            "classes": [asdict(entry) for entry in self.relevant_classes],
-        }
-
-    def to_json(self) -> str:
-        return json.dumps(self.to_dict())
-
-
 class MissionReasoner:
     def __init__(
         self,
@@ -130,22 +40,133 @@ class MissionReasoner:
     ) -> None:
         self._model_name = model_name
         self._top_k = top_k
+        self._coco_catalog = coco_class_catalog()
 
-    def plan(self, mission: str) -> MissionPlan:
+    def plan(
+        self,
+        mission: str,
+        *,
+        context: MissionContext,
+        cues: Mapping[str, Any] | None = None,
+    ) -> MissionPlan:
         mission = (mission or "").strip()
         if not mission:
             raise ValueError("Mission prompt cannot be empty.")
-        response_payload = self._query_llm(mission)
+        available_pipelines = self._candidate_pipelines(mission, context, cues)
+        candidate_ids = [spec["id"] for spec in available_pipelines] or [PIPELINE_SPECS[0]["id"]]
+        lock_pipeline_id = candidate_ids[0] if len(candidate_ids) == 1 else None
+        response_payload = self._query_llm(
+            mission,
+            context=context,
+            cues=None,
+            pipeline_ids=candidate_ids,
+        )
         relevant = self._parse_plan(response_payload, fallback_mission=mission)
-        return MissionPlan(mission=response_payload.get("mission", mission), relevant_classes=relevant[: self._top_k])
+        enriched_context = self._merge_context(context, response_payload.get("context"))
+        if lock_pipeline_id:
+            pipeline_rec = PipelineRecommendation(
+                primary_id=lock_pipeline_id,
+                primary_reason="Only pipeline compatible with mission context.",
+            )
+        else:
+            pipeline_rec = self._parse_pipeline_recommendation(
+                response_payload.get("pipelines") or response_payload.get("pipeline"),
+                available_pipelines,
+                context,
+            )
+        return MissionPlan(
+            mission=response_payload.get("mission", mission),
+            relevant_classes=relevant[: self._top_k],
+            context=enriched_context,
+            pipeline=pipeline_rec,
+        )
+
+    def _render_pipeline_catalog(self, specs: List[Dict[str, object]]) -> str:
+        if not specs:
+            return "No compatible pipelines available."
+        sections: List[str] = []
+        for spec in specs:
+            reason = spec.get("availability_reason") or "Compatible with mission context."
+            hf_bindings = spec.get("huggingface") or {}
+
+            def _format_models(models: List[Dict[str, object]]) -> str:
+                if not models:
+                    return "none"
+                labels = []
+                for entry in models:
+                    model_id = entry.get("model_id") or entry.get("name") or "unknown"
+                    label = entry.get("label") or model_id
+                    suffix = " (optional)" if entry.get("optional") else ""
+                    labels.append(f"{label}{suffix}")
+                return ", ".join(labels)
+
+            detection_models = _format_models(hf_bindings.get("detection", []))
+            segmentation_models = _format_models(hf_bindings.get("segmentation", []))
+            tracking_models = _format_models(hf_bindings.get("tracking", []))
+            hf_notes = hf_bindings.get("notes") or ""
+            sections.append(
+                "\n".join(
+                    [
+                        f"{spec['id']} pipeline",
+                        f"  Modalities: {', '.join(spec.get('modalities', ())) or 'unspecified'}",
+                        f"  Locations: {', '.join(spec.get('location_types', ())) or 'any'}",
+                        f"  Time of day: {', '.join(spec.get('time_of_day', ())) or 'any'}",
+                        f"  Availability: {reason}",
+                        f"  HF detection: {detection_models}",
+                        f"  HF segmentation: {segmentation_models}",
+                        f"  Tracking: {tracking_models}",
+                        f"  Notes: {hf_notes or 'n/a'}",
+                    ]
+                )
+            )
+        return "\n\n".join(sections)
+
+    def _candidate_pipelines(
+        self,
+        mission: str,
+        context: MissionContext,
+        cues: Mapping[str, Any] | None,
+    ) -> List[Dict[str, object]]:
+        filtered = filter_pipelines_for_context(context)
+        if filtered:
+            return filtered
+        fallback_spec = fallback_pipeline_for_context(context, [])
+        if fallback_spec is None:
+            logging.error("No fallback pipeline available; mission context=%s", context)
+            return [dict(spec) for spec in PIPELINE_SPECS]
+        logging.warning(
+            "No compatible pipelines for context %s; selecting fallback %s.",
+            context,
+            fallback_spec["id"],
+        )
+        fallback_copy = dict(fallback_spec)
+        fallback_copy["availability_reason"] = (
+            "Fallback engaged because no specialized pipeline matched this mission context."
+        )
+        return [fallback_copy]
 
-    def _query_llm(self, mission: str) -> Dict[str, object]:
+    def _query_llm(
+        self,
+        mission: str,
+        *,
+        context: MissionContext,
+        cues: Mapping[str, Any] | None = None,
+        pipeline_ids: List[str] | None,
+    ) -> Dict[str, object]:
         client = get_openai_client()
         system_prompt = mission_planner_system_prompt()
-        user_prompt = mission_planner_user_prompt(mission, YOLO_CLASSES, self._top_k)
+        context_payload = context.to_prompt_payload()
+        user_prompt = mission_planner_user_prompt(
+            mission,
+            self._top_k,
+            context=context_payload,
+            cues=cues,
+            pipeline_candidates=pipeline_ids,
+            coco_catalog=self._coco_catalog,
+        )
         completion = client.chat.completions.create(
             model=self._model_name,
-            temperature=0.2,
+            temperature=0.1,
             response_format={"type": "json_object"},
             messages=[
                 {"role": "system", "content": system_prompt},
@@ -160,7 +181,12 @@ class MissionReasoner:
             return {"mission": mission, "classes": []}
 
     def _parse_plan(self, payload: Dict[str, object], fallback_mission: str) -> List[MissionClass]:
-        entries = payload.get("classes") or payload.get("relevant_classes") or []
+        entries = (
+            payload.get("entities")
+            or payload.get("classes")
+            or payload.get("relevant_classes")
+            or []
+        )
         mission = payload.get("mission") or fallback_mission
         parsed: List[MissionClass] = []
         seen = set()
@@ -168,35 +194,204 @@ class MissionReasoner:
             if not isinstance(entry, dict):
                 continue
             name = str(entry.get("name") or "").strip()
-            if not name or name not in YOLO_CLASSES or name in seen:
+            if not name:
+                continue
+            canonical_name = canonicalize_coco_name(name)
+            if not canonical_name:
+                logging.warning("Skipping non-COCO entity '%s'.", name)
                 continue
-            seen.add(name)
+            if canonical_name in seen:
+                continue
+            seen.add(canonical_name)
             score_raw = entry.get("score")
             try:
                 score = float(score_raw)
             except (TypeError, ValueError):
                 score = 0.5
             rationale = str(entry.get("rationale") or f"Track '{name}' for mission '{mission}'.")
-            parsed.append(MissionClass(name=name, score=max(0.0, min(1.0, score)), rationale=rationale))
-
-        if not parsed:
-            logging.warning("LLM returned no usable classes. Falling back to default YOLO list.")
-            parsed = [
+            parsed.append(
                 MissionClass(
-                    name=label,
-                    score=1.0 - (idx * 0.05),
-                    rationale=f"Fallback selection for mission '{mission}'.",
+                    name=canonical_name,
+                    score=max(0.0, min(1.0, score)),
+                    rationale=rationale,
                 )
-                for idx, label in enumerate(YOLO_CLASSES[: self._top_k])
-            ]
+            )
+
+        if not parsed:
+            raise RuntimeError("LLM returned no semantic entities; aborting instead of fabricating outputs.")
         return parsed
 
+    def _merge_context(
+        self,
+        base_context: MissionContext,
+        context_payload: Dict[str, object] | None,
+    ) -> MissionContext:
+        payload = context_payload or {}
+        if not isinstance(payload, dict):
+            return base_context
+
+        def _coerce_choice(value: object | None, allowed: Tuple[str, ...]) -> str | None:
+            if value is None:
+                return None
+            candidate = str(value).strip().lower()
+            return candidate if candidate in allowed else None
+
+        updates: Dict[str, Any] = {}
+        new_mission_type = _coerce_choice(payload.get("mission_type"), MISSION_TYPE_OPTIONS)
+        new_location_type = _coerce_choice(payload.get("location_type"), LOCATION_TYPE_OPTIONS)
+        new_time_of_day = _coerce_choice(payload.get("time_of_day"), TIME_OF_DAY_OPTIONS)
+        new_priority = _coerce_choice(payload.get("priority_level"), PRIORITY_LEVEL_OPTIONS)
+
+        if new_mission_type:
+            updates["mission_type"] = new_mission_type
+        if new_location_type:
+            updates["location_type"] = new_location_type
+        if new_time_of_day:
+            updates["time_of_day"] = new_time_of_day
+        if new_priority:
+            updates["priority_level"] = new_priority
+
+        if not updates:
+            return base_context
+        return replace(base_context, **updates)
+
+    def _parse_pipeline_recommendation(
+        self,
+        payload: object,
+        available_specs: List[Dict[str, object]],
+        context: MissionContext,
+    ) -> PipelineRecommendation | None:
+        if not isinstance(payload, dict):
+            return self._validate_pipeline_selection(None, available_specs, context)
+
+        if "id" in payload or "pipeline_id" in payload or "pipeline" in payload:
+            pipeline_id_raw = payload.get("id") or payload.get("pipeline_id") or payload.get("pipeline")
+            pipeline_id = str(pipeline_id_raw or "").strip()
+            reason = str(payload.get("reason") or "").strip() or None
+            candidate = PipelineRecommendation(primary_id=pipeline_id or None, primary_reason=reason)
+            return self._validate_pipeline_selection(candidate, available_specs, context)
+
+        def _extract_entry(entry_key: str) -> tuple[str | None, str | None]:
+            value = payload.get(entry_key)
+            if not isinstance(value, dict):
+                return None, None
+            pipeline_id_raw = value.get("id") or value.get("pipeline_id") or value.get("pipeline")
+            pipeline_id = str(pipeline_id_raw).strip()
+            if not pipeline_id:
+                return None, None
+            if not get_pipeline_spec(pipeline_id):
+                return None, None
+            reason = str(value.get("reason") or "").strip() or None
+            return pipeline_id, reason
+
+        primary_id, primary_reason = _extract_entry("primary")
+        fallback_id, fallback_reason = _extract_entry("fallback")
+
+        rec = PipelineRecommendation(
+            primary_id=primary_id,
+            primary_reason=primary_reason,
+            fallback_id=fallback_id,
+            fallback_reason=fallback_reason,
+        )
+        return self._validate_pipeline_selection(rec, available_specs, context)
+
+    def _validate_pipeline_selection(
+        self,
+        candidate: PipelineRecommendation | None,
+        available_specs: List[Dict[str, object]],
+        context: MissionContext,
+    ) -> PipelineRecommendation | None:
+        if not available_specs:
+            return None
+        available_ids = {spec["id"] for spec in available_specs}
+
+        def _normalize_reason(reason: str | None, default: str) -> str:
+            text = (reason or "").strip()
+            return text or default
+
+        primary_id = candidate.primary_id if candidate and candidate.primary_id in available_ids else None
+        if not primary_id:
+            fallback_spec = fallback_pipeline_for_context(context, available_specs)
+            if fallback_spec is None:
+                logging.warning("No pipelines available even after fallback.")
+                return None
+            logging.warning(
+                "Pipeline recommendation invalid or missing. Defaulting to %s.", fallback_spec["id"]
+            )
+            return PipelineRecommendation(
+                primary_id=fallback_spec["id"],
+                primary_reason=_normalize_reason(
+                    candidate.primary_reason if candidate else None,
+                    "Auto-selected based on available sensors and context.",
+                ),
+                fallback_id=None,
+                fallback_reason=None,
+            )
+
+        primary_reason = _normalize_reason(candidate.primary_reason if candidate else None, "LLM-selected.")
+
+        fallback_allowed = context.priority_level in {"elevated", "high"}
+        fallback_id = candidate.fallback_id if candidate else None
+        fallback_reason = candidate.fallback_reason if candidate else None
+
+        if not fallback_allowed or fallback_id not in available_ids or fallback_id == primary_id:
+            if fallback_id:
+                logging.info("Dropping fallback pipeline %s due to priority/context constraints.", fallback_id)
+            fallback_id_valid = None
+            fallback_reason_valid = None
+        else:
+            fallback_id_valid = fallback_id
+            fallback_reason_valid = _normalize_reason(fallback_reason, "Fallback allowed due to priority level.")
+
+        return PipelineRecommendation(
+            primary_id=primary_id,
+            primary_reason=primary_reason,
+            fallback_id=fallback_id_valid,
+            fallback_reason=fallback_reason_valid,
+        )
+
 
 _REASONER: MissionReasoner | None = None
 
 
-def get_mission_plan(mission: str) -> MissionPlan:
+def get_mission_plan(
+    mission: str,
+    *,
+    latitude: float | None = None,
+    longitude: float | None = None,
+    context_overrides: MissionContext | None = None,
+) -> MissionPlan:
     global _REASONER
     if _REASONER is None:
         _REASONER = MissionReasoner()
-    return _REASONER.plan(mission)
+    context = context_overrides or MissionContext()
+    cues = build_prompt_hints(mission, latitude, longitude)
+    if latitude is not None and longitude is not None:
+        logging.info("Mission location coordinates: lat=%s, lon=%s", latitude, longitude)
+    local_time_hint = cues.get("local_time") if isinstance(cues, Mapping) else None
+    if local_time_hint:
+        logging.info("Derived local mission time: %s", local_time_hint)
+    timezone_hint = cues.get("timezone") if isinstance(cues, Mapping) else None
+    if timezone_hint:
+        logging.info("Derived local timezone: %s", timezone_hint)
+    locality_hint = cues.get("nearest_locality") if isinstance(cues, Mapping) else None
+    if locality_hint:
+        logging.info("Reverse geocoded locality: %s", locality_hint)
+    inferred_time = _infer_time_of_day_from_cues(context, cues)
+    if inferred_time and context.time_of_day != inferred_time:
+        context = replace(context, time_of_day=inferred_time)
+    return _REASONER.plan(mission, context=context, cues=cues)
+
+
+def _infer_time_of_day_from_cues(context: MissionContext, cues: Mapping[str, Any] | None) -> str | None:
+    if context.time_of_day or not cues:
+        return context.time_of_day
+    local_time_raw = cues.get("local_time") if isinstance(cues, Mapping) else None
+    if not local_time_raw:
+        return None
+    try:
+        local_dt = datetime.fromisoformat(str(local_time_raw))
+    except (ValueError, TypeError):
+        return None
+    hour = local_dt.hour
+    return "day" if 6 <= hour < 18 else "night"

mission_planner_cli.py

@@ -0,0 +1,126 @@
+#!/usr/bin/env python3
+"""Command-line helper to run the mission planner independently."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import logging
+import sys
+from typing import Dict, Any
+
+from mission_context import (
+    COMPUTE_MODE_OPTIONS,
+    INTEL_FLAG_OPTIONS,
+    LOCATION_TYPE_OPTIONS,
+    MISSION_TYPE_OPTIONS,
+    SENSOR_TYPE_OPTIONS,
+    TIME_OF_DAY_OPTIONS,
+    PRIORITY_LEVEL_OPTIONS,
+    MissionContext,
+)
+from mission_planner import get_mission_plan
+
+
+def _parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(
+        description="Run the mission planner with a prompt and location context."
+    )
+    parser.add_argument(
+        "--prompt",
+        required=True,
+        help="Mission description to plan for.",
+    )
+    parser.add_argument(
+        "--latitude",
+        type=float,
+        required=True,
+        help="Latitude for the current operation.",
+    )
+    parser.add_argument(
+        "--longitude",
+        type=float,
+        required=True,
+        help="Longitude for the current operation.",
+    )
+    parser.add_argument(
+        "--mission-type",
+        choices=MISSION_TYPE_OPTIONS,
+        help="Optional override for mission type.",
+    )
+    parser.add_argument(
+        "--location-type",
+        choices=LOCATION_TYPE_OPTIONS,
+        help="Optional override for location type.",
+    )
+    parser.add_argument(
+        "--time-of-day",
+        choices=TIME_OF_DAY_OPTIONS,
+        help="Optional override for time of day.",
+    )
+    parser.add_argument(
+        "--priority-level",
+        choices=PRIORITY_LEVEL_OPTIONS,
+        help="Optional override for mission priority.",
+    )
+    parser.add_argument(
+        "--intel-flag",
+        choices=INTEL_FLAG_OPTIONS,
+        help="Operator-set intel status override.",
+    )
+    parser.add_argument(
+        "--sensor-type",
+        choices=SENSOR_TYPE_OPTIONS,
+        help="Override for primary sensor type.",
+    )
+    parser.add_argument(
+        "--compute-mode",
+        choices=COMPUTE_MODE_OPTIONS,
+        help="Override for deployment compute mode.",
+    )
+    parser.add_argument(
+        "--indent",
+        type=int,
+        default=2,
+        help="Pretty-print JSON using this indent (set to 0 for compact output).",
+    )
+    return parser.parse_args()
+
+
+def main() -> int:
+    logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s")
+    args = _parse_args()
+    prompt = (args.prompt or "").strip()
+    if not prompt:
+        print("Prompt cannot be empty.", file=sys.stderr)
+        return 1
+
+    context_kwargs: Dict[str, Any] = {}
+    if args.mission_type:
+        context_kwargs["mission_type"] = args.mission_type
+    if args.location_type:
+        context_kwargs["location_type"] = args.location_type
+    if args.time_of_day:
+        context_kwargs["time_of_day"] = args.time_of_day
+    if args.priority_level:
+        context_kwargs["priority_level"] = args.priority_level
+    if args.intel_flag:
+        context_kwargs["intel_flag"] = args.intel_flag
+    if args.sensor_type:
+        context_kwargs["sensor_type"] = args.sensor_type
+    if args.compute_mode:
+        context_kwargs["compute_mode"] = args.compute_mode
+    context = MissionContext(**context_kwargs) if context_kwargs else None
+    plan = get_mission_plan(
+        prompt,
+        latitude=args.latitude,
+        longitude=args.longitude,
+        context_overrides=context,
+    )
+    indent = None if args.indent <= 0 else args.indent
+    print(json.dumps(plan.to_dict(), indent=indent))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

models/detectors/detr.py

@@ -0,0 +1,48 @@
+import logging
+from typing import Sequence
+
+import numpy as np
+import torch
+from transformers import DetrForObjectDetection, DetrImageProcessor
+
+from models.detectors.base import DetectionResult, ObjectDetector
+
+
+class DetrDetector(ObjectDetector):
+    """Wrapper around facebook/detr-resnet-50 for mission-aligned detection."""
+
+    MODEL_NAME = "facebook/detr-resnet-50"
+
+    def __init__(self, score_threshold: float = 0.3) -> None:
+        self.name = "detr_resnet50"
+        self.score_threshold = score_threshold
+        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
+        logging.info("Loading %s onto %s", self.MODEL_NAME, self.device)
+        self.processor = DetrImageProcessor.from_pretrained(self.MODEL_NAME)
+        self.model = DetrForObjectDetection.from_pretrained(self.MODEL_NAME)
+        self.model.to(self.device)
+        self.model.eval()
+
+    def predict(self, frame: np.ndarray, queries: Sequence[str]) -> DetectionResult:
+        inputs = self.processor(images=frame, return_tensors="pt")
+        inputs = {key: value.to(self.device) for key, value in inputs.items()}
+        with torch.no_grad():
+            outputs = self.model(**inputs)
+        target_sizes = torch.tensor([frame.shape[:2]], device=self.device)
+        processed = self.processor.post_process_object_detection(
+            outputs,
+            threshold=self.score_threshold,
+            target_sizes=target_sizes,
+        )[0]
+        boxes = processed["boxes"].cpu().numpy()
+        scores = processed["scores"].cpu().tolist()
+        labels = processed["labels"].cpu().tolist()
+        label_names = [
+            self.model.config.id2label.get(int(idx), f"class_{idx}") for idx in labels
+        ]
+        return DetectionResult(
+            boxes=boxes,
+            scores=scores,
+            labels=labels,
+            label_names=label_names,
+        )

models/detectors/grounding_dino.py

@@ -0,0 +1,56 @@
+import logging
+from typing import Sequence
+
+import numpy as np
+import torch
+from transformers import GroundingDinoForObjectDetection, GroundingDinoProcessor
+
+from models.detectors.base import DetectionResult, ObjectDetector
+
+
+class GroundingDinoDetector(ObjectDetector):
+    """IDEA-Research Grounding DINO-B detector for open-vocabulary missions."""
+
+    MODEL_NAME = "IDEA-Research/grounding-dino-base"
+
+    def __init__(self, box_threshold: float = 0.35, text_threshold: float = 0.25) -> None:
+        self.name = "grounding_dino"
+        self.box_threshold = box_threshold
+        self.text_threshold = text_threshold
+        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
+        logging.info("Loading %s onto %s", self.MODEL_NAME, self.device)
+        self.processor = GroundingDinoProcessor.from_pretrained(self.MODEL_NAME)
+        self.model = GroundingDinoForObjectDetection.from_pretrained(self.MODEL_NAME)
+        self.model.to(self.device)
+        self.model.eval()
+
+    def _build_prompt(self, queries: Sequence[str]) -> str:
+        filtered = [query.strip() for query in queries if query and query.strip()]
+        if not filtered:
+            return "object."
+        return " ".join(f"{term}." for term in filtered)
+
+    def predict(self, frame: np.ndarray, queries: Sequence[str]) -> DetectionResult:
+        prompt = self._build_prompt(queries)
+        inputs = self.processor(images=frame, text=prompt, return_tensors="pt")
+        inputs = {key: value.to(self.device) for key, value in inputs.items()}
+        with torch.no_grad():
+            outputs = self.model(**inputs)
+        target_sizes = torch.tensor([frame.shape[:2]], device=self.device)
+        processed = self.processor.post_process_grounded_object_detection(
+            outputs,
+            inputs["input_ids"],
+            box_threshold=self.box_threshold,
+            text_threshold=self.text_threshold,
+            target_sizes=target_sizes,
+        )[0]
+        boxes = processed["boxes"].cpu().numpy()
+        scores = processed["scores"].cpu().tolist()
+        label_names = list(processed.get("labels") or [])
+        label_ids = list(range(len(label_names)))
+        return DetectionResult(
+            boxes=boxes,
+            scores=scores,
+            labels=label_ids,
+            label_names=label_names,
+        )

models/detectors/owlv2.py

@@ -9,7 +9,7 @@ from models.detectors.base import DetectionResult, ObjectDetector
 
 
 class Owlv2Detector(ObjectDetector):
-    MODEL_NAME = "google/owlv2-large-patch14"
+    MODEL_NAME = "google/owlv2-base-patch32"
 
     def __init__(self) -> None:
         self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
@@ -21,7 +21,7 @@ class Owlv2Detector(ObjectDetector):
         )
         self.model.to(self.device)
         self.model.eval()
-        self.name = "owlv2"
+        self.name = "owlv2_base"
 
     def predict(self, frame: np.ndarray, queries: Sequence[str]) -> DetectionResult:
         inputs = self.processor(text=queries, images=frame, return_tensors="pt")

models/detectors/yolov12_bot_sort.py

@@ -1,56 +0,0 @@
-import logging
-from typing import Sequence
-
-import numpy as np
-import torch
-from huggingface_hub import hf_hub_download
-from ultralytics import YOLO
-
-from models.detectors.base import DetectionResult, ObjectDetector
-
-
-class HuggingFaceYoloV12BotSortDetector(ObjectDetector):
-    """YOLOv12 model (BoT-SORT + ReID) hosted on Hugging Face."""
-
-    REPO_ID = "wish44165/YOLOv12-BoT-SORT-ReID"
-    WEIGHT_FILE = "MOT_yolov12n.pt"
-
-    def __init__(self, score_threshold: float = 0.3) -> None:
-        self.name = "hf_yolov12_bot_sort"
-        self.score_threshold = score_threshold
-        self.device = "cuda:0" if torch.cuda.is_available() else "cpu"
-        logging.info(
-            "Loading Hugging Face YOLOv12 BoT-SORT weights %s/%s onto %s",
-            self.REPO_ID,
-            self.WEIGHT_FILE,
-            self.device,
-        )
-        weight_path = hf_hub_download(repo_id=self.REPO_ID, filename=self.WEIGHT_FILE)
-        self.model = YOLO(weight_path)
-        self.model.to(self.device)
-        self.class_names = self.model.names
-
-    def predict(self, frame: np.ndarray, queries: Sequence[str]) -> DetectionResult:
-        device_arg = 0 if self.device.startswith("cuda") else "cpu"
-        results = self.model.predict(
-            source=frame,
-            device=device_arg,
-            conf=self.score_threshold,
-            verbose=False,
-        )
-        result = results[0]
-        boxes = result.boxes
-        if boxes is None or boxes.xyxy is None:
-            empty = np.empty((0, 4), dtype=np.float32)
-            return DetectionResult(empty, [], [], [])
-
-        xyxy = boxes.xyxy.cpu().numpy()
-        scores = boxes.conf.cpu().numpy().tolist()
-        label_ids = boxes.cls.cpu().numpy().astype(int).tolist()
-        label_names = [self.class_names.get(idx, f"class_{idx}") for idx in label_ids]
-        return DetectionResult(
-            boxes=xyxy,
-            scores=scores,
-            labels=label_ids,
-            label_names=label_names,
-        )

models/detectors/yolov8_defence.py

@@ -1,12 +0,0 @@
-from models.detectors.yolov8 import HuggingFaceYoloV8Detector
-
-
-class HuggingFaceYoloV8DefenceDetector(HuggingFaceYoloV8Detector):
-    """YOLOv8m detector fine-tuned on defence data hosted on Hugging Face."""
-
-    REPO_ID = "spencercdz/YOLOv8m_defence"
-    WEIGHT_FILE = "yolov8m_defence.pt"
-
-    def __init__(self, score_threshold: float = 0.3) -> None:
-        super().__init__(score_threshold=score_threshold)
-        self.name = "hf_yolov8_defence"

models/model_loader.py

@@ -3,18 +3,18 @@ from functools import lru_cache
 from typing import Callable, Dict, Optional
 
 from models.detectors.base import ObjectDetector
+from models.detectors.detr import DetrDetector
+from models.detectors.grounding_dino import GroundingDinoDetector
 from models.detectors.owlv2 import Owlv2Detector
 from models.detectors.yolov8 import HuggingFaceYoloV8Detector
-from models.detectors.yolov8_defence import HuggingFaceYoloV8DefenceDetector
-from models.detectors.yolov12_bot_sort import HuggingFaceYoloV12BotSortDetector
 
-DEFAULT_DETECTOR = "owlv2"
+DEFAULT_DETECTOR = "owlv2_base"
 
 _REGISTRY: Dict[str, Callable[[], ObjectDetector]] = {
-    "owlv2": Owlv2Detector,
+    "owlv2_base": Owlv2Detector,
     "hf_yolov8": HuggingFaceYoloV8Detector,
-    "hf_yolov8_defence": HuggingFaceYoloV8DefenceDetector,
-    "hf_yolov12_bot_sort": HuggingFaceYoloV12BotSortDetector,
+    "detr_resnet50": DetrDetector,
+    "grounding_dino": GroundingDinoDetector,
 }
 
 

pipeline_registry.py

@@ -0,0 +1,249 @@
+from __future__ import annotations
+
+from typing import Any, Dict, List, Tuple
+
+from mission_context import MissionContext
+
+
+PipelineSpec = Dict[str, object]
+
+
+PIPELINE_SPECS: Tuple[PipelineSpec, ...] = (
+    {
+        "id": "RGB_DAY",
+        "modalities": ("rgb",),
+        "location_types": ("urban", "suburban", "rural"),
+        "time_of_day": ("day",),
+        "huggingface": {
+            "detection": [
+                {
+                    "model_id": "facebook/detr-resnet-50",
+                    "task": "object-detection",
+                    "label": "DETR-ResNet-50",
+                    "detector_key": "detr_resnet50",
+                },
+                {
+                    "model_id": "Ultralytics/YOLOv8",
+                    "task": "object-detection",
+                    "label": "YOLOv8",
+                    "detector_key": "hf_yolov8",
+                },
+            ],
+            "segmentation": [
+                {
+                    "model_id": "facebook/mask2former-swin-base-coco",
+                    "task": "image-segmentation",
+                    "label": "Mask2Former Swin-B",
+                },
+                {
+                    "model_id": "facebook/segformer-b5-finetuned-ade-640-640",
+                    "task": "image-segmentation",
+                    "label": "SegFormer B5 ADE",
+                },
+            ],
+            "tracking": [
+                {"name": "ByteTrack", "notes": "GPU MOT for vehicles + pedestrians"},
+            ],
+            "notes": "Best balance of speed and quality for daylight RGB scenes.",
+        },
+    },
+    {
+        "id": "THERMAL_NIGHT",
+        "modalities": ("thermal",),
+        "location_types": ("urban", "industrial", "rural"),
+        "time_of_day": ("night",),
+        "huggingface": {
+            "detection": [
+                {
+                    "model_id": "Ultralytics/YOLOv8",
+                    "task": "object-detection",
+                    "label": "YOLOv8 (thermal tuned later)",
+                    "notes": "Use RGB weights until thermal finetune is ready.",
+                    "detector_key": "hf_yolov8",
+                },
+                {
+                    "model_id": "facebook/detr-resnet-50",
+                    "task": "object-detection",
+                    "label": "DETR baseline",
+                    "detector_key": "detr_resnet50",
+                },
+            ],
+            "segmentation": [
+                {
+                    "model_id": "facebook/sam-vit-base",
+                    "task": "image-segmentation",
+                    "label": "SAM ViT-B",
+                    "optional": True,
+                    "notes": "Optional: prompt SAM with thermal detections.",
+                },
+            ],
+            "tracking": [
+                {"name": "ByteTrack", "notes": "Thermal tracking excels with ByteTrack."},
+            ],
+            "notes": "Night/thermal focus; skip heavy segmentation unless SAM is needed.",
+        },
+    },
+    {
+        "id": "COASTAL_OPEN",
+        "modalities": ("rgb", "open_vocab"),
+        "location_types": ("coastal", "harbor", "bridge"),
+        "time_of_day": ("day", "night"),
+        "huggingface": {
+            "detection": [
+                {
+                    "model_id": "IDEA-Research/grounding-dino-base",
+                    "task": "zero-shot-object-detection",
+                    "label": "Grounding DINO-B",
+                    "detector_key": "grounding_dino",
+                },
+                {
+                    "model_id": "google/owlvit-base-patch32",
+                    "task": "zero-shot-object-detection",
+                    "label": "OWLv2 Base",
+                    "detector_key": "owlv2_base",
+                },
+            ],
+            "segmentation": [
+                {"model_id": "facebook/sam-vit-base", "task": "image-segmentation", "label": "SAM ViT-B"},
+                {
+                    "model_id": "facebook/segformer-b5-finetuned-ade-640-640",
+                    "task": "image-segmentation",
+                    "label": "SegFormer B5 ADE",
+                    "notes": "Great for water/sky layout.",
+                },
+            ],
+            "tracking": [
+                {"name": "ByteTrack", "notes": "Track boats, drones, aircraft."},
+            ],
+            "notes": "Strong combo for open-world discovery plus structural understanding.",
+        },
+    },
+    {
+        "id": "INDOOR",
+        "modalities": ("rgb",),
+        "location_types": ("indoor", "industrial"),
+        "time_of_day": ("day", "night"),
+        "huggingface": {
+            "detection": [
+                {
+                    "model_id": "facebook/detr-resnet-50",
+                    "task": "object-detection",
+                    "label": "DETR-ResNet-50",
+                    "detector_key": "detr_resnet50",
+                },
+                {
+                    "model_id": "Ultralytics/YOLOv8",
+                    "task": "object-detection",
+                    "label": "YOLOv8",
+                    "detector_key": "hf_yolov8",
+                },
+            ],
+            "segmentation": [
+                {
+                    "model_id": "facebook/mask2former-swin-base-coco",
+                    "task": "image-segmentation",
+                    "label": "Mask2Former Swin-B",
+                },
+                {
+                    "model_id": "facebook/segformer-b5-finetuned-ade-640-640",
+                    "task": "image-segmentation",
+                    "label": "SegFormer B5 ADE",
+                },
+            ],
+            "tracking": [
+                {"name": "DeepSORT", "notes": "Appearance-assisted tracking fits indoor scenes."},
+            ],
+            "notes": "Balanced coverage for indoor/industrial monitoring.",
+        },
+    },
+)
+
+PIPELINE_SPECS_BY_ID: Dict[str, PipelineSpec] = {spec["id"]: spec for spec in PIPELINE_SPECS}
+
+
+def get_pipeline_spec(pipeline_id: str | None) -> PipelineSpec | None:
+    if not pipeline_id:
+        return None
+    return PIPELINE_SPECS_BY_ID.get(pipeline_id)
+
+
+def huggingface_model_bindings(pipeline_id: str | None) -> Dict[str, Any]:
+    spec = get_pipeline_spec(pipeline_id)
+    if not spec:
+        return {}
+    return spec.get("huggingface") or {}
+
+
+def build_hf_inference_clients(
+    pipeline_id: str,
+    *,
+    token: str | None = None,
+) -> Dict[str, List[Dict[str, Any]]]:
+    try:
+        from huggingface_hub import InferenceClient
+    except ImportError as exc:  # pragma: no cover - dependency already required elsewhere.
+        raise RuntimeError("huggingface_hub is required to build inference clients.") from exc
+    bindings = huggingface_model_bindings(pipeline_id)
+    clients: Dict[str, List[Dict[str, Any]]] = {}
+    for task, entries in bindings.items():
+        if task == "notes" or not isinstance(entries, list):
+            continue
+        for entry in entries:
+            model_id = entry.get("model_id")
+            if not model_id:
+                continue
+            client = InferenceClient(model=model_id, token=token)
+            payload = {"model_id": model_id, "client": client, "metadata": entry}
+            clients.setdefault(task, []).append(payload)
+    return clients
+
+
+def filter_pipelines_for_context(context: MissionContext) -> List[PipelineSpec]:
+    filtered: List[PipelineSpec] = []
+    for spec in PIPELINE_SPECS:
+        reason = _availability_reason(spec, context)
+        if reason:
+            spec_copy = dict(spec)
+            spec_copy["availability_reason"] = reason
+            filtered.append(spec_copy)
+    return filtered
+
+
+def fallback_pipeline_for_context(
+    context: MissionContext,
+    available_specs: List[PipelineSpec],
+) -> PipelineSpec | None:
+    if available_specs:
+        return available_specs[0]
+    return _first_compatible_pipeline(context) or PIPELINE_SPECS[0]
+
+
+def _first_compatible_pipeline(context: MissionContext) -> PipelineSpec | None:
+    for spec in PIPELINE_SPECS:
+        if _supports_time(spec, context.time_of_day) and _supports_location(spec, context.location_type):
+            return spec
+    return None
+
+
+def _supports_time(spec: PipelineSpec, time_of_day: str | None) -> bool:
+    if not time_of_day:
+        return True
+    return time_of_day in spec["time_of_day"]
+
+
+def _supports_location(spec: PipelineSpec, location_type: str | None) -> bool:
+    if not location_type or location_type == "unknown":
+        return True
+    return location_type in spec["location_types"]
+
+
+def _availability_reason(spec: PipelineSpec, context: MissionContext) -> str | None:
+    if not _supports_time(spec, context.time_of_day):
+        return None
+    if not _supports_location(spec, context.location_type):
+        return None
+    reasons = [
+        f"time {context.time_of_day or 'any'}",
+        f"location {context.location_type or 'any'}",
+    ]
+    return ", ".join(reasons)

prompt.py

@@ -2,36 +2,175 @@
 
 from __future__ import annotations
 
-from typing import Iterable
+from typing import Any, Iterable, Mapping
 
 
 def mission_planner_system_prompt() -> str:
     return (
-        "You are a mission-planning assistant helping a vision system select which YOLO object "
-        "classes to detect. You must only reference the provided list of YOLO classes."
+        "You are an expert mission-aware perception planner for an intelligent sensing system. "
+        "Your task is to translate a high-level mission directive and contextual cues into a "
+        "prioritized set of semantic entities, objects, or events that the perception stack "
+        "should sense.\n\n"
+
+        "Think in terms of mission-relevant semantics rather than detector-specific classes. "
+        "Objects may represent physical entities, behaviors, configurations, or latent threats. "
+        "Do not constrain recommendations to any fixed taxonomy.\n\n"
+
+        "For each recommendation, reason causally about how sensing this entity reduces mission "
+        "uncertainty, enables downstream decisions, or serves as an early indicator of mission-relevant events. "
+        "Favor entities that are information-dense and mission-critical.\n\n"
+
+        "If the mission directive is abstract (e.g., tracking, monitoring, movement),"
+        "you must decompose it into concrete, observable semantic entities or events"
+        "that a perception system could plausibly sense.\n\n"
+
+        "Ground every recommendation in observable perceptual evidence (visual, spatial, or temporal), "
+        "but do not assume a specific detector architecture. "
+        "Avoid hallucinations beyond the provided mission and cues."
     )
 
 
-def mission_planner_user_prompt(mission: str, available_classes: Iterable[str], top_k: int) -> str:
-    classes_blob = ", ".join(available_classes)
+def mission_planner_user_prompt(
+    mission: str,
+    top_k: int,
+    *,
+    context: Mapping[str, Any] | None = None,
+    cues: Mapping[str, Any] | None = None,
+    pipeline_candidates: Iterable[str] | None = None,
+    coco_catalog: str | None = None,
+) -> str:
+    mission_text = mission.strip() or "N/A"
+    context_blob = _format_context_blob(context)
+    cues_blob = _format_cues_blob(cues)
+    candidate_text = (
+        ", ".join(sorted({candidate.strip() for candidate in pipeline_candidates if candidate.strip()}))
+        if pipeline_candidates
+        else "Preselected by mission context."
+    )
+    coco_text = coco_catalog.strip() if coco_catalog else "COCO classes unavailable."
+
     return (
-        f"Mission: {mission}\n"
-        f"Available YOLO classes: {classes_blob}\n"
-        "Return JSON with: mission (string) and classes (array). "
-        "Each entry needs name, score (0-1 float), rationale. "
-        f"Limit to at most {top_k} classes. Only choose names from the list."
+        f"Mission directive:\n{mission_text}\n\n"
+        f"Structured context inputs:\n{context_blob}\n\n"
+        f"Derived location/mission cues:\n{cues_blob}\n\n"
+        f"Selectable pipeline IDs: {candidate_text}\n\n"
+        "Downstream detector recognizes ONLY these COCO objects (use exact spelling):\n"
+        f"{coco_text}\n\n"
+
+        "Instructions:\n"
+        "- Interpret the mission directive and contextual cues to determine what **semantic entities, "
+        "objects, or events** are most critical for mission success.\n"
+
+        "- If the mission is abstract (e.g., tracking, monitoring, movement, surveillance), you MUST "
+        "decompose it into **concrete, observable entities or events** that a perception system could "
+        "plausibly sense. Do not leave the mission at an abstract level.\n"
+
+        "- Infer mission_type, location_type, time_of_day, and priority_level "
+        "(bounded to allowed enums). Do NOT modify intel_flag, sensor_type, or compute_mode.\n"
+
+        "- If multiple pipeline IDs are listed, select exactly ONE id from that list. "
+        "If only one ID is given, respect it.\n"
+
+        "- Never invent new pipelines or modify existing pipeline definitions.\n"
+
+        "- Recommendations are NOT limited to static objects. They may include:\n"
+        "  • dynamic entities (e.g., vehicles, aircraft, people)\n"
+        "  • behaviors or motion patterns\n"
+        "  • spatial configurations or interactions\n"
+        "  • anomalous or unexpected activity\n"
+
+        "- Each recommendation MUST correspond to a **specific, real-world, observable phenomenon**.\n"
+        "- Do NOT use placeholder names such as 'Objective', 'Target', 'Entity', or generic labels.\n"
+        "- Every entity name MUST be an exact string match to one of the provided COCO classes above. "
+        "If the mission demands a concept outside this list, decompose it into observable COCO objects.\n"
+
+        "- Provide at most "
+        f"{top_k} "
+        "recommendations, ordered by priority and expected information gain for the mission.\n"
+
+        "- Scores MUST be floats in [0, 1] and represent **relative mission importance**, "
+        "NOT detection confidence or likelihood.\n"
+
+        "- For EACH recommendation, you MUST explain:\n"
+        "  • why it matters to the mission\n"
+        "  • what observable perceptual cues (visual, spatial, or temporal) would indicate its presence\n"
+        "  • how sensing it informs downstream decision-making or reduces mission uncertainty\n\n"
+
+        "Return JSON with the following schema ONLY (no extra text):\n"
+        "{\n"
+        '  "mission": "<possibly refined mission summary>",\n'
+        '  "context": {\n'
+        '      "mission_type": "<one of surveillance|tracking|threat_detection|safety_monitoring>",\n'
+        '      "location_type": "<one of urban|suburban|rural|industrial|coastal|harbor|bridge|roadway|indoor|unknown>",\n'
+        '      "time_of_day": "<day|night|null>",\n'
+        '      "priority_level": "<routine|elevated|high|null>"\n'
+        "  },\n"
+        '  "pipeline": {"id": "<pipeline_id_from_list>", "reason": "<concise justification>"},\n'
+        '  "entities": [\n'
+        "    {\n"
+        '      "name": "<semantic entity or event>",\n'
+        '      "score": <float>,\n'
+        '      "semantic_role": "<primary_target|threat_proxy|contextual_indicator|operational_constraint>",\n'
+        '      "perceptual_cues": "<observable visual/spatial/temporal evidence>",\n'
+        '      "rationale": "<mission-grounded justification>"\n'
+        "    }\n"
+        "  ]\n"
+        "}"
     )
 
 
+
+def _format_context_blob(context: Mapping[str, Any] | None) -> str:
+    if not context:
+        return "No additional context provided."
+
+    lines = []
+    mission_type = context.get("mission_type")
+    location_type = context.get("location_type")
+    time_of_day = context.get("time_of_day")
+
+    if mission_type:
+        lines.append(f"- Mission type: {mission_type}")
+    if location_type:
+        lines.append(f"- Location type: {location_type}")
+    if time_of_day:
+        lines.append(f"- Time of day: {time_of_day}")
+
+    if not lines:
+        return "Context provided but no structured cues supplied."
+    return "\n".join(lines)
+
+
+def _format_cues_blob(cues: Mapping[str, Any] | None) -> str:
+    if not cues:
+        return "No derived cues."
+    lines = []
+    for key, value in cues.items():
+        if value is None:
+            continue
+        if isinstance(value, (list, tuple)):
+            serialized = ", ".join(str(item) for item in value if item)
+        else:
+            serialized = str(value)
+        if not serialized:
+            continue
+        lines.append(f"- {key.replace('_', ' ').capitalize()}: {serialized}")
+    if not lines:
+        return "Derived cues provided but empty."
+    return "\n".join(lines)
+
+
 def mission_summarizer_system_prompt() -> str:
     return (
-        "You are a surveillance analyst. Review structured detections aligned to a mission and "
-        "summarize actionable insights, highlighting objects of interest, temporal trends, and any "
-        "security concerns. Base conclusions solely on the provided data; if nothing is detected, "
-        "explicitly state that."
+        "You are a surveillance analyst producing brief situation reports. "
+        "Review the provided mission context and detections, then respond with a concise summary "
+        "(at most three short sentences) covering key findings only. "
+        "If no detections appear, clearly state that fact. Avoid extra commentary or formatting."
     )
 
 
 def mission_summarizer_user_prompt(payload_json: str) -> str:
-    return "Use this JSON to summarize the mission outcome:\n" f"{payload_json}"
-
+    return (
+        "Summarize this mission outcome succinctly (<=3 sentences, no bullet points):\n"
+        f"{payload_json}"
+    )

requirements.txt

@@ -10,3 +10,5 @@ scipy
 openai
 huggingface-hub
 ultralytics
+reverse_geocoder
+timezonefinder