Add midplane z-slice to TestCase and thread through runner

benchmark/testset.py

@@ -41,6 +41,7 @@ class TestCase:
     image_b64: str  # Combined view (for backward compatibility)
     top_image_b64: Optional[str]  # TOP view only
     side_image_b64: Optional[str]  # SIDE view only
+    midplane_b64: Optional[str]  # Single XY slice at z=Z//2 (no projection)
     volume: Optional[np.ndarray]
     ground_truth_stage: Optional[str]
 
@@ -215,6 +216,45 @@ def _create_three_view_image(volume: np.ndarray, max_dim: int = 1500) -> str:
     return base64.b64encode(buffer.getvalue()).decode("utf-8")
 
 
+def _create_slice_image(
+    volume: np.ndarray, z: Optional[int] = None, max_dim: int = 800
+) -> str:
+    """Render a single XY z-slice as base64 JPEG.
+
+    Unlike the projection helpers, this preserves depth information at one
+    plane: overlapping body segments stay separated rather than fusing.
+    Uses the same crop as the three-view projection so framing matches.
+
+    Parameters
+    ----------
+    volume : np.ndarray
+        3D volume array (Z, Y, X)
+    z : int, optional
+        Slice index. Defaults to the midplane (Z // 2).
+    max_dim : int
+        Maximum output dimension in pixels.
+    """
+    _ensure_dependencies()
+
+    if z is None:
+        z = volume.shape[0] // 2
+    z = max(0, min(z, volume.shape[0] - 1))
+
+    bounds = _compute_crop_bounds(volume)
+    slice_img = volume[z, bounds[0]:bounds[1], bounds[2]:bounds[3]]
+    slice_norm = _normalize_image(slice_img)
+
+    pil_img = PIL_Image.fromarray(slice_norm)
+    if max(pil_img.size) > max_dim:
+        scale = max_dim / max(pil_img.size)
+        new_size = (int(pil_img.size[0] * scale), int(pil_img.size[1] * scale))
+        pil_img = pil_img.resize(new_size, PIL_Image.Resampling.LANCZOS)
+
+    buffer = io.BytesIO()
+    pil_img.save(buffer, format="JPEG", quality=90)
+    return base64.b64encode(buffer.getvalue()).decode("utf-8")
+
+
 def _create_separate_view_images(volume: np.ndarray, max_dim: int = 1000) -> Tuple[str, str]:
     """Create separate TOP and SIDE view images from volume, return base64 tuple.
 
@@ -359,15 +399,12 @@ def iter_embryo(
             volume = _load_volume(vol_path) if self.load_volumes else None
 
             # Create images
-            if volume is not None:
-                image_b64 = _create_three_view_image(volume)
-                top_b64, side_b64 = _create_separate_view_images(volume)
-            else:
-                # Load just for image if not loading full volumes
-                temp_vol = _load_volume(vol_path)
-                image_b64 = _create_three_view_image(temp_vol)
-                top_b64, side_b64 = _create_separate_view_images(temp_vol)
-                del temp_vol
+            vol = volume if volume is not None else _load_volume(vol_path)
+            image_b64 = _create_three_view_image(vol)
+            top_b64, side_b64 = _create_separate_view_images(vol)
+            midplane_b64 = _create_slice_image(vol)
+            if volume is None:
+                del vol
 
             # Get ground truth
             gt_stage = self.ground_truth.get_stage_at(embryo_id, timepoint)
@@ -378,6 +415,7 @@ def iter_embryo(
                 image_b64=image_b64,
                 top_image_b64=top_b64,
                 side_image_b64=side_b64,
+                midplane_b64=midplane_b64,
                 volume=volume,
                 ground_truth_stage=gt_stage,
             )

experiments/program.md

@@ -52,9 +52,13 @@ async def perceive_xxx(
     references: dict[str, list[str]],  # stage -> [base64 reference images]
     history: list[dict],         # previous timepoints: [{timepoint, stage}]
     timepoint: int,              # current timepoint number
+    midplane_b64: str | None = None,  # optional: single XY z-slice at z=Z//2
 ) -> PerceptionOutput:           # from perception._base
 ```
 
+`midplane_b64` is only passed if your function declares it (or `**kwargs`).
+The runner inspects the signature, so existing variants don't need to change.
+
 ## Available Utilities (from `perception._base`)
 
 - `call_claude(system, content)` — single API call, returns response text

program.md

@@ -52,9 +52,13 @@ async def perceive_xxx(
     references: dict[str, list[str]],  # stage -> [base64 reference images]
     history: list[dict],         # previous timepoints: [{timepoint, stage}]
     timepoint: int,              # current timepoint number
+    midplane_b64: str | None = None,  # optional: single XY z-slice at z=Z//2
 ) -> PerceptionOutput:           # from perception._base
 ```
 
+`midplane_b64` is only passed if your function declares it (or `**kwargs`).
+The runner inspects the signature, so existing variants don't need to change.
+
 ## Available Utilities (from `perception._base`)
 
 - `call_claude(system, content)` — single API call, returns response text

run.py

@@ -20,6 +20,7 @@
 
 import argparse
 import asyncio
+import inspect
 import json
 import logging
 import sys
@@ -95,6 +96,17 @@ def make_prediction_dict(output, timepoint, ground_truth_stage) -> dict:
     }
 
 
+_OPTIONAL_PERCEIVE_KWARGS = ("midplane_b64",)
+
+
+def _accepted_optional_kwargs(perceive_fn) -> set[str]:
+    """Subset of _OPTIONAL_PERCEIVE_KWARGS that perceive_fn's signature accepts."""
+    sig = inspect.signature(perceive_fn)
+    if any(p.kind is p.VAR_KEYWORD for p in sig.parameters.values()):
+        return set(_OPTIONAL_PERCEIVE_KWARGS)
+    return {k for k in _OPTIONAL_PERCEIVE_KWARGS if k in sig.parameters}
+
+
 async def run_variant(variant_name, perceive_fn, testset, references, max_timepoints,
                       target_stages=None):
     """Run a single variant. Returns accuracy and full report dict.
@@ -109,6 +121,7 @@ async def run_variant(variant_name, perceive_fn, testset, references, max_timepo
     started_at = datetime.now()
     all_predictions = []
     embryo_results = []
+    extra_kwargs = _accepted_optional_kwargs(perceive_fn)
 
     for embryo_id, tp_iter in testset.iter_all():
         logger.info(f"[{variant_name}] Starting {embryo_id}")
@@ -135,6 +148,7 @@ async def run_variant(variant_name, perceive_fn, testset, references, max_timepo
                     references=references,
                     history=history,
                     timepoint=tc.timepoint,
+                    **{k: getattr(tc, k) for k in extra_kwargs},
                 )
             except Exception as e:
                 logger.error(f"[{variant_name}/{embryo_id}] T{tc.timepoint} error: {e}")