diff --git a/docs/benchmarking/README.md b/docs/benchmarking/README.md index 4e9920fa..1316595f 100644 --- a/docs/benchmarking/README.md +++ b/docs/benchmarking/README.md @@ -32,6 +32,7 @@ definitions, honest comparisons, and the next benchmark decision. | Reproduce DiDeMo or HiREST | [Adapter validation ledger](adapter_validation.md) | | Understand the benchmark-ready Python structure | [Core contract](core_contract.md) | | See which benchmarks exist and what each measures | [Benchmark catalog](benchmark_catalog.md) | +| Evaluate the complete public search path end-to-end | [End-to-end retrieval evaluation](end_to_end_evaluation.md) | | Understand the current model and benchmark choices | [Multimodal model direction](model_selection.md) | | Run the Codex MCP-on/MCP-off experiment | [Codex agent ablation](agent_ablation.md) | | Find exact published competitor scores | [Published comparison results](published_results.md) | diff --git a/docs/benchmarking/end_to_end_evaluation.md b/docs/benchmarking/end_to_end_evaluation.md new file mode 100644 index 00000000..5226c21d --- /dev/null +++ b/docs/benchmarking/end_to_end_evaluation.md @@ -0,0 +1,92 @@ +# End-to-end retrieval evaluation + +Collection index: [Benchmarking research](README.md) + +The dataset adapters ([DiDeMo](adapter_validation.md), HiREST) score one +modality against an official evaluator through the benchmark-ready core. This +harness is complementary: it drives the same public `search` operation a real +client calls and reports, in one place, how the complete product path behaves +on a labeled set. It does not replace the published-dataset baselines and does +not claim a published score. + +The harness lives in `vidxp.benchmarks.end_to_end`. + +## What it measures + +Each measure maps to one line of the request in issue #76: + +| Measure | Metric | +|---|---| +| Whether relevant moments are found | `recall_at_k` at a relevance IoU (default 0.5) | +| Timestamp and temporal-range accuracy | `mean_top1_iou`, plus per-case `best_iou` | +| Ranking quality across modalities | `mean_reciprocal_rank`, `ndcg_at_k` (IoU-graded), `modality_contribution` | +| Whether returned evidence supports the result | `evidence_support_rate` from delivered evidence ranges | +| Latency, failures, degraded/partial results | `latency_ms_*`, `failed_cases`, `no_result_cases`, per-case `evidence_degraded` | + +Ranking scores are ordering-only: they sort results within one response and are +not probabilities. Calibrated scoring is a separate concern (see #90). + +## How it works + +The evaluator is dependency-injected. A caller supplies a `SearchFn` that maps +an `EvaluationCase` to a public `FusedSearchResult`. In tests this is a +deterministic fake, so every metric is exercised without loading models. In +production, bind the real public operation with `application_search_fn`: + +```python +from vidxp.benchmarks.end_to_end import ( + EvaluationDataset, + application_search_fn, + evaluate_end_to_end, +) + +dataset = EvaluationDataset.model_validate_json( + labeled_cases_path.read_text(encoding="utf-8") +) + +search_fn = application_search_fn( + application.search, # the public VidXPApplication.search + snapshot=snapshot_reference, # pin a snapshot for reproducibility + top_k=10, +) + +report = evaluate_end_to_end(dataset, search_fn, relevance_iou=0.5) +report_path.write_text( + report.model_dump_json(indent=2), encoding="utf-8" +) +``` + +Because `search_fn` calls `VidXPApplication.search` with the same +`SearchCommand` a client sends, the harness measures the public product path +rather than an internal shortcut. Passing an `evidence_delivery` policy to +`application_search_fn` also exercises the evidence path, which enables +`evidence_support_rate`. + +A case whose search fails is recorded as failed and excluded from the quality +metrics, so one broken query never hides the rest. + +## Dataset shape + +```json +{ + "name": "smoke", + "cases": [ + { + "case_id": "taxi-night", + "query": "a taxi at night", + "media_id": "…optional single-media scope…", + "modalities": ["scene", "speech"], + "relevant": [ + {"media_id": "…", "start": 12.0, "end": 18.5} + ] + } + ] +} +``` + +## Scope + +This change adds the transport-neutral evaluator and its metrics. Wiring it to +a CLI, HTTP, or MCP surface is intentionally a follow-up so the tested core +lands first; `application_search_fn` already shows the exact binding a surface +adapter needs. diff --git a/src/vidxp/benchmarks/end_to_end.py b/src/vidxp/benchmarks/end_to_end.py new file mode 100644 index 00000000..3256b04d --- /dev/null +++ b/src/vidxp/benchmarks/end_to_end.py @@ -0,0 +1,449 @@ +"""End-to-end retrieval evaluation over VidXP's public search path. + +The dataset benchmarks (DiDeMo, HiREST) score a single modality against an +official evaluator through the benchmark-ready core. This harness is +complementary: it drives the same public ``search`` operation that real +clients call and reports, in one place, whether relevant moments are found, +how accurate their timestamps are, how well results are ranked across +modalities, whether delivered evidence supports the result, and the +operational cost (latency, failures, and degraded evidence). + +The evaluator is dependency-injected. A caller supplies a ``SearchFn`` that +maps an :class:`EvaluationCase` to a public ``FusedSearchResult``; in +production that closure calls ``VidXPApplication.search`` (see +:func:`application_search_fn`), while tests pass a deterministic fake. All +metric helpers are pure functions so the scoring is exercised without models. + +Scores in the returned report inherit VidXP's ordering-only meaning: they rank +results within one response and are not probabilities. +""" + +from __future__ import annotations + +import statistics +from math import log2 +from time import perf_counter +from typing import Callable + +from pydantic import BaseModel, ConfigDict, Field, model_validator + +from vidxp.application_models import ( + ApplicationError, + EvidenceDeliveryState, + FusedSearchResult, + IndexSnapshotReference, + InitialEvidenceDeliveryPolicy, + SearchCommand, +) + + +class EvalModel(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + + +class GroundTruthInterval(EvalModel): + """A relevant time span for a query, in one media item.""" + + media_id: str = Field(min_length=1) + start: float = Field(ge=0) + end: float = Field(gt=0) + + @model_validator(mode="after") + def _validate_interval(self) -> "GroundTruthInterval": + if self.end <= self.start: + raise ValueError("Ground-truth end must be greater than start.") + return self + + +class EvaluationCase(EvalModel): + """One labeled query and the moments that should be retrieved for it.""" + + case_id: str = Field(min_length=1) + query: str = Field(min_length=1) + media_id: str | None = Field( + default=None, + description="Optional single-media scope passed through to search.", + ) + modalities: tuple[str, ...] = () + relevant: tuple[GroundTruthInterval, ...] = Field(min_length=1) + + +class EvaluationDataset(EvalModel): + name: str = Field(min_length=1) + cases: tuple[EvaluationCase, ...] = Field(min_length=1) + + @model_validator(mode="after") + def _unique_case_ids(self) -> "EvaluationDataset": + ids = [case.case_id for case in self.cases] + if len(ids) != len(set(ids)): + raise ValueError("Evaluation case ids must be unique.") + return self + + +class CaseResult(EvalModel): + """Per-case outcome; failed cases carry an error and no metrics.""" + + case_id: str + status: str = Field(description="ok, no_result, or failed.") + returned_moments: int = 0 + top1_iou: float = 0.0 + best_iou: float = 0.0 + first_relevant_rank: int | None = None + contributing_modalities: tuple[str, ...] = () + evidence_delivered: bool = False + evidence_supported: bool | None = None + evidence_degraded: bool = False + latency_ms: float = 0.0 + error_code: str | None = None + + +class AggregateMetrics(EvalModel): + evaluated_cases: int + failed_cases: int + no_result_cases: int + recall_at_k: dict[int, float] = Field(default_factory=dict) + mean_top1_iou: float = 0.0 + mean_reciprocal_rank: float = 0.0 + ndcg_at_k: dict[int, float] = Field(default_factory=dict) + modality_contribution: dict[str, int] = Field(default_factory=dict) + evidence_support_rate: float | None = None + latency_ms_mean: float = 0.0 + latency_ms_p50: float = 0.0 + latency_ms_p95: float = 0.0 + + +class EndToEndEvaluationReport(EvalModel): + dataset_name: str + total_cases: int + relevance_iou: float + k_values: tuple[int, ...] + score_meaning: str = Field( + default="ordering_only", + description="Search scores rank within one response; not a probability.", + ) + aggregate: AggregateMetrics + cases: tuple[CaseResult, ...] + + +SearchFn = Callable[[EvaluationCase], FusedSearchResult] + + +def temporal_iou( + a_start: float, + a_end: float, + b_start: float, + b_end: float, +) -> float: + """Intersection-over-union of two time intervals; 0.0 when disjoint.""" + + intersection = max(0.0, min(a_end, b_end) - max(a_start, b_start)) + if intersection <= 0.0: + return 0.0 + union = (a_end - a_start) + (b_end - b_start) - intersection + if union <= 0.0: + return 0.0 + return intersection / union + + +def best_interval_iou( + *, + media_id: str, + start: float, + end: float, + relevant: tuple[GroundTruthInterval, ...], +) -> float: + """Best temporal IoU of one span against same-media ground truth.""" + + return max( + ( + temporal_iou(start, end, gt.start, gt.end) + for gt in relevant + if gt.media_id == media_id + ), + default=0.0, + ) + + +def _dcg(gains: list[float]) -> float: + return sum(gain / log2(rank + 1) for rank, gain in enumerate(gains, start=1)) + + +def ndcg_at_k(gains: list[float], k: int) -> float: + """Normalized DCG over graded (IoU) gains for the first ``k`` results. + + Normalizes against the same retrieved gains sorted descending, so the + metric reflects how well the ranking ordered the moments it returned. + """ + + top = gains[:k] + ideal = _dcg(sorted(top, reverse=True)) + if ideal <= 0.0: + return 0.0 + return _dcg(top) / ideal + + +def _percentile(values: list[float], fraction: float) -> float: + if not values: + return 0.0 + ordered = sorted(values) + if len(ordered) == 1: + return ordered[0] + position = fraction * (len(ordered) - 1) + lower = int(position) + upper = min(lower + 1, len(ordered) - 1) + weight = position - lower + return ordered[lower] * (1.0 - weight) + ordered[upper] * weight + + +def _evidence_outcome( + result: FusedSearchResult, + case: EvaluationCase, + *, + relevance_iou: float, +) -> tuple[bool, bool | None, bool]: + """Return (delivered, supported, degraded) for a case's evidence.""" + + delivery = result.evidence_delivery + if delivery is None or not delivery.items: + return False, None, False + supported = False + degraded = False + for item in delivery.items: + if item.state != EvidenceDeliveryState.ready: + degraded = True + if item.range is None: + continue + iou = best_interval_iou( + media_id=item.media_id, + start=item.range.source_start_seconds, + end=item.range.source_end_seconds, + relevant=case.relevant, + ) + if iou >= relevance_iou: + supported = True + return True, supported, degraded + + +def evaluate_case( + case: EvaluationCase, + result: FusedSearchResult, + *, + relevance_iou: float, + latency_ms: float, +) -> tuple[CaseResult, list[float]]: + """Score one already-executed case; also return the per-moment IoU gains.""" + + moments = tuple(sorted(result.moments, key=lambda moment: moment.rank)) + gains = [ + best_interval_iou( + media_id=moment.media_id, + start=moment.start, + end=moment.end, + relevant=case.relevant, + ) + for moment in moments + ] + first_relevant_rank: int | None = None + contributing: tuple[str, ...] = () + for position, (moment, gain) in enumerate(zip(moments, gains), start=1): + if gain >= relevance_iou: + first_relevant_rank = position + contributing = tuple(moment.modalities) + break + delivered, supported, degraded = _evidence_outcome( + result, + case, + relevance_iou=relevance_iou, + ) + status = "ok" if moments else "no_result" + case_result = CaseResult( + case_id=case.case_id, + status=status, + returned_moments=len(moments), + top1_iou=gains[0] if gains else 0.0, + best_iou=max(gains, default=0.0), + first_relevant_rank=first_relevant_rank, + contributing_modalities=contributing, + evidence_delivered=delivered, + evidence_supported=supported, + evidence_degraded=degraded, + latency_ms=latency_ms, + ) + return case_result, gains + + +def evaluate_end_to_end( + dataset: EvaluationDataset, + search_fn: SearchFn, + *, + relevance_iou: float = 0.5, + k_values: tuple[int, ...] = (1, 5, 10), +) -> EndToEndEvaluationReport: + """Run every case through ``search_fn`` and aggregate the five measures. + + A case whose search raises is recorded as failed and excluded from the + quality metrics; the run continues so one broken query never hides the + rest. An :class:`ApplicationError` keeps its ``code``; any other exception + is recorded under its class name. + """ + + if not 0.0 < relevance_iou <= 1.0: + raise ValueError("relevance_iou must be within (0, 1].") + if not k_values or any(k <= 0 for k in k_values): + raise ValueError("k_values must be positive.") + ordered_k = tuple(sorted(set(k_values))) + + case_results: list[CaseResult] = [] + gains_by_case: list[list[float]] = [] + latencies: list[float] = [] + + for case in dataset.cases: + started = perf_counter() + try: + result = search_fn(case) + except Exception as error: + # A reliability harness must survive one broken query and still + # report it, so every search failure is recorded and the run + # continues. Metric computation below is outside this guard, so a + # bug in the harness itself still surfaces. + latency_ms = (perf_counter() - started) * 1000.0 + latencies.append(latency_ms) + error_code = ( + error.code + if isinstance(error, ApplicationError) + else type(error).__name__ + ) + case_results.append( + CaseResult( + case_id=case.case_id, + status="failed", + latency_ms=latency_ms, + error_code=error_code, + ) + ) + gains_by_case.append([]) + continue + latency_ms = (perf_counter() - started) * 1000.0 + latencies.append(latency_ms) + case_result, gains = evaluate_case( + case, + result, + relevance_iou=relevance_iou, + latency_ms=latency_ms, + ) + case_results.append(case_result) + gains_by_case.append(gains) + + aggregate = _aggregate( + case_results, + gains_by_case, + latencies, + relevance_iou=relevance_iou, + k_values=ordered_k, + ) + return EndToEndEvaluationReport( + dataset_name=dataset.name, + total_cases=len(dataset.cases), + relevance_iou=relevance_iou, + k_values=ordered_k, + aggregate=aggregate, + cases=tuple(case_results), + ) + + +def _aggregate( + case_results: list[CaseResult], + gains_by_case: list[list[float]], + latencies: list[float], + *, + relevance_iou: float, + k_values: tuple[int, ...], +) -> AggregateMetrics: + scored = [ + (result, gains) + for result, gains in zip(case_results, gains_by_case) + if result.status != "failed" + ] + evaluated = len(scored) + failed = sum(1 for result in case_results if result.status == "failed") + no_result = sum(1 for result in case_results if result.status == "no_result") + + recall_at_k: dict[int, float] = {} + ndcg_at_k_values: dict[int, float] = {} + modality_contribution: dict[str, int] = {} + reciprocal_ranks: list[float] = [] + top1_ious: list[float] = [] + evidence_flags: list[bool] = [] + + for result, gains in scored: + top1_ious.append(gains[0] if gains else 0.0) + rank = result.first_relevant_rank + reciprocal_ranks.append(1.0 / rank if rank is not None else 0.0) + for modality in result.contributing_modalities: + modality_contribution[modality] = ( + modality_contribution.get(modality, 0) + 1 + ) + if result.evidence_delivered and result.evidence_supported is not None: + evidence_flags.append(result.evidence_supported) + + for k in k_values: + hits = sum( + 1 + for _, gains in scored + if any(gain >= relevance_iou for gain in gains[:k]) + ) + recall_at_k[k] = hits / evaluated if evaluated else 0.0 + ndcg_scores = [ndcg_at_k(gains, k) for _, gains in scored] + ndcg_at_k_values[k] = ( + statistics.fmean(ndcg_scores) if ndcg_scores else 0.0 + ) + + return AggregateMetrics( + evaluated_cases=evaluated, + failed_cases=failed, + no_result_cases=no_result, + recall_at_k=recall_at_k, + mean_top1_iou=statistics.fmean(top1_ious) if top1_ious else 0.0, + mean_reciprocal_rank=( + statistics.fmean(reciprocal_ranks) if reciprocal_ranks else 0.0 + ), + ndcg_at_k=ndcg_at_k_values, + modality_contribution=modality_contribution, + evidence_support_rate=( + statistics.fmean(1.0 if flag else 0.0 for flag in evidence_flags) + if evidence_flags + else None + ), + latency_ms_mean=statistics.fmean(latencies) if latencies else 0.0, + latency_ms_p50=_percentile(latencies, 0.50), + latency_ms_p95=_percentile(latencies, 0.95), + ) + + +def application_search_fn( + search: Callable[..., FusedSearchResult], + *, + snapshot: IndexSnapshotReference | None = None, + top_k: int = 10, + evidence_delivery: InitialEvidenceDeliveryPolicy | None = None, +) -> SearchFn: + """Bind VidXP's public ``search`` operation into a ``SearchFn``. + + ``search`` is normally ``VidXPApplication.search``. Each case is turned + into the same ``SearchCommand`` a real client would send, so the harness + measures the public product path rather than an internal shortcut. + """ + + def run(case: EvaluationCase) -> FusedSearchResult: + command = SearchCommand( + query=case.query, + modalities=case.modalities, + media_id=case.media_id, + top_k=top_k, + evidence_delivery=evidence_delivery, + ) + if snapshot is None: + return search(command) + return search(command, snapshot=snapshot) + + return run diff --git a/tests/test_end_to_end_eval.py b/tests/test_end_to_end_eval.py new file mode 100644 index 00000000..e557caf7 --- /dev/null +++ b/tests/test_end_to_end_eval.py @@ -0,0 +1,315 @@ +import unittest + +from vidxp.application_models import ( + ApplicationError, + ErrorCategory, + EvidenceDeliveryItem, + EvidenceDeliveryPolicy, + EvidenceDeliveryResult, + EvidenceDeliveryState, + EvidenceRangeResolution, + FusedMoment, + FusedSearchResult, + FusionProvenance, + SearchCommand, + SearchHit, +) +from vidxp.benchmarks.end_to_end import ( + EvaluationCase, + EvaluationDataset, + GroundTruthInterval, + application_search_fn, + best_interval_iou, + evaluate_end_to_end, + ndcg_at_k, + temporal_iou, +) + + +MEDIA_ID = "123456781234423481234567890abcde" +OTHER_MEDIA = "223456781234423481234567890abcde" +GENERATION_ID = "323456781234423481234567890abcde" +EVIDENCE_ID = "ab" * 32 + + +def make_hit(modality: str, media_id: str, start: float, end: float, rank: int): + return SearchHit( + rank=rank, + media_id=media_id, + video_id=media_id, + generation_id=GENERATION_ID, + start=start, + end=end, + score=-float(rank), + raw_distance=float(rank), + modality=modality, + source_id=f"{modality}:{rank}", + ) + + +def make_moment( + rank: int, + media_id: str, + start: float, + end: float, + modalities: tuple[str, ...], +): + hits = tuple(make_hit(m, media_id, start, end, rank) for m in modalities) + return FusedMoment( + rank=rank, + score=1.0 / (60 + rank), + media_id=media_id, + start=start, + end=end, + modalities=tuple(sorted(set(modalities))), + hits=hits, + ) + + +def make_result(moments, *, modalities=("scene",), evidence=None): + return FusedSearchResult( + query_id="fused:test", + query="taxi", + modalities=modalities, + moments=tuple(moments), + fusion=FusionProvenance( + requested_modalities=modalities, + searched_modalities=modalities, + ), + evidence_delivery=evidence, + ) + + +def make_evidence(start: float, end: float, *, state=EvidenceDeliveryState.ready): + return EvidenceDeliveryResult( + policy=EvidenceDeliveryPolicy(), + items=( + EvidenceDeliveryItem( + evidence_id=EVIDENCE_ID, + rank=1, + media_id=MEDIA_ID, + generation_id=GENERATION_ID, + modalities=("scene",), + state=state, + range=EvidenceRangeResolution( + source_start_seconds=start, + source_end_seconds=end, + representative_timestamp_seconds=start, + clip_start_seconds=start, + clip_end_seconds=end, + requested_padding_before_seconds=0.0, + requested_padding_after_seconds=0.0, + applied_padding_before_seconds=0.0, + applied_padding_after_seconds=0.0, + ), + ), + ), + ) + + +def gt_case(case_id: str, start: float, end: float) -> EvaluationCase: + return EvaluationCase( + case_id=case_id, + query="taxi", + media_id=MEDIA_ID, + modalities=("scene",), + relevant=(GroundTruthInterval(media_id=MEDIA_ID, start=start, end=end),), + ) + + +class MetricFunctionTests(unittest.TestCase): + def test_temporal_iou_overlap_disjoint_and_identical(self): + self.assertEqual(temporal_iou(0, 10, 0, 10), 1.0) + self.assertEqual(temporal_iou(0, 10, 20, 30), 0.0) + # [0,10] vs [5,15]: intersection 5, union 15 -> 1/3 + self.assertAlmostEqual(temporal_iou(0, 10, 5, 15), 1 / 3) + + def test_best_interval_iou_ignores_other_media(self): + relevant = ( + GroundTruthInterval(media_id=MEDIA_ID, start=10, end=20), + GroundTruthInterval(media_id=OTHER_MEDIA, start=0, end=100), + ) + # A perfect overlap exists only on OTHER_MEDIA; scoped to MEDIA_ID it is 0. + self.assertEqual( + best_interval_iou(media_id=MEDIA_ID, start=0, end=5, relevant=relevant), + 0.0, + ) + self.assertEqual( + best_interval_iou( + media_id=MEDIA_ID, start=10, end=20, relevant=relevant + ), + 1.0, + ) + + def test_ndcg_rewards_relevant_first(self): + self.assertEqual(ndcg_at_k([], 5), 0.0) + # Perfect gain ordering normalizes to 1.0. + self.assertAlmostEqual(ndcg_at_k([1.0, 0.5, 0.0], 3), 1.0) + # A relevant item buried below an irrelevant one scores below ideal. + self.assertLess(ndcg_at_k([0.0, 1.0], 2), 1.0) + + +class EvaluateEndToEndTests(unittest.TestCase): + def _run(self, cases, results_by_id, **kwargs): + dataset = EvaluationDataset(name="unit", cases=tuple(cases)) + + def search_fn(case): + outcome = results_by_id[case.case_id] + if isinstance(outcome, BaseException): + raise outcome + return outcome + + return evaluate_end_to_end(dataset, search_fn, **kwargs) + + def test_perfect_hit_scores_top_metrics(self): + case = gt_case("a", 10, 20) + result = make_result([make_moment(1, MEDIA_ID, 10, 20, ("scene",))]) + report = self._run([case], {"a": result}) + + agg = report.aggregate + self.assertEqual(agg.evaluated_cases, 1) + self.assertEqual(agg.recall_at_k[1], 1.0) + self.assertEqual(agg.mean_top1_iou, 1.0) + self.assertEqual(agg.mean_reciprocal_rank, 1.0) + self.assertEqual(agg.ndcg_at_k[1], 1.0) + self.assertEqual(agg.modality_contribution, {"scene": 1}) + + def test_relevant_below_rank_one_lowers_recall_and_mrr(self): + case = gt_case("b", 10, 20) + result = make_result( + [ + make_moment(1, MEDIA_ID, 100, 110, ("scene",)), # miss + make_moment(2, MEDIA_ID, 11, 19, ("scene",)), # relevant + ] + ) + report = self._run([case], {"b": result}, k_values=(1, 5)) + + agg = report.aggregate + self.assertEqual(agg.recall_at_k[1], 0.0) + self.assertEqual(agg.recall_at_k[5], 1.0) + self.assertEqual(agg.mean_reciprocal_rank, 0.5) + self.assertEqual(report.cases[0].first_relevant_rank, 2) + + def test_failed_case_is_isolated_from_quality_metrics(self): + cases = [gt_case("ok", 10, 20), gt_case("bad", 10, 20)] + results = { + "ok": make_result([make_moment(1, MEDIA_ID, 10, 20, ("scene",))]), + "bad": ApplicationError( + "boom", ErrorCategory.internal, "search exploded" + ), + } + report = self._run(cases, results) + + agg = report.aggregate + self.assertEqual(agg.failed_cases, 1) + self.assertEqual(agg.evaluated_cases, 1) + self.assertEqual(agg.recall_at_k[1], 1.0) # denominator excludes failure + failed = next(c for c in report.cases if c.case_id == "bad") + self.assertEqual(failed.status, "failed") + self.assertEqual(failed.error_code, "boom") + + def test_unexpected_exception_is_recorded_not_raised(self): + # search() only wraps known errors; an unexpected type must not abort + # the whole run. It is recorded as failed with its class name. + cases = [gt_case("ok", 10, 20), gt_case("boom", 10, 20)] + results = { + "ok": make_result([make_moment(1, MEDIA_ID, 10, 20, ("scene",))]), + "boom": RuntimeError("unexpected"), + } + report = self._run(cases, results) + + self.assertEqual(report.aggregate.failed_cases, 1) + self.assertEqual(report.aggregate.evaluated_cases, 1) + failed = next(c for c in report.cases if c.case_id == "boom") + self.assertEqual(failed.status, "failed") + self.assertEqual(failed.error_code, "RuntimeError") + + def test_empty_result_counts_as_no_result(self): + case = gt_case("empty", 10, 20) + report = self._run([case], {"empty": make_result([])}) + + self.assertEqual(report.aggregate.no_result_cases, 1) + self.assertEqual(report.aggregate.mean_top1_iou, 0.0) + self.assertEqual(report.cases[0].status, "no_result") + + def test_evidence_support_and_degradation(self): + supported = gt_case("sup", 10, 20) + unsupported = gt_case("uns", 10, 20) + results = { + "sup": make_result( + [make_moment(1, MEDIA_ID, 10, 20, ("scene",))], + evidence=make_evidence(10, 20), + ), + "uns": make_result( + [make_moment(1, MEDIA_ID, 10, 20, ("scene",))], + evidence=make_evidence( + 500, 510, state=EvidenceDeliveryState.partial + ), + ), + } + report = self._run([supported, unsupported], results) + + self.assertEqual(report.aggregate.evidence_support_rate, 0.5) + degraded = next(c for c in report.cases if c.case_id == "uns") + self.assertTrue(degraded.evidence_degraded) + self.assertFalse(degraded.evidence_supported) + + def test_evidence_rate_is_none_when_no_evidence_delivered(self): + case = gt_case("a", 10, 20) + result = make_result([make_moment(1, MEDIA_ID, 10, 20, ("scene",))]) + report = self._run([case], {"a": result}) + + self.assertIsNone(report.aggregate.evidence_support_rate) + + def test_invalid_parameters_are_rejected(self): + dataset = EvaluationDataset(name="unit", cases=(gt_case("a", 10, 20),)) + with self.assertRaises(ValueError): + evaluate_end_to_end(dataset, lambda case: make_result([]), relevance_iou=0) + with self.assertRaises(ValueError): + evaluate_end_to_end(dataset, lambda case: make_result([]), k_values=()) + + +class ApplicationSearchFnTests(unittest.TestCase): + def test_builds_public_search_command(self): + captured = {} + result = make_result([make_moment(1, MEDIA_ID, 10, 20, ("scene",))]) + + def fake_search(command, **kwargs): + captured["command"] = command + captured["kwargs"] = kwargs + return result + + run = application_search_fn(fake_search, top_k=7) + case = EvaluationCase( + case_id="a", + query="a taxi at night", + media_id=MEDIA_ID, + modalities=("scene", "speech"), + relevant=(GroundTruthInterval(media_id=MEDIA_ID, start=1, end=2),), + ) + returned = run(case) + + self.assertIs(returned, result) + command = captured["command"] + self.assertIsInstance(command, SearchCommand) + self.assertEqual(command.query, "a taxi at night") + self.assertEqual(command.modalities, ("scene", "speech")) + self.assertEqual(command.media_id, MEDIA_ID) + self.assertEqual(command.top_k, 7) + self.assertEqual(captured["kwargs"], {}) # no snapshot passed + + def test_passes_snapshot_when_supplied(self): + captured = {} + + def fake_search(command, **kwargs): + captured["kwargs"] = kwargs + return make_result([]) + + run = application_search_fn(fake_search, snapshot="snap-ref") + run(gt_case("a", 1, 2)) + + self.assertEqual(captured["kwargs"], {"snapshot": "snap-ref"}) + + +if __name__ == "__main__": + unittest.main()