"""Pipeline orchestrator: logical processing services over a job queue. Services (spec §5): label-generator, anomaly-detector, regression-tracker, grader, quality-validator, rare-event-detector, scenario-miner, triage-engine, review-service, metrics-service — modular workers inside one app, surfaced to the UI as logical services with live status. Flow: Ingestion -> Label Generation -> Queue -> Evaluation workers -> Validation workers -> Grading workers -> Triage -> (Auto-grade | HITL). """ from __future__ import annotations import math import threading import time from typing import Dict, List, Optional import numpy as np from sensorflow.evaluation import synthetic from sensorflow.evaluation.detectors import AnomalyEnsemble from sensorflow.evaluation.graders import dataset_grader_statistics, grade_annotation from sensorflow.evaluation.process_units import ProcessMeter from sensorflow.evaluation.queue import EventQueue, make_queue from sensorflow.evaluation.rare_events import detect_rare_events from sensorflow.evaluation.records import ( Alert, Annotation, AnomalyDetection, DatasetLineage, EvalStore, Model, ModelEvaluation, RelabelingAction, ReviewResolution, ReviewTask, HumanReview, TrackingEvidence, TrainingDataset, TrainingJob, get_store, new_id, ) from sensorflow.evaluation.regression import compare_runs, find_affected_classes from sensorflow.evaluation.triage import QualityPolicy, load_policy, triage_annotation from sensorflow.evaluation.validation import validate_annotation from sensorflow.evaluation import reporting SERVICES = [ "label-generator", "anomaly-detector", "regression-tracker", "grader", "quality-validator", "rare-event-detector", "scenario-miner", "triage-engine", "review-service", "metrics-service", ] DEFAULT_ANOMALY_CONFIG: Dict = { "imbalance": {"method": "class_weights", "minority_boost": 2.0}, "detectors": { "knn": {"enabled": True, "k": 10}, "lof": {"enabled": True, "n_neighbors": 20}, "isolation_forest": {"enabled": True, "n_estimators": 100}, "ocsvm": {"enabled": True, "nu": 0.05}, "dbscan": {"enabled": True, "eps": 1.2, "min_samples": 5}, }, "deep": { "autoencoder": {"enabled": True, "latent_dim": 4, "epochs": 60}, "vae": {"enabled": True, "latent_dim": 5}, "gan": {"enabled": False}, "reconstruction_threshold": 0.9, }, "advanced": { "few_shot": {"enabled": True, "support_per_class": 20}, "ensemble_strategy": "weighted_average", "decision_threshold": 0.9, }, } BATCH_SIZE = 64 BATCH_PAUSE_S = 0.015 # keeps the live queue/SSE view meaningful without slowing tests much class LabelEvalPipeline: """Singleton orchestrator holding queue, services and live run state.""" def __init__(self, store: Optional[EvalStore] = None, queue: Optional[EventQueue] = None): self.store = store or get_store() self.queue = queue or make_queue("memory") self.lock = threading.RLock() self.services: Dict[str, Dict] = { s: {"state": "IDLE", "processed": 0, "total": 0, "detail": "", "process_units": 0} for s in SERVICES } self.running = False self.stage = "idle" self.last_run_id: Optional[str] = None self.active_dataset: Optional[str] = None self.anomaly_config: Dict = dict(DEFAULT_ANOMALY_CONFIG) self.regression_alert = False self._thread: Optional[threading.Thread] = None self._train_threads: Dict[str, threading.Thread] = {} self._score_context: Optional[Dict] = None # for re-scoring corrected labels # restore persisted light state meta = self.store.meta self.active_dataset = meta.get("active_dataset") self.last_run_id = meta.get("last_run_id") self.anomaly_config = meta.get("anomaly_config", dict(DEFAULT_ANOMALY_CONFIG)) self.regression_alert = bool(meta.get("regression_alert", False)) # ------------------------------------------------------------ service status def _svc(self, name: str, state: Optional[str] = None, processed: Optional[int] = None, total: Optional[int] = None, detail: Optional[str] = None, add_units: int = 0) -> None: with self.lock: svc = self.services[name] if state is not None: svc["state"] = state if processed is not None: svc["processed"] = processed if total is not None: svc["total"] = total if detail is not None: svc["detail"] = detail svc["process_units"] += add_units def service_status(self) -> List[Dict]: with self.lock: return [{"service": s, **self.services[s]} for s in SERVICES] # ------------------------------------------------------------ run orchestration def run(self, dataset_id: str, background: bool = True, policy: Optional[QualityPolicy] = None) -> str: with self.lock: if self.running: raise RuntimeError("A pipeline run is already in progress") self.running = True run_id = new_id("run") self.last_run_id = run_id self.active_dataset = dataset_id if background: self._thread = threading.Thread(target=self._run_safe, args=(dataset_id, run_id, policy), daemon=True) self._thread.start() else: self._run_safe(dataset_id, run_id, policy) return run_id def _run_safe(self, dataset_id: str, run_id: str, policy: Optional[QualityPolicy]) -> None: try: self._execute(dataset_id, run_id, policy or load_policy()) except Exception as exc: # surface failures as service state, not silent death self.stage = "failed" for s in SERVICES: if self.services[s]["state"] == "RUNNING": self._svc(s, state="FAILED", detail=str(exc)[:200]) self.store.audit("pipeline_failed", "Run", run_id, str(exc)[:400]) finally: with self.lock: self.running = False self.store.meta.update({ "active_dataset": self.active_dataset, "last_run_id": self.last_run_id, "anomaly_config": self.anomaly_config, "regression_alert": self.regression_alert, }) self.store.save() def _execute(self, dataset_id: str, run_id: str, policy: QualityPolicy) -> None: store = self.store meter = ProcessMeter(store, run_id) dataset = store.get("datasets", dataset_id) if dataset is None: raise ValueError(f"Unknown dataset {dataset_id}") self.queue.reset() # ---- 1. ingestion self.stage = "ingestion" frames = sorted(store.where("frames", dataset_id=dataset_id), key=lambda f: f.index) units = meter.record("ingestion", len(frames)) self._svc("label-generator", state="RUNNING", processed=0, total=len(frames), detail="ingesting sensor frames", add_units=units) # ---- 2. label generation (labels are hypotheses; skip if already labeled) self.stage = "label_generation" annotations = store.where("annotations", dataset_id=dataset_id) model_version = self._active_model_version() if not annotations: annotations = synthetic.generate_labels(store, dataset, model_version=model_version) else: model_version = annotations[0].model_version # Fresh evaluation: reset transient statuses of non-human-reviewed labels. for a in annotations: if a.source == "auto": a.status = "PENDING" store.put("annotations", a) units = meter.record("label_generation", len(annotations)) self._svc("label-generator", state="HEALTHY", processed=len(frames), total=len(frames), detail=f"{len(annotations)} labels by {model_version}", add_units=units) # ---- 3. enqueue for evaluation self.stage = "queueing" for ann in annotations: self.queue.publish("evaluation", {"annotation_id": ann.annotation_id}) frames_by_id = {f.frame_id: f for f in frames} ann_by_id = {a.annotation_id: a for a in annotations} # ---- 4. anomaly detection (fit once on the full population) self.stage = "anomaly_detection" self._svc("anomaly-detector", state="RUNNING", processed=0, total=len(annotations), detail="fitting detector ensemble") X, ids = synthetic.annotation_features(store, dataset_id) supervision = np.array([1.0 if ann_by_id[i].injected_errors else 0.0 for i in ids]) if len(ids) else None ensemble = AnomalyEnsemble(self.anomaly_config, seed=dataset.seed) scores, raw, norm = ensemble.run(X, supervision=supervision) strategy = ensemble.strategy threshold = 0.5 if strategy == "majority_vote" else ensemble.threshold anomaly_by_id: Dict[str, AnomalyDetection] = {} for j, aid in enumerate(ids): det = AnomalyDetection( annotation_id=aid, score=round(float(scores[j]), 4), is_anomaly=bool(scores[j] >= threshold), detector_scores={k: round(float(v[j]), 4) for k, v in raw.items()}, normalized_scores={k: round(float(v[j]), 4) for k, v in norm.items()}, ensemble_strategy=strategy, ensemble_score=round(float(scores[j]), 4), decision_threshold=threshold, ) anomaly_by_id[aid] = det store.put("anomalies", det) self._score_context = { "scaler": ensemble.scaler, "detectors": ensemble.detectors, "raw": raw, "strategy": strategy, "threshold": threshold, } units = meter.record("anomaly_detection", len(ids)) self._svc("anomaly-detector", state="HEALTHY", processed=len(ids), total=len(ids), detail=f"{sum(1 for d in anomaly_by_id.values() if d.is_anomaly)} anomalies " f"({strategy})", add_units=units) # ---- 5. queue-driven evaluation -> validation -> grading workers self.stage = "validation" self._svc("quality-validator", state="RUNNING", processed=0, total=len(annotations)) self._svc("grader", state="RUNNING", processed=0, total=len(annotations)) points_cache: Dict[str, np.ndarray] = {} tracking_by_id = self._tracking_evidence(annotations, frames_by_id) processed_v = processed_g = 0 while True: batch = self.queue.consume("evaluation", BATCH_SIZE) if not batch: break for msg in batch: self.queue.publish("validation", msg) self.queue.ack("evaluation", len(batch)) vbatch = self.queue.consume("validation", BATCH_SIZE) for msg in vbatch: ann = ann_by_id[msg["annotation_id"]] frame = frames_by_id[ann.frame_id] if ann.frame_id not in points_cache: points_cache[ann.frame_id] = synthetic.frame_points(store, frame) validate_annotation(store, ann, frame, points=points_cache[ann.frame_id]) processed_v += 1 self.queue.publish("grading", msg) self.queue.ack("validation", len(vbatch)) self._svc("quality-validator", processed=processed_v) gbatch = self.queue.consume("grading", BATCH_SIZE) for msg in gbatch: ann = ann_by_id[msg["annotation_id"]] frame = frames_by_id[ann.frame_id] tev = tracking_by_id.get(ann.annotation_id) grade_annotation(store, ann, frame, temporal_quality=tev.track_quality if tev else None, seed=dataset.seed) processed_g += 1 self.queue.publish("triage", msg) self.queue.ack("grading", len(gbatch)) self._svc("grader", processed=processed_g) time.sleep(BATCH_PAUSE_S) units = meter.record("validation", processed_v) self._svc("quality-validator", state="HEALTHY", detail="geometric validation complete", add_units=units) units = meter.record("grading", processed_g) stats = dataset_grader_statistics(store, dataset_id) self._svc("grader", state="HEALTHY", detail=f"kappa(C)={stats.get('cohen_kappa', 0):.2f} fleiss={stats.get('fleiss_kappa', 0):.2f} " f"alpha={stats.get('krippendorff_alpha', 0):.2f}", add_units=units) self.store.meta.setdefault("grader_stats", {})[dataset_id] = stats # ---- 6. rare events + scenario mining self.stage = "rare_events" self._svc("rare-event-detector", state="RUNNING", detail="mining scenarios") self._svc("scenario-miner", state="RUNNING") frame_scores: Dict[str, List[float]] = {} for aid, det in anomaly_by_id.items(): frame_scores.setdefault(ann_by_id[aid].frame_id, []).append(det.score) mean_frame_scores = {fid: float(np.mean(v)) for fid, v in frame_scores.items()} events = detect_rare_events(store, dataset_id, mean_frame_scores) units = meter.record("rare_event_detection", len(frames)) self._svc("rare-event-detector", state="HEALTHY", processed=len(events), total=len(events), detail=f"{len(events)} rare events", add_units=units) n_scen = len(store.where("scenarios", dataset_id=dataset_id)) self._svc("scenario-miner", state="HEALTHY", processed=n_scen, total=n_scen, detail=f"{n_scen} scenario groups") # ---- 7. regression tracking (vs previous run baseline) self.stage = "regression" self._svc("regression-tracker", state="RUNNING") current = reporting.quality_metrics(store, dataset_id) cur_global = current["global"] cur_per_class = {c["class_name"]: c for c in current["per_class"]} baseline = self.store.meta.get("baseline_metrics") affected: List[str] = [] if baseline: affected = find_affected_classes(cur_per_class, baseline.get("per_class", {})) reg = compare_runs( store, cur_global, model_version, dataset.version, run_id, baseline_metrics=baseline.get("global") if baseline else None, baseline_version=baseline.get("model_version") if baseline else None, affected_classes=affected, affected_scenarios=[s["scenario"] for s in current["per_scenario"] if s.get("recall") is not None and s["recall"] < 0.5], ) self.regression_alert = reg.regression_detected units = meter.record("regression_tracking", len(annotations)) self._svc("regression-tracker", state="DEGRADED" if reg.regression_detected else "HEALTHY", detail="REGRESSION DETECTED" if reg.regression_detected else f"stable vs {reg.baseline_version or 'no baseline'}", add_units=units) # Current run becomes the next baseline. self.store.meta["baseline_metrics"] = { "global": cur_global, "per_class": cur_per_class, "model_version": model_version, } # ---- 8. triage self.stage = "triage" self._svc("triage-engine", state="RUNNING", processed=0, total=len(annotations)) flagged: List[Annotation] = [] processed_t = 0 while True: batch = self.queue.consume("triage", BATCH_SIZE) if not batch: break for msg in batch: ann = ann_by_id[msg["annotation_id"]] decision = triage_annotation( store, ann, validation=store.get("validations", ann.annotation_id), anomaly=store.get("anomalies", ann.annotation_id), grading=store.get("grader_comparisons", ann.annotation_id), tracking=tracking_by_id.get(ann.annotation_id), policy=policy, model_regressed=reg.regression_detected, ) if decision.status == "AUTO_GRADED": ann.status = "VERIFIED" # auto-release into the verified pool else: ann.status = "FLAGGED" flagged.append(ann) store.put("annotations", ann) processed_t += 1 self.queue.ack("triage", len(batch)) self._svc("triage-engine", processed=processed_t) self._svc("triage-engine", state="HEALTHY", detail=f"{processed_t - len(flagged)} auto-graded / {len(flagged)} flagged") # ---- 9. review tasks for flagged labels self.stage = "review_dispatch" self._svc("review-service", state="RUNNING") batch_id = f"review-batch-{run_id[-6:]}" open_tasks = {t.annotation_id for t in store.all("review_tasks") if t.status != "resolved"} created = 0 for ann in flagged: if ann.annotation_id in open_tasks: continue decision = store.get("triage_decisions", ann.annotation_id) task = ReviewTask( task_id=new_id("task"), annotation_id=ann.annotation_id, frame_id=ann.frame_id, dataset_id=dataset_id, failure_reasons=decision.failure_reasons if decision else [], primary_failure_reason=decision.primary_failure_reason if decision else None, review_batch=batch_id, ) store.put("review_tasks", task) created += 1 n_open = len([t for t in store.all("review_tasks") if t.status != "resolved"]) self._svc("review-service", state="HEALTHY", processed=n_open, total=n_open, detail=f"{created} new tasks (batch {batch_id})") # ---- 10. metrics + alerts self.stage = "metrics" self._svc("metrics-service", state="RUNNING") self._refresh_alerts(dataset_id) dataset.status = "evaluated" store.put("datasets", dataset) store.audit("pipeline_completed", "Run", run_id, f"dataset={dataset_id} model={model_version} labels={len(annotations)} flagged={len(flagged)}") self._svc("metrics-service", state="HEALTHY", detail="aggregates refreshed") self.stage = "complete" # ------------------------------------------------------------ tracking evidence def _tracking_evidence(self, annotations: List[Annotation], frames_by_id: Dict) -> Dict[str, TrackingEvidence]: """Detect ID switches and fragmentation from label tracks vs GT instances.""" store = self.store # GT instance -> ordered (frame_index, annotation) by_instance: Dict[str, List] = {} track_instances: Dict[str, set] = {} for a in annotations: if a.matched_gt_id is None or a.track_id is None: continue frame = frames_by_id.get(a.frame_id) gt = next((g for g in frame.gt_boxes if g.gt_id == a.matched_gt_id), None) if frame else None if gt is None: continue by_instance.setdefault(gt.track_instance_id, []).append((frame.index, a)) track_instances.setdefault(a.track_id, set()).add(gt.track_instance_id) evidence: Dict[str, TrackingEvidence] = {} # A pred track spanning multiple GT instances => identity contamination. contaminated_tracks = {t for t, insts in track_instances.items() if len(insts) > 1} for inst, items in by_instance.items(): items.sort(key=lambda p: p[0]) track_ids = [a.track_id for _, a in items] distinct = set(track_ids) fragmented = len(distinct) > 1 prev_tid = None for idx, (_, a) in enumerate(items): switched = prev_tid is not None and a.track_id != prev_tid prev_tid = a.track_id contaminated = a.track_id in contaminated_tracks quality = 1.0 if fragmented: quality -= 0.3 if switched or contaminated: quality -= 0.4 ev = TrackingEvidence( annotation_id=a.annotation_id, id_switch=bool(switched or contaminated), fragmentation=bool(fragmented and idx >= 1 and a.track_id != track_ids[0]), track_quality=round(max(quality, 0.0), 3), ) evidence[a.annotation_id] = ev store.put("tracking_evidence", ev) # Labels without tracks (e.g. false positives) get neutral-low evidence. for a in annotations: if a.annotation_id not in evidence: ev = TrackingEvidence(annotation_id=a.annotation_id, track_quality=0.7 if a.track_id else None) evidence[a.annotation_id] = ev store.put("tracking_evidence", ev) return evidence # ------------------------------------------------------------ alerts def _refresh_alerts(self, dataset_id: str) -> None: store = self.store store.clear("alerts") m = reporting.quality_metrics(store, dataset_id)["global"] counters = reporting.counters(store, dataset_id) qs = self.queue.stats() def alert(kind: str, severity: str, message: str, page: str, eid: Optional[str] = None): store.put("alerts", Alert(alert_id=new_id("alert"), kind=kind, severity=severity, message=message, evidence_page=page, evidence_id=eid)) if self.regression_alert: alert("model_regression", "critical", "Model regression detected vs baseline", "regression") if m["anomaly_rate"] > 0.15: alert("high_anomaly_rate", "warning", f"Anomaly rate {m['anomaly_rate']:.1%} exceeds 15%", "rare-events") if m["grader_consensus"] is not None and m["grader_consensus"] < 0.85: alert("low_consensus", "warning", f"Mean grader consensus {m['grader_consensus']:.1%} below 85%", "quality") if m["id_swap_rate"] is not None and m["id_swap_rate"] > 0.05: alert("high_id_swaps", "warning", f"ID swap rate {m['id_swap_rate']:.1%} above 5%", "quality") if m["fragmentation_rate"] is not None and m["fragmentation_rate"] > 0.08: alert("high_fragmentation", "warning", f"Track fragmentation {m['fragmentation_rate']:.1%} above 8%", "quality") total = max(counters["evaluated"], 1) hitl_rate = counters["flagged"] / total if hitl_rate > 0.30: alert("high_hitl_rate", "warning", f"HITL rate {hitl_rate:.1%} above 30%", "review") ver_rate = counters["verified"] / total if ver_rate < 0.5 and counters["evaluated"] > 0: alert("low_verification_rate", "warning", f"Verification rate {ver_rate:.1%} below 50%", "datasets") if qs["pending"] > 5000: alert("queue_backlog", "warning", f"Queue backlog {qs['pending']} messages", "pipeline") if qs["failed"] > 0: alert("processing_failure", "critical", f"{qs['failed']} failed queue messages", "pipeline") if m["safety_critical_recall"] is not None and m["safety_critical_recall"] < 0.8: alert("safety_recall_low", "critical", f"Safety-critical recall {m['safety_critical_recall']:.1%} below 80%", "quality") # ------------------------------------------------------------ HITL review def apply_review(self, task_id: str, action: RelabelingAction, actor: str = "human-reviewer") -> Dict: """Apply a human action, then ALWAYS re-run validation, grading and triage gates before any verified/rejected outcome (spec §23).""" store = self.store task = store.get("review_tasks", task_id) if task is None: raise KeyError(f"Unknown review task {task_id}") ann = store.get("annotations", task.annotation_id) frame = store.get("frames", ann.frame_id) dataset = store.get("datasets", ann.dataset_id) store.put("human_reviews", HumanReview( review_id=new_id("rev"), task_id=task_id, annotation_id=ann.annotation_id, reviewer=actor, action=action)) meter = ProcessMeter(store, self.last_run_id or "") meter.record("hitl", 1) act = action.action if act == "reject": ann.status = "REJECTED" store.put("annotations", ann) task.status = "resolved" task.resolution = ReviewResolution(action=act, revalidation_passed=None, final_status="REJECTED") store.put("review_tasks", task) store.audit("review_rejected", "Annotation", ann.annotation_id, "human rejected label", actor) store.save() return {"task": task, "revalidation": None, "message": "Label rejected by human review."} if act == "mark_ignore": ann.status = "REJECTED" store.put("annotations", ann) task.status = "resolved" task.resolution = ReviewResolution(action=act, revalidation_passed=None, final_status="REJECTED") store.put("review_tasks", task) store.audit("review_ignored", "Annotation", ann.annotation_id, "marked ignore region", actor) store.save() return {"task": task, "revalidation": None, "message": "Marked as ignore; excluded from training."} if act == "correct": if action.corrected_bbox_3d: ann.bbox_3d = [float(v) for v in action.corrected_bbox_3d] ann.bbox_2d = synthetic.project_bbox_2d(ann.bbox_3d) if action.corrected_class: ann.class_name = action.corrected_class ann.source = "human_corrected" self._set_human_reference(frame, ann) elif act == "approve": # Human asserts the label is correct: label geometry becomes the # human-verified reference (the human overrules the vendor GT). ann.source = "human_approved" self._set_human_reference(frame, ann) elif act == "merge_tracks": if action.merge_with_track_id: ann.track_id = action.merge_with_track_id ann.source = "human_corrected" elif act == "split_track": ann.track_id = f"{ann.track_id or 'trk'}-split-{task_id[-4:]}" ann.source = "human_corrected" store.put("annotations", ann) # --- re-validation -> re-grading -> re-triage (gates always re-run) points = synthetic.frame_points(store, frame) validation = validate_annotation(store, ann, frame, points=points) tev = store.get("tracking_evidence", ann.annotation_id) if act in ("merge_tracks", "split_track") and tev is not None: tev.id_switch = False tev.fragmentation = False tev.track_quality = max(tev.track_quality or 0.7, 0.8) store.put("tracking_evidence", tev) grading = grade_annotation(store, ann, frame, temporal_quality=tev.track_quality if tev else None, seed=dataset.seed if dataset else 7) anomaly = self._rescore_anomaly(ann) decision = triage_annotation( store, ann, validation=validation, anomaly=anomaly, grading=grading, tracking=tev, policy=load_policy(), model_regressed=False) if decision.status == "AUTO_GRADED": ann.status = "VERIFIED" final = "VERIFIED" else: ann.status = "FLAGGED" final = "FLAGGED" store.put("annotations", ann) task.status = "resolved" task.resolution = ReviewResolution( action=act, corrected_bbox_3d=action.corrected_bbox_3d, corrected_class=action.corrected_class, revalidation_passed=decision.status == "AUTO_GRADED", final_status=final, ) store.put("review_tasks", task) # Dataset lineage: corrected by this review batch. if dataset is not None and task.review_batch: dataset.lineage.corrected_by_review_batch = task.review_batch store.put("datasets", dataset) store.audit(f"review_{act}", "Annotation", ann.annotation_id, f"re-validation {'PASSED' if final == 'VERIFIED' else 'FAILED: ' + ', '.join(decision.failure_reasons)}", actor) self._refresh_alerts(ann.dataset_id) store.save() msg = ("Re-validation re-ran all gates: PASSED — label verified." if final == "VERIFIED" else f"Re-validation re-ran all gates: still failing {', '.join(decision.failure_reasons)}.") return {"task": task, "revalidation": None, "message": msg} def _set_human_reference(self, frame, ann: Annotation) -> None: """Human-verified geometry becomes the reference ground truth.""" store = self.store gt = next((g for g in frame.gt_boxes if g.gt_id == ann.matched_gt_id), None) if gt is None: from sensorflow.evaluation.records import GroundTruthBox gt = GroundTruthBox( gt_id=f"{frame.frame_id}-human-{ann.annotation_id[-6:]}", class_name=ann.class_name, bbox_3d=list(ann.bbox_3d), track_instance_id=ann.object_id, gt_type="HUMAN_VERIFIED_GROUND_TRUTH", ) frame.gt_boxes.append(gt) ann.matched_gt_id = gt.gt_id else: gt.class_name = ann.class_name gt.bbox_3d = list(ann.bbox_3d) gt.gt_type = "HUMAN_VERIFIED_GROUND_TRUTH" gt.bbox_2d = synthetic.project_bbox_2d(gt.bbox_3d) store.put("frames", frame) def _rescore_anomaly(self, ann: Annotation) -> Optional[AnomalyDetection]: """Re-score a corrected label against the fitted ensemble via score percentiles; falls back to the stored record when no context exists.""" stored = self.store.get("anomalies", ann.annotation_id) ctx = self._score_context if ctx is None: return stored try: X, ids = synthetic.annotation_features(self.store, ann.dataset_id) if ann.annotation_id not in ids: return stored row = X[ids.index(ann.annotation_id)][None, :] Xs = ctx["scaler"].transform(row) raw_scores, norm_scores = {}, {} for det in ctx["detectors"]: s = float(det.score(Xs)[0]) raw_scores[det.name] = round(s, 4) population = ctx["raw"][det.name] norm_scores[det.name] = round(float((population < s).mean()), 4) from sensorflow.evaluation.detectors import DEFAULT_WEIGHTS names = list(norm_scores) if ctx["strategy"] == "majority_vote": score = float(np.mean([norm_scores[n] >= ctx["threshold"] for n in names])) else: w = np.array([DEFAULT_WEIGHTS.get(n, 1.0) for n in names]) score = float((np.array([norm_scores[n] for n in names]) * w).sum() / w.sum()) det_rec = AnomalyDetection( annotation_id=ann.annotation_id, score=round(score, 4), is_anomaly=bool(score >= ctx["threshold"]), detector_scores=raw_scores, normalized_scores=norm_scores, ensemble_strategy=ctx["strategy"], ensemble_score=round(score, 4), decision_threshold=ctx["threshold"], ) self.store.put("anomalies", det_rec) return det_rec except Exception: return stored # ------------------------------------------------------------ training flywheel def _active_model_version(self) -> str: models = sorted(self.store.all("models"), key=lambda m: m.created_at) return models[-1].model_version if models else "model-v1" def start_training(self, dataset_version: str, model_version: Optional[str] = None, configuration: Optional[Dict] = None, quality_policy: str = "", training_parameters: Optional[Dict] = None) -> TrainingJob: store = self.store dataset = store.get("datasets", dataset_version) if dataset is None: for d in store.all("datasets"): if d.version == dataset_version or d.name == dataset_version: dataset = d break if dataset is None: raise KeyError(f"Unknown dataset {dataset_version}") verified = [a for a in store.where("annotations", dataset_id=dataset.dataset_id) if a.status == "VERIFIED"] if not verified: raise ValueError("No verified labels available for training — run the evaluation pipeline first") policy = load_policy() existing = store.all("models") next_num = len(existing) + 2 # model-v1 is the implicit generator new_version = model_version or f"model-v{next_num}" model_id = new_id("mdl") lineage = DatasetLineage( generated_from_model=dataset.lineage.generated_from_model, corrected_by_review_batch=dataset.lineage.corrected_by_review_batch, validated_by_policy=quality_policy or policy.policy_id, parent_dataset=dataset.dataset_id, ) tds = TrainingDataset( training_dataset_id=new_id("tds"), source_dataset_id=dataset.dataset_id, version=f"{dataset.version}-train", num_verified_labels=len(verified), lineage=lineage, ) store.put("training_datasets", tds) dataset.lineage.validated_by_policy = policy.policy_id store.put("datasets", dataset) params = training_parameters or {} job = TrainingJob( job_id=new_id("job"), model_id=model_id, model_version=new_version, dataset_version=dataset.dataset_id, total_epochs=int(params.get("epochs", 10)), status="queued", quality_policy=quality_policy or policy.policy_id, configuration=configuration or {}, lineage=lineage, ) job.logs.append(f"[queued] training {new_version} on {len(verified)} verified labels " f"(dataset {dataset.dataset_id}, policy {job.quality_policy})") store.put("training_jobs", job) store.audit("training_started", "TrainingJob", job.job_id, f"{new_version} on {tds.training_dataset_id} ({len(verified)} verified labels)") t = threading.Thread(target=self._train_loop, args=(job.job_id, len(verified)), daemon=True) self._train_threads[job.job_id] = t t.start() return job def _train_loop(self, job_id: str, n_verified: int) -> None: store = self.store job = store.get("training_jobs", job_id) meter = ProcessMeter(store, job_id) job.status = "running" store.put("training_jobs", job) rng = np.random.default_rng(abs(hash(job_id)) % (2 ** 32)) # Data volume drives the achievable ceiling (flywheel effect). ceiling = min(0.985, 0.80 + 0.03 * math.log10(max(n_verified, 10))) prev = self._latest_completed_model() for epoch in range(1, job.total_epochs + 1): time.sleep(0.6) frac = epoch / job.total_epochs job.epoch = epoch job.loss = round(float(2.2 * math.exp(-2.5 * frac) + rng.uniform(0.01, 0.04)), 4) job.rare_recall = round(float(min(ceiling - 0.05, 0.45 + (ceiling - 0.48) * frac + rng.uniform(-0.01, 0.01))), 4) job.safety_recall = round(float(min(ceiling, 0.55 + (ceiling - 0.55) * frac + rng.uniform(-0.01, 0.01))), 4) units = meter.record("training", n_verified // job.total_epochs + 1) job.process_units += units job.logs.append( f"[epoch {epoch}/{job.total_epochs}] loss={job.loss:.4f} " f"rare_recall={job.rare_recall:.3f} safety_recall={job.safety_recall:.3f} pu+={units}") store.put("training_jobs", job) job.status = "completed" job.logs.append(f"[done] {job.model_version} trained; evaluating vs " f"{prev.model_version if prev else 'no previous model'}") store.put("training_jobs", job) # Model evaluation + regression/improvement detection. metrics = ModelEvaluation( precision=round(min(0.99, ceiling + 0.005), 4), recall=round(ceiling - 0.01, 4), f1=round(ceiling - 0.005, 4), map_3d=round(ceiling - 0.03, 4), safety_critical_recall=job.safety_recall, rare_recall=job.rare_recall, ) status = "baseline" if prev is not None and prev.metrics.f1 is not None: status = "improved" if (metrics.f1 or 0) >= prev.metrics.f1 else "regressed" compare_runs( store, {"precision": metrics.precision, "recall": metrics.recall, "safety_critical_recall": metrics.safety_critical_recall}, job.model_version, job.dataset_version, job.job_id, baseline_metrics={"precision": prev.metrics.precision, "recall": prev.metrics.recall, "safety_critical_recall": prev.metrics.safety_critical_recall}, baseline_version=prev.model_version, ) model = Model( model_id=job.model_id, model_version=job.model_version, name="synthlab-detector", trained_on_dataset=job.dataset_version, status="ready", metrics=metrics, regression_status=status, ) store.put("models", model) store.audit("training_completed", "Model", model.model_id, f"{model.model_version} f1={metrics.f1} status={status}") store.save() def _latest_completed_model(self) -> Optional[Model]: models = sorted(self.store.all("models"), key=lambda m: m.created_at) return models[-1] if models else None _PIPELINE: Optional[LabelEvalPipeline] = None _PIPE_LOCK = threading.Lock() def get_pipeline() -> LabelEvalPipeline: global _PIPELINE with _PIPE_LOCK: if _PIPELINE is None: _PIPELINE = LabelEvalPipeline() return _PIPELINE def reset_pipeline(store: Optional[EvalStore] = None) -> LabelEvalPipeline: global _PIPELINE with _PIPE_LOCK: _PIPELINE = LabelEvalPipeline(store=store) return _PIPELINE