satquery_agent/analytics.py

"""Process-local, metadata-only research analytics for SatQuery.

No upload content, query text, filename, filesystem path, answer text, or model
output is retained here. The store intentionally resets when the backend does.
"""

from __future__ import annotations

import importlib.metadata
import os
import platform
import re
import threading
import time
from collections import Counter, deque
from datetime import datetime, timezone
from typing import Any, Deque, Dict, Iterable, List, Optional

from .models import (
    AgentResponse,
    AnalyticsCache,
    AnalyticsCapability,
    AnalyticsDataset,
    AnalyticsExecution,
    AnalyticsPlatform,
    AnalyticsReports,
    AnalyticsResponse,
    AnalyticsSummary,
    AnalyticsSVE,
    AnalyticsTTP,
    AnalyticsTool,
    AnalyticsTraceStep,
    AnalyticsWorkflowMetric,
    ChangeAnalysisResponse,
    CrossModalAnalysisResponse,
    ImplementationStatus,
    ScientificTransparency,
)


MAX_HISTORY = 50
PROCESS_STARTED = time.monotonic()


def utc_now() -> str:
    return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")


def _version(package: str) -> Optional[str]:
    try:
        return importlib.metadata.version(package)
    except importlib.metadata.PackageNotFoundError:
        return None


def _preview_count(value: Any) -> int:
    found = set()

    def visit(item: Any) -> None:
        if isinstance(item, dict):
            for nested in item.values():
                visit(nested)
        elif isinstance(item, list):
            for nested in item:
                visit(nested)
        elif isinstance(item, str) and item.startswith("/api/agent/previews/"):
            found.add(item)

    visit(value)
    return len(found)


def _device_from_response(response: AgentResponse) -> Optional[str]:
    if response.sve_result is not None and response.sve_result.device:
        return response.sve_result.device
    if response.grounding_result is not None:
        return response.grounding_result.device
    if response.caption_details is not None:
        return response.caption_details.device
    return None


_SAFE_PARAMETER_KEYS = {
    "alignment_level", "connectivity", "detections", "device", "filesystem_paths_exposed",
    "format", "ground_truth_claimed", "image_representation", "language_model",
    "language_model_used", "mask_refinement", "max_analysis_dimension",
    "maximum_detections", "minimum_region_pixels", "model_input_height",
    "model_input_width", "normalization", "registration", "registration_performed",
    "reprojection", "resampling", "reused", "selected_bands", "template_generated_summary",
    "eligible", "reason", "checkpoint_verified", "width", "height", "binary",
    "changed_pixels", "region_count", "iou", "agreement_percentage", "primary_mask",
    "supporting_mask", "mask_averaging", "reused_model", "exact_source_bytes",
    "checksum_verified", "cache", "first_cache", "second_cache", "top_labels",
    "calibrated", "candidate_count", "selected_candidate_index", "consistency_state",
    "support_state", "input", "transform", "comparison", "influence_applied",
}
_LOCAL_PATH = re.compile(r"(?:/Users/|/home/|/root/|[A-Za-z]:\\)[^\s,;]+")


def _safe_text(value: Optional[str]) -> Optional[str]:
    if not value:
        return None
    return _LOCAL_PATH.sub("[redacted local path]", value)[:600]


def _safe_trace_parameters(parameters: Dict[str, Any]) -> Dict[str, Any]:
    output: Dict[str, Any] = {}
    for key, value in parameters.items():
        if key not in _SAFE_PARAMETER_KEYS:
            continue
        if isinstance(value, str):
            output[key] = _safe_text(value)
        elif value is None or isinstance(value, (bool, int, float)):
            output[key] = value
        elif isinstance(value, list) and all(isinstance(item, (str, int, float, bool)) for item in value):
            output[key] = [(_safe_text(item) if isinstance(item, str) else item) for item in value[:12]]
    return output


def _safe_trace(steps: Iterable[Any]) -> List[AnalyticsTraceStep]:
    return [
        AnalyticsTraceStep(
            tool=str(step.tool),
            status=getattr(step.status, "value", str(step.status)),
            duration_ms=max(0, int(step.duration_ms)),
            parameters=_safe_trace_parameters(dict(getattr(step, "parameters", {}) or {})),
        )
        for step in steps
    ]


