Upload trends.py

quread/trends.py

@@ -0,0 +1,160 @@
+from __future__ import annotations
+
+import json
+from typing import Any, Dict, List, Tuple
+
+import numpy as np
+
+from .metrics import (
+    MetricThresholds,
+    MetricWeights,
+    compute_metrics_from_csv,
+)
+
+_TREND_METRIC_ALIASES = {
+    "composite": "composite_risk",
+    "composite_risk": "composite_risk",
+    "gate_error": "gate_error",
+    "readout_error": "readout_error",
+    "decoherence_risk": "decoherence_risk",
+    "fidelity": "fidelity",
+    "state_fidelity": "state_fidelity",
+    "process_fidelity": "process_fidelity",
+    "coherence_health": "coherence_health",
+}
+
+
+def _resolve_metric(metric: str) -> str:
+    key = str(metric or "composite_risk").strip().lower()
+    return _TREND_METRIC_ALIASES.get(key, "composite_risk")
+
+
+def _snapshot_timestamp(snapshot: Dict[str, Any], idx: int) -> str:
+    ts = snapshot.get("timestamp")
+    if ts is None:
+        ts = snapshot.get("ts")
+    if ts is None:
+        ts = snapshot.get("date")
+    text = str(ts).strip() if ts is not None else ""
+    if text:
+        return text
+    return f"snapshot_{idx + 1}"
+
+
+def _normalize_snapshot(snapshot: Dict[str, Any], idx: int) -> Dict[str, Any] | None:
+    if not isinstance(snapshot, dict):
+        return None
+
+    ts = _snapshot_timestamp(snapshot, idx)
+    if isinstance(snapshot.get("calibration"), dict):
+        payload = snapshot["calibration"]
+    elif isinstance(snapshot.get("qubits"), dict):
+        payload = {"qubits": snapshot["qubits"]}
+    else:
+        qubit_like = {
+            str(k): v
+            for k, v in snapshot.items()
+            if str(k).strip().isdigit() and isinstance(v, dict)
+        }
+        if not qubit_like:
+            return None
+        payload = {"qubits": qubit_like}
+
+    return {
+        "timestamp": ts,
+        "calibration_json": json.dumps(payload),
+    }
+
+
+def parse_calibration_snapshots_text(text: str) -> Tuple[List[Dict[str, Any]], Dict[str, Any]]:
+    raw = str(text or "").strip()
+    if not raw:
+        return [], {"format": "empty", "parsed": 0, "skipped": 0}
+
+    snapshots: List[Dict[str, Any]] = []
+    skipped = 0
+    fmt = "unknown"
+
+    try:
+        parsed = json.loads(raw)
+        if isinstance(parsed, dict) and isinstance(parsed.get("snapshots"), list):
+            iterable = parsed["snapshots"]
+            fmt = "json:snapshots"
+        elif isinstance(parsed, list):
+            iterable = parsed
+            fmt = "json:list"
+        elif isinstance(parsed, dict):
+            iterable = [parsed]
+            fmt = "json:single"
+        else:
+            iterable = []
+            skipped += 1
+    except Exception:
+        fmt = "jsonl"
+        iterable = []
+        for line in raw.splitlines():
+            chunk = line.strip()
+            if not chunk:
+                continue
+            try:
+                iterable.append(json.loads(chunk))
+            except Exception:
+                skipped += 1
+
+    for idx, snap in enumerate(iterable):
+        normalized = _normalize_snapshot(snap, idx)
+        if normalized is None:
+            skipped += 1
+            continue
+        snapshots.append(normalized)
+
+    return snapshots, {"format": fmt, "parsed": len(snapshots), "skipped": int(skipped)}
+
+
+def compute_metric_trends(
+    csv_text: str,
+    n_qubits: int,
+    snapshots_text: str,
+    *,
+    metric: str = "composite_risk",
+    state_vector: np.ndarray | None = None,
+    weights: MetricWeights | None = None,
+    thresholds: MetricThresholds | None = None,
+) -> Tuple[np.ndarray, List[str], List[Dict[str, float]], Dict[str, Any]]:
+    metric_key = _resolve_metric(metric)
+    snapshots, meta = parse_calibration_snapshots_text(snapshots_text)
+    if not snapshots:
+        raise ValueError("No valid calibration snapshots found.")
+
+    series = []
+    labels: List[str] = []
+    for snap in snapshots:
+        labels.append(str(snap["timestamp"]))
+        metrics, _ = compute_metrics_from_csv(
+            csv_text,
+            int(n_qubits),
+            calibration_json=str(snap["calibration_json"]),
+            state_vector=state_vector,
+            weights=weights,
+            thresholds=thresholds,
+        )
+        series.append(np.asarray(metrics[metric_key], dtype=float))
+
+    arr = np.vstack(series)
+    latest = arr[-1]
+    baseline = arr[0]
+
+    ranking: List[Dict[str, float]] = []
+    for q in range(int(n_qubits)):
+        ranking.append(
+            {
+                "qubit": float(q),
+                "latest": float(latest[q]),
+                "baseline": float(baseline[q]),
+                "delta": float(latest[q] - baseline[q]),
+            }
+        )
+    ranking.sort(key=lambda r: r["latest"], reverse=True)
+    meta["metric"] = metric_key
+    meta["points"] = int(arr.shape[0])
+    return arr, labels, ranking, meta