manual-sync chart-synthesis 2026-07-02T22:45:08Z workspace/dovla_cil/generation/tangent_chart_synthesis.py

workspace/dovla_cil/generation/tangent_chart_synthesis.py

@@ -0,0 +1,260 @@
+from __future__ import annotations
+
+import math
+from collections import Counter
+from collections.abc import Sequence
+from dataclasses import asdict, dataclass, field
+from typing import Any
+
+from dovla_cil.generation.tangent_spline_cvae import (
+    SplineTangentRow,
+    _aggregate_rows,
+    _evaluate_generated_group,
+    _split_rows,
+    _validate_eval_args,
+    build_spline_tangent_rows,
+)
+
+
+@dataclass(frozen=True)
+class ChartSynthesisConfig:
+    obs_dim: int = 96
+    text_dim: int = 64
+    val_fraction: float = 0.2
+    seed: int = 0
+    neighbor_pool: int = 64
+    direct_count: int = 12
+    barycentric_windows: tuple[int, ...] = field(default_factory=lambda: (4, 8, 16, 32))
+    barycentric_temperature: float = 0.5
+    utility_weight: float = 0.0
+    order: str = "direct_first"
+    same_task_only: bool = True
+
+    def __post_init__(self) -> None:
+        if self.obs_dim <= 0:
+            raise ValueError("obs_dim must be positive")
+        if self.text_dim <= 0:
+            raise ValueError("text_dim must be positive")
+        if not 0.0 < self.val_fraction < 1.0:
+            raise ValueError("val_fraction must be in (0, 1)")
+        if self.neighbor_pool <= 0:
+            raise ValueError("neighbor_pool must be positive")
+        if self.direct_count < 0:
+            raise ValueError("direct_count must be non-negative")
+        if any(window <= 0 for window in self.barycentric_windows):
+            raise ValueError("barycentric_windows must contain positive values")
+        if self.barycentric_temperature <= 0.0:
+            raise ValueError("barycentric_temperature must be positive")
+        if self.utility_weight < 0.0:
+            raise ValueError("utility_weight must be non-negative")
+        if self.order not in {"direct_first", "means_first", "interleave"}:
+            raise ValueError("order must be direct_first, means_first, or interleave")
+
+
+def evaluate_chart_synthesis(
+    targets: Sequence[dict[str, Any]],
+    *,
+    config: ChartSynthesisConfig | None = None,
+    k_values: Sequence[int] = (1, 2, 4, 8, 16),
+    thresholds: Sequence[float] = (0.05, 0.1, 0.2, 0.4),
+) -> dict[str, Any]:
+    """Evaluate local chart synthesis from train-only positive tangents.
+
+    Local-atlas retrieval asks whether a heldout positive tangent already appears
+    among nearby train positives. This diagnostic takes one step toward a
+    generator: it keeps a few nearest chart atoms, then adds barycentric means
+    over increasingly wide local neighborhoods. These means are still
+    deployment-clean because they use only train positives and the current
+    observation-language condition.
+    """
+
+    config = config or ChartSynthesisConfig()
+    _validate_eval_args(k_values, thresholds)
+    rows, task_ids, code_dim, horizon, action_dim = build_spline_tangent_rows(
+        targets,
+        obs_dim=config.obs_dim,
+        text_dim=config.text_dim,
+    )
+    if not rows:
+        raise ValueError("no usable spline tangent rows")
+    split = _split_rows(rows, val_fraction=config.val_fraction, seed=config.seed)
+    train_positive = [
+        row for row in split["train"]
+        if row.example.tangent_label == "positive"
+    ]
+    if not train_positive:
+        raise ValueError("no train positive tangents")
+    train_positive_by_task: dict[str, list[SplineTangentRow]] = {}
+    for row in train_positive:
+        train_positive_by_task.setdefault(row.example.task_id, []).append(row)
+
+    max_k = max(int(k) for k in k_values)
+    val_by_group: dict[str, list[SplineTangentRow]] = {}
+    for row in split["val"]:
+        val_by_group.setdefault(row.example.group_id, []).append(row)
+
+    groups: list[dict[str, Any]] = []
+    for group_id, group_rows in sorted(val_by_group.items()):
+        positives = [
+            row.example for row in group_rows
+            if row.example.tangent_label == "positive"
+        ]
+        if not positives:
+            continue
+        negatives = [
+            row.example for row in group_rows
+            if row.example.tangent_label == "negative"
+        ]
+        proposals = _select_chart_proposals(
+            group_rows[0],
+            train_positive=train_positive,
+            train_positive_by_task=train_positive_by_task,
+            config=config,
+            max_k=max_k,
+        )
+        groups.append(
+            _evaluate_generated_group(
+                group_id,
+                task_id=group_rows[0].example.task_id,
+                proposals=proposals,
+                positives=positives,
+                negatives=negatives,
+                k_values=k_values,
+                thresholds=thresholds,
+            )
+        )
+
+    return {
+        "report_type": "positive_tangent_chart_synthesis_eval",
+        "metric_scope": "offline_support_proxy",
+        "note": (
+            "Chart synthesis proposes train-only positive tangents from local "
+            "atlas neighborhoods plus barycentric chart means. It tests whether "
+            "local positive support is better represented as chart coordinates "
+            "than as raw prototype replay."
+        ),
+        "config": {
+            **asdict(config),
+            "barycentric_windows": list(config.barycentric_windows),
+        },
+        "task_ids": list(task_ids),
+        "code_dim": code_dim,
+        "horizon": horizon,
+        "action_dim": action_dim,
+        "num_examples": len(rows),
+        "num_groups": len({row.example.group_id for row in rows}),
+        "num_train_examples": len(split["train"]),
+        "num_val_examples": len(split["val"]),
+        "num_train_positive": len(train_positive),
+        "num_val_groups_with_positive": len(groups),
+        "label_counts": dict(Counter(row.example.tangent_label for row in rows)),
+        "train_positive_by_task": {
+            task_id: len(train_positive_by_task.get(task_id, []))
+            for task_id in task_ids
+        },
+        "overall": _aggregate_rows(groups, k_values=k_values, thresholds=thresholds),
+        "per_task": {
+            task_id: _aggregate_rows(
+                [group for group in groups if group["task_id"] == task_id],
+                k_values=k_values,
+                thresholds=thresholds,
+            )
+            for task_id in task_ids
+        },
+        "groups": groups,
+    }
+
+
+def _select_chart_proposals(
+    query: SplineTangentRow,
+    *,
+    train_positive: Sequence[SplineTangentRow],
+    train_positive_by_task: dict[str, list[SplineTangentRow]],
+    config: ChartSynthesisConfig,
+    max_k: int,
+) -> list[list[float]]:
+    if config.same_task_only:
+        pool = train_positive_by_task.get(query.example.task_id, []) or list(train_positive)
+    else:
+        pool = list(train_positive)
+    ranked = sorted(
+        (
+            (_condition_distance(query.condition, row.condition), row)
+            for row in pool
+        ),
+        key=lambda item: (item[0], -item[1].example.delta_utility, item[1].example.group_id),
+    )[: max(int(config.neighbor_pool), max_k)]
+    direct_rows = [row for _, row in ranked[: int(config.direct_count)]]
+    direct = [_flat_delta(row) for row in direct_rows]
+    means = [
+        _weighted_barycenter(
+            ranked[: min(int(window), len(ranked))],
+            temperature=config.barycentric_temperature,
+            utility_weight=config.utility_weight,
+        )
+        for window in config.barycentric_windows
+        if ranked
+    ]
+    means = [mean for mean in means if mean]
+    if config.order == "means_first":
+        proposals = means + direct
+    elif config.order == "interleave":
+        proposals = []
+        for index in range(max(len(direct), len(means))):
+            if index < len(direct):
+                proposals.append(direct[index])
+            if index < len(means):
+                proposals.append(means[index])
+    else:
+        proposals = direct + means
+
+    used_direct = len(direct)
+    while len(proposals) < max_k and used_direct < len(ranked):
+        proposals.append(_flat_delta(ranked[used_direct][1]))
+        used_direct += 1
+    return proposals[:max_k]
+
+
+def _weighted_barycenter(
+    ranked: Sequence[tuple[float, SplineTangentRow]],
+    *,
+    temperature: float,
+    utility_weight: float,
+) -> list[float]:
+    if not ranked:
+        return []
+    nonzero = sorted(distance for distance, _ in ranked if distance > 1.0e-9)
+    local_scale = nonzero[len(nonzero) // 2] if nonzero else 1.0
+    scale = max(local_scale * float(temperature), 1.0e-6)
+    scores = [
+        -float(distance) / scale + float(utility_weight) * row.example.delta_utility
+        for distance, row in ranked
+    ]
+    max_score = max(scores)
+    weights = [math.exp(score - max_score) for score in scores]
+    total = sum(weights) or 1.0
+    vectors = [_flat_delta(row) for _, row in ranked]
+    size = min(len(vector) for vector in vectors)
+    return [
+        sum(weight * vector[index] for weight, vector in zip(weights, vectors, strict=True))
+        / total
+        for index in range(size)
+    ]
+
+
+def _condition_distance(left: Sequence[float], right: Sequence[float]) -> float:
+    return _rms_distance(left, right)
+
+
+def _flat_delta(row: SplineTangentRow) -> list[float]:
+    return row.example.flat_delta
+
+
+def _rms_distance(left: Sequence[float], right: Sequence[float]) -> float:
+    size = min(len(left), len(right))
+    if size <= 0:
+        return 0.0
+    return math.sqrt(
+        sum((float(left[index]) - float(right[index])) ** 2 for index in range(size))
+        / size
+    )