class AnalyticsStore:
    def __init__(self) -> None:
        self._history: Deque[AnalyticsExecution] = deque(maxlen=MAX_HISTORY)
        self._total_executions = 0
        self._successful_executions = 0
        self._report_requests = 0
        self._report_artifacts = 0
        self._report_formats: Counter[str] = Counter()
        self._lock = threading.RLock()

    def record(self, execution: AnalyticsExecution) -> None:
        with self._lock:
            self._history.appendleft(execution.model_copy(deep=True))
            self._total_executions += 1
            if execution.status in {"success", "partial"}:
                self._successful_executions += 1

    def record_report(self, request_id: str, formats: Iterable[str]) -> None:
        normalized = [getattr(value, "value", str(value)) for value in formats]
        with self._lock:
            self._report_requests += 1
            self._report_artifacts += len(normalized)
            self._report_formats.update(normalized)
            for index, execution in enumerate(self._history):
                if execution.request_id == request_id:
                    self._history[index] = execution.model_copy(update={"report_generated": True})
                    break

    def snapshot(self) -> Dict[str, Any]:
        with self._lock:
            return {
                "history": [item.model_copy(deep=True) for item in self._history],
                "total_executions": self._total_executions,
                "successful_executions": self._successful_executions,
                "report_requests": self._report_requests,
                "report_artifacts": self._report_artifacts,
                "report_formats": dict(self._report_formats),
            }

    def clear(self) -> None:
        with self._lock:
            self._history.clear()
            self._total_executions = 0
            self._successful_executions = 0
            self._report_requests = 0
            self._report_artifacts = 0
            self._report_formats.clear()


ANALYTICS_STORE = AnalyticsStore()


def record_agent_response(
    response: AgentResponse,
    *,
    started_at: str,
    primary_modality: Optional[str],
    secondary_modality: Optional[str],
) -> None:
    ANALYTICS_STORE.record(
        AnalyticsExecution(
            request_id=response.request_id,
            started_at=started_at,
            completed_at=utc_now(),
            task=response.task.value,
            input_mode=response.execution.input_mode.value,
            primary_modality=primary_modality,
            secondary_modality=secondary_modality,
            status=response.status.value,
            selected_tools=list(response.execution.selected_tools),
            duration_ms=response.execution.duration_ms,
            warning_count=len(response.warnings),
            output_count=_preview_count(response.model_dump(mode="json")),
            cache_status="hit" if response.cache and response.cache.cached else "fresh",
            device=_device_from_response(response),
            selection_reason=_safe_text(response.execution.selection_reason),
            confidence_level=response.confidence.level.value,
            confidence_reason=_safe_text(response.confidence.reason),
            warnings=[safe for warning in response.warnings if (safe := _safe_text(warning))],
            trace=_safe_trace(response.execution.steps),
        )
    )


def record_change_response(response: ChangeAnalysisResponse, *, started_at: str, modality: str) -> None:
    ANALYTICS_STORE.record(
        AnalyticsExecution(
            request_id=response.request_id,
            started_at=started_at,
            completed_at=utc_now(),
            task="change_analysis",
            input_mode=response.execution.input_mode.value,
            primary_modality=modality,
            secondary_modality=modality,
            status=response.status.value,
            selected_tools=list(response.execution.selected_tools),
            duration_ms=response.execution.duration_ms,
            warning_count=len(response.warnings),
            output_count=_preview_count(response.model_dump(mode="json")),
            cache_status="not_requested",
            selection_reason=_safe_text(response.execution.selection_reason),
            warnings=[safe for warning in response.warnings if (safe := _safe_text(warning))],
            trace=_safe_trace(response.execution.steps),
        )
    )


def record_cross_modal_response(
    response: CrossModalAnalysisResponse,
    *,
    started_at: str,
    optical_modality: str,
    sar_modality: str,
) -> None:
    ANALYTICS_STORE.record(
        AnalyticsExecution(
            request_id=response.request_id,
            started_at=started_at,
            completed_at=utc_now(),
            task="cross_modal_analysis",
            input_mode=response.execution.input_mode.value,
            primary_modality=optical_modality,
            secondary_modality=sar_modality,
            status=response.result.status.value,
            selected_tools=list(response.execution.selected_tools),
            duration_ms=response.execution.duration_ms,
            warning_count=len(response.result.warnings),
            output_count=_preview_count(response.model_dump(mode="json")),
            cache_status="not_requested",
            selection_reason=_safe_text(response.execution.selection_reason),
            confidence_level=response.result.confidence.level.value,
            confidence_reason=_safe_text(response.result.confidence.reason),
            warnings=[safe for warning in response.result.warnings if (safe := _safe_text(warning))],
            trace=_safe_trace(response.execution.steps),
        )
    )


def record_report_generation(request_id: str, formats: Iterable[Any]) -> None:
    ANALYTICS_STORE.record_report(request_id, formats)


def _hardware() -> str:
    try:
        import torch

        if torch.cuda.is_available():
            return "CUDA"
        mps = getattr(torch.backends, "mps", None)
        if mps is not None and mps.is_available():
            return "Apple MPS"
    except Exception:
        pass
    return "CPU"


def _workflow_metrics(history: List[AnalyticsExecution]) -> List[AnalyticsWorkflowMetric]:
    task_order = [
        "captioning", "vqa", "grounding", "change_analysis", "change_description",
        "change_vqa", "cross_modal_analysis",
    ]
    output: List[AnalyticsWorkflowMetric] = []
    for task in task_order:
        runs = [item for item in history if item.task == task]
        last = runs[0] if runs else None
        output.append(
            AnalyticsWorkflowMetric(
                task=task,
                executions=len(runs),
                successful_executions=sum(item.status in {"success", "partial"} for item in runs),
                average_runtime_ms=(round(sum(item.duration_ms for item in runs) / len(runs), 1) if runs else None),
                last_runtime_ms=last.duration_ms if last else None,
                last_device=last.device if last else None,
                last_completed_at=last.completed_at if last else None,
            )
        )
    return output


def _datasets() -> List[AnalyticsDataset]:
    return [
        AnalyticsDataset(
            name="LEVIR-CD",
            usage_status="Active Model Training Provenance",
            used_by=["ttp_change_detector"],
            purpose="Building-change training provenance for the official TTP epoch_260 checkpoint.",
            source="https://github.com/KyanChen/TTP",
            note="Training provenance is disclosed as a domain limitation; GeoVision does not claim universal change-detection accuracy.",
        ),
        AnalyticsDataset(
            name="RSICD",
            usage_status="Active",
            used_by=["rs_captioner"],
            purpose="Remote-sensing adaptation provenance for optical image captioning.",
            source="https://huggingface.co/Gurveer05/blip-image-captioning-base-rscid-finetuned",
            note="The connected BLIP checkpoint declares RSICD adaptation; no local sample count is asserted.",
        ),
        AnalyticsDataset(
            name="SAR-to-optical training corpus",
            usage_status="Undeclared",
            used_by=["pix2pix_reconstruction", "sarfusionformer_analysis"],
            purpose="Training provenance for the checked-in SAR reconstruction checkpoints.",
            note="The exact training dataset is not declared in the checked-in runtime metadata, so SEN12MS-CR is not claimed.",
        ),
        AnalyticsDataset(
            name="RSVQA",
            usage_status="Evaluation Candidate / Not Used",
            used_by=[],
            purpose="Potential future remote-sensing VQA evaluation or adaptation.",
            note="Not connected and not used to fine-tune the current deterministic VQA specialist.",
        ),
        AnalyticsDataset(
            name="BigEarthNet.txt / BigEarthNet v2 Lithuania Summer",
            usage_status="Active Model Adaptation Provenance",
            used_by=["satquery_vision_encoder_v1"],
            purpose="Image-text adaptation provenance for scene-level Earth-observation embeddings.",
            sample_count=4008,
            note="The recorded training split contains 4,008 pairs; validation and test contain 2,291 and 2,053 pairs. Retrieval improved over generic OpenCLIP on validation but remains low in absolute terms.",
        ),
        AnalyticsDataset(
            name="Grounding DINO model provenance",
            usage_status="Model Provenance",
            used_by=["rs_grounder"],
            purpose="Official zero-shot open-vocabulary detector checkpoint provenance.",
            source="https://huggingface.co/IDEA-Research/grounding-dino-tiny",
            note="This is model provenance, not a GeoVision project dataset or remote-sensing fine-tuning claim.",
        ),
    ]


def _capabilities() -> List[AnalyticsCapability]:
    rows = [
        ("Optical / multispectral ingestion", "Available", "Validated PNG, JPEG, TIFF, and GeoTIFF ingestion."),
        ("SAR ingestion", "Available", "Validated SAR metadata and preview path."),
        ("Raster metadata", "Available", "Rasterio-backed CRS, transform, bounds, dtype, and nodata metadata."),
        ("Optical captioning", "Available", "RSICD-adapted BLIP specialist is registered."),
        ("Controlled optical VQA", "Available", "Deterministic evidence-grounded question taxonomy."),
        ("Text-guided grounding boxes", "Available", "Local Grounding DINO box and alignment-score output."),
        ("Grounding masks", "Not Implemented", "Mask refinement is explicitly not connected."),
        ("Bi-temporal change analysis", "Available", "Normalized difference, mask, morphology, regions, and overlays."),
        ("TTP hybrid optical change analysis", "Available when CUDA service is ready", "TTP learned mask is primary; deterministic mask remains independent supporting evidence with explicit fallback."),
        ("Change VQA", "Available", "Controlled answers use measured change products only."),
        ("Optical–SAR joint analysis", "Available", "Exactly aligned deterministic evidence fusion."),
        ("Automatic image registration", "Not Implemented", "Alignment-required is returned without silent registration."),
        ("SAR captioning", "Unsupported", "The optical captioner does not accept SAR imagery."),
        ("Single-image SAR VQA", "Unsupported", "The controlled single-image VQA taxonomy is optical-only."),
        ("Deterministic routing", "Available", "Explicit task and specialist selection rules."),
        ("Evidence and execution traces", "Available", "Safe previews, regions, warnings, confidence rationale, and ordered steps."),
        ("Remote-sensing scene embeddings", "Available when SVE is ready", "SatQuery Vision Encoder v1 supplies scene priors, similarity, consistency, and routing support without replacing specialist models."),
        ("Mission reports", "Available", "Backend-authoritative PDF, JSON, CSV, and ZIP artifacts."),
        ("Offline demo workflow", "Available", "Environment-gated local sample workflow; current activation is reported separately."),
    ]
    return [AnalyticsCapability(name=name, status=status, evidence=evidence) for name, status, evidence in rows]


def build_analytics() -> AnalyticsResponse:
    from .compliance import compliance_summary
    from .demo import demo_manifest
    from .registry import public_tool_registry
    from .reporting import ARTIFACT_STORE, MISSION_STORE
    from .specialists.captioner import get_captioner
    from .specialists.grounder import get_grounder
    from .specialists.ttp_change import TTP_CLIENT, TTPClientError, ttp_enabled
    from .services.sve_service import get_sve_service

    state = ANALYTICS_STORE.snapshot()
    history: List[AnalyticsExecution] = state["history"]
    compliance = compliance_summary()
    registry = public_tool_registry()
    cache = MISSION_STORE.snapshot()
    artifacts = ARTIFACT_STORE.snapshot()
    caption_health = get_captioner().health()
    grounder_health = get_grounder().health()
    sve_service = get_sve_service()
    sve_health = sve_service.health()
    lifecycle = {"rs_captioner": caption_health, "rs_grounder": grounder_health, "satquery_vision_encoder_v1": sve_health}
    failed_specialists = [name for name, health in lifecycle.items() if health.status == "failed"]
    demo = demo_manifest()
    ttp_health: Dict[str, Any] = {}
    if ttp_enabled():
        try:
            ttp_health = TTP_CLIENT.health()
            ttp_status = str(ttp_health.get("status", "unavailable"))
        except TTPClientError:
            ttp_status = "unavailable"
    else:
        ttp_status = "disabled"
    offline_requirements: List[str] = []
    if caption_health.status != "ready":
        offline_requirements.append("Load and verify the RSICD caption checkpoint before disconnecting if captioning will be used.")
    if grounder_health.status != "ready":
        offline_requirements.append("Load and verify the Grounding DINO checkpoint before disconnecting if grounding will be used.")
    if sve_health.status != "ready":
        offline_requirements.append("Load and verify SatQuery Vision Encoder v1 and its OpenCLIP backbone before disconnecting if scene embedding evidence will be used.")
    if not demo.enabled:
        offline_requirements.append("Set SATQUERY_DEMO_MODE=true to expose approved local demo samples.")
    offline_ready = not offline_requirements
    if ttp_enabled() and ttp_status != "ready":
        offline_requirements.append("Start, verify, and warm the persistent TTP CUDA service before the hybrid presentation workflow.")
        offline_ready = False

    last_by_tool: Dict[str, AnalyticsExecution] = {}
    tool_aliases = {"deterministic_change_analysis": "bitemporal_change_analyzer"}
    for execution in history:
        for tool_id in execution.selected_tools:
            last_by_tool.setdefault(tool_aliases.get(tool_id, tool_id), execution)
    tools = []
    for tool in registry:
        health = lifecycle.get(tool.id)
        last = last_by_tool.get(tool.id)
        tools.append(
            AnalyticsTool(
                id=tool.id,
                display_name=tool.display_name,
                implementation_status=tool.status.value,
                lifecycle_status=health.status if health else ("registered" if tool.status == ImplementationStatus.AVAILABLE else "not_implemented"),
                device=health.device if health else (last.device if last else None),
                last_runtime_ms=last.duration_ms if last else None,
                last_completed_at=last.completed_at if last else None,
                method_type=tool.method_type,
                checkpoint=tool.checkpoint,
                adaptation_dataset=tool.adaptation_dataset,
                remote_sensing_adapted=tool.remote_sensing_adapted,
                service_path=tool.service_path,
            )
        )

    hits = cache["hits"]
    misses = cache["misses"]
    lookups = hits + misses
    ttp_metrics = TTP_CLIENT.metrics()
    fallback_count = sum(
        1 for execution in history for step in execution.trace if step.tool == "deterministic_fallback"
    )
    ious = [
        float(step.parameters["iou"])
        for execution in history for step in execution.trace
        if step.tool == "mask_comparison" and isinstance(step.parameters.get("iou"), (int, float))
    ]
    runtime_versions = {
        name: _version(package)
        for name, package in {
            "fastapi": "fastapi", "torch": "torch", "torchvision": "torchvision",
            "transformers": "transformers", "rasterio": "rasterio",
            "open_clip": "open-clip-torch", "timm": "timm",
        }.items()
    }
    return AnalyticsResponse(
        generated_at=utc_now(),
        platform=AnalyticsPlatform(
            backend_status="degraded" if failed_specialists else "healthy",
            uptime_seconds=max(0, int(time.monotonic() - PROCESS_STARTED)),
            python_version=platform.python_version(),
            operating_system=platform.system(),
            architecture=platform.machine(),
            process_memory_mb=None,
            runtime_versions=runtime_versions,
            hardware_acceleration=_hardware(),
            demo_mode=demo.enabled,
            offline_ready=offline_ready,
            offline_readiness_requirements=offline_requirements,
        ),
        summary=AnalyticsSummary(
            total_executions_current_process=state["total_executions"],
            successful_executions_current_process=state["successful_executions"],
            registered_tools=len(registry),
            available_tools=sum(tool.status == ImplementationStatus.AVAILABLE for tool in registry),
            mandatory_satisfied=compliance.mandatory_satisfied,
            mandatory_total=compliance.mandatory_total,
            cache_hits_current_process=hits,
            cache_misses_current_process=misses,
            report_artifacts_generated_current_process=state["report_artifacts"],
        ),
        cache=AnalyticsCache(
            stored_results=cache["stored_results"],
            max_results=cache["max_results"],
            ttl_seconds=cache["ttl_seconds"],
            hits=hits,
            misses=misses,
            hit_rate_percent=round(hits * 100 / lookups, 1) if lookups else None,
        ),
        reports=AnalyticsReports(
            requests_generated_current_process=state["report_requests"],
            artifacts_generated_current_process=state["report_artifacts"],
            artifacts_currently_available=artifacts["stored_artifacts"],
            formats=state["report_formats"],
        ),
        tools=tools,
        datasets=_datasets(),
        capabilities=_capabilities(),
        workflow_metrics=_workflow_metrics(history),
        recent_executions=history,
        last_execution_trace=history[0].trace if history else [],
        scientific_transparency=ScientificTransparency(
            metric_source="Current backend process, public registry, compliance matrix, and completed SatQuery execution summaries.",
            history_retention="Metadata-only, newest first, maximum 50 executions; cleared on backend restart.",
            unavailable_value_policy="Unavailable or unmeasured values are null or explicitly labelled; zero is used only for measured counters.",
            caveats=[
                "No benchmark accuracy, dataset sample count, model quality score, or unexecuted runtime is inferred.",
                "Process memory is null because a safe current-RSS provider is not installed.",
                "Tool availability describes the registered implementation; lifecycle status separately reports lazy model state.",
                "History excludes uploaded content, filenames, queries, answers, hashes, paths, credentials, and environment values.",
            ],
        ),
        sve=AnalyticsSVE(**sve_service.metrics()),
        ttp=AnalyticsTTP(
            enabled=ttp_enabled(),
            service_status=ttp_status,
            model_load_count=int(ttp_health.get("model_load_count", 0)),
            model_reuse_count=int(ttp_metrics["model_reuse_count"] or 0),
            inference_count=int(ttp_metrics["inference_count"] or 0),
            failure_count=int(ttp_metrics["failure_count"] or 0),
            timeout_count=int(ttp_metrics["timeout_count"] or 0),
            oom_count=int(ttp_metrics["oom_count"] or 0),
            fallback_count=fallback_count,
            average_runtime_ms=ttp_metrics["average_runtime_ms"],
            average_mask_iou=round(sum(ious) / len(ious), 6) if ious else None,
        ),
    )