[DRAFT] Benchmark platform and runtime artifact foundation

benchmark/__init__.py

@@ -0,0 +1,65 @@
+"""Public API for the benchmark platform."""
+
+import openhcs as _openhcs_dependency_bootstrap
+
+from benchmark.contracts.dataset import DatasetSpec, AcquiredDataset
+from benchmark.contracts.metric import MetricCollector
+from benchmark.contracts.tool_adapter import (
+    BenchmarkResult,
+    ToolAdapter,
+    ToolAdapterError,
+    ToolExecutionError,
+    ToolNotInstalledError,
+    ToolVersionError,
+)
+from benchmark.datasets.registry import BBBC021_SINGLE_PLATE, get_dataset_spec, DATASET_REGISTRY
+from benchmark.datasets.acquire import acquire_dataset, DatasetAcquisitionError
+from benchmark.metrics.time import TimeMetric
+from benchmark.metrics.memory import MemoryMetric
+from benchmark.pipelines.registry import (
+    PipelineSpec,
+    NUCLEI_SEGMENTATION,
+    get_pipeline_spec,
+    PIPELINE_REGISTRY,
+)
+from benchmark.adapters.openhcs import OpenHCSAdapter
+from benchmark.adapters.cellprofiler import CellProfilerAdapter
+from benchmark.runner import (
+    CellProfilerCompatibilityResult,
+    run_benchmark,
+    run_cellprofiler_compatibility_benchmark,
+)
+
+__all__ = [
+    # Contracts
+    "DatasetSpec",
+    "AcquiredDataset",
+    "MetricCollector",
+    "BenchmarkResult",
+    "ToolAdapter",
+    "ToolAdapterError",
+    "ToolExecutionError",
+    "ToolNotInstalledError",
+    "ToolVersionError",
+    # Datasets
+    "DatasetAcquisitionError",
+    "acquire_dataset",
+    "BBBC021_SINGLE_PLATE",
+    "DATASET_REGISTRY",
+    "get_dataset_spec",
+    # Pipelines
+    "PipelineSpec",
+    "NUCLEI_SEGMENTATION",
+    "PIPELINE_REGISTRY",
+    "get_pipeline_spec",
+    # Metrics
+    "TimeMetric",
+    "MemoryMetric",
+    # Adapters
+    "CellProfilerAdapter",
+    "OpenHCSAdapter",
+    # Runner
+    "CellProfilerCompatibilityResult",
+    "run_benchmark",
+    "run_cellprofiler_compatibility_benchmark",
+]

benchmark/adapters/__init__.py

@@ -0,0 +1,6 @@
+"""Tool adapters."""
+
+from benchmark.adapters.cellprofiler import CellProfilerAdapter
+from benchmark.adapters.openhcs import OpenHCSAdapter
+
+__all__ = ["CellProfilerAdapter", "OpenHCSAdapter"]

benchmark/adapters/cellprofiler.py

@@ -0,0 +1,194 @@
+"""Native CellProfiler tool adapter."""
+
+from __future__ import annotations
+
+import shutil
+import subprocess
+from contextlib import ExitStack
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+from benchmark.adapters.cppipe_source import (
+    CPPipeSourceRequest,
+    resolve_cppipe_source,
+)
+from benchmark.contracts.metric import MetricCollector
+from benchmark.contracts.tool_adapter import (
+    BenchmarkResult,
+    ToolAdapter,
+    ToolExecutionError,
+    ToolNotInstalledError,
+)
+from openhcs.core.runtime_equivalence import RuntimeOutputSnapshot
+
+
+@dataclass(frozen=True, slots=True)
+class CellProfilerRunRequest:
+    """Authoritative native CellProfiler run request."""
+
+    dataset_path: Path
+    pipeline_name: str
+    pipeline_params: dict[str, Any]
+    metrics: tuple[MetricCollector, ...]
+    output_dir: Path
+
+    @property
+    def dataset_id(self) -> str:
+        return str(self.pipeline_params.get("dataset_id", self.dataset_path.name))
+
+    @property
+    def timeout_seconds(self) -> float | None:
+        value = self.pipeline_params.get("cellprofiler_timeout_seconds")
+        if value is None:
+            return None
+        return float(value)
+
+    @property
+    def cppipe_source(self) -> CPPipeSourceRequest:
+        return CPPipeSourceRequest.from_pipeline_params(
+            dataset_id=self.dataset_id,
+            output_dir=self.output_dir,
+            pipeline_params=self.pipeline_params,
+        )
+
+
+class CellProfilerAdapter(ToolAdapter):
+    """Run a native CellProfiler `.cppipe` as the semantic reference tool."""
+
+    name = "CellProfiler"
+
+    def __init__(self, executable: str | Path | None = None) -> None:
+        self._configured_executable = Path(executable) if executable else None
+        self.version = "unknown"
+
+    def validate_installation(self) -> None:
+        """Check that the CellProfiler command-line runner is available."""
+        executable = self._cellprofiler_executable()
+        try:
+            result = subprocess.run(
+                [str(executable), "--version"],
+                capture_output=True,
+                text=True,
+                timeout=30,
+                check=False,
+            )
+        except FileNotFoundError as exc:
+            raise ToolNotInstalledError(
+                f"CellProfiler executable not found: {executable}"
+            ) from exc
+        if result.returncode != 0:
+            raise ToolExecutionError(
+                "Failed to query CellProfiler version:\n"
+                + _subprocess_output(result)
+            )
+        self.version = (result.stdout or result.stderr).strip() or "unknown"
+
+    def run(
+        self,
+        dataset_path: Path,
+        pipeline_name: str,
+        pipeline_params: dict[str, Any],
+        metrics: list[Any],
+        output_dir: Path,
+    ) -> BenchmarkResult:
+        """Execute a native CellProfiler pipeline headlessly."""
+        request = CellProfilerRunRequest(
+            dataset_path=Path(dataset_path),
+            pipeline_name=pipeline_name,
+            pipeline_params=dict(pipeline_params),
+            metrics=self._validated_metric_collectors(metrics),
+            output_dir=Path(output_dir),
+        )
+        request.output_dir.mkdir(parents=True, exist_ok=True)
+        source = resolve_cppipe_source(request.cppipe_source)
+        native_output_root = (
+            request.output_dir
+            / f"{request.dataset_path.name}_{request.pipeline_name}_native_cellprofiler"
+        )
+        native_output_root.mkdir(parents=True, exist_ok=True)
+        command = (
+            str(self._cellprofiler_executable()),
+            "-c",
+            "-r",
+            "-p",
+            str(source.path),
+            "-i",
+            str(request.dataset_path),
+            "-o",
+            str(native_output_root),
+        )
+
+        with ExitStack() as stack:
+            for metric in request.metrics:
+                stack.enter_context(metric)
+            try:
+                result = subprocess.run(
+                    command,
+                    capture_output=True,
+                    text=True,
+                    timeout=request.timeout_seconds,
+                    check=False,
+                )
+            except FileNotFoundError as exc:
+                raise ToolNotInstalledError(
+                    f"CellProfiler executable not found: {command[0]}"
+                ) from exc
+        if result.returncode != 0:
+            raise ToolExecutionError(
+                "Native CellProfiler execution failed:\n"
+                + _subprocess_output(result)
+            )
+
+        snapshot = RuntimeOutputSnapshot.from_output_root(native_output_root)
+        provenance: dict[str, Any] = {
+            "cellprofiler_version": self.version,
+            "pipeline_source": "native_cppipe",
+            "cppipe_path": str(source.path),
+            "csv_output_count": len(snapshot.tables),
+            "image_output_count": len(snapshot.images),
+        }
+        if source.reference_url is not None:
+            provenance["cppipe_reference_url"] = source.reference_url
+        return BenchmarkResult(
+            tool_name=self.name,
+            dataset_id=request.dataset_id,
+            pipeline_name=request.pipeline_name,
+            metrics={
+                metric.name: metric.get_result()
+                for metric in request.metrics
+            },
+            output_path=native_output_root,
+            success=True,
+            error_message=None,
+            provenance=provenance,
+        )
+
+    def _cellprofiler_executable(self) -> Path:
+        if self._configured_executable is not None:
+            return self._configured_executable
+        executable = shutil.which("cellprofiler")
+        if executable is None:
+            raise ToolNotInstalledError(
+                "CellProfiler executable not found in PATH."
+            )
+        return Path(executable)
+
+    def _validated_metric_collectors(
+        self,
+        metrics: list[Any],
+    ) -> tuple[MetricCollector, ...]:
+        validated_metrics: list[MetricCollector] = []
+        for metric in metrics:
+            if not isinstance(metric, MetricCollector):
+                raise ToolExecutionError(
+                    f"Metric {metric} does not extend MetricCollector"
+                )
+            validated_metrics.append(metric)
+        return tuple(validated_metrics)
+
+
+def _subprocess_output(result: subprocess.CompletedProcess[str]) -> str:
+    stdout = (result.stdout or "").strip()
+    stderr = (result.stderr or "").strip()
+    return "\n".join(part for part in (stdout, stderr) if part)

benchmark/adapters/cppipe_source.py

@@ -0,0 +1,129 @@
+"""Shared .cppipe source resolution for benchmark adapters."""
+
+from __future__ import annotations
+
+from collections.abc import Callable, Mapping
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+from urllib.parse import urlparse
+from urllib.request import urlopen
+
+from benchmark.contracts.tool_adapter import ToolExecutionError
+from benchmark.datasets.registry import get_dataset_spec
+
+
+CPPipeReferenceMaterializer = Callable[[str, Path], Path]
+
+
+@dataclass(frozen=True, slots=True)
+class CPPipeSourceRequest:
+    """Typed request for resolving a CellProfiler pipeline source."""
+
+    dataset_id: str
+    output_dir: Path
+    cppipe_path: Path | None = None
+    reference_url: str | None = None
+    reference_index: int | None = None
+
+    @classmethod
+    def from_pipeline_params(
+        cls,
+        *,
+        dataset_id: str,
+        output_dir: Path,
+        pipeline_params: Mapping[str, Any],
+    ) -> "CPPipeSourceRequest":
+        cppipe_value = pipeline_params.get("cppipe_path") or pipeline_params.get(
+            "cppipe_file"
+        )
+        reference_url = pipeline_params.get("cppipe_reference_url")
+        reference_index = pipeline_params.get("cppipe_reference_index")
+        return cls(
+            dataset_id=dataset_id,
+            output_dir=Path(output_dir),
+            cppipe_path=Path(cppipe_value) if cppipe_value else None,
+            reference_url=str(reference_url) if reference_url is not None else None,
+            reference_index=(
+                int(reference_index) if reference_index is not None else None
+            ),
+        )
+
+    def __post_init__(self) -> None:
+        if not self.dataset_id:
+            raise ValueError("CPPipeSourceRequest.dataset_id cannot be empty.")
+        object.__setattr__(self, "output_dir", Path(self.output_dir))
+        if self.reference_url == "":
+            raise ValueError("CPPipeSourceRequest.reference_url cannot be empty.")
+
+
+@dataclass(frozen=True, slots=True)
+class CPPipeSourceResolution:
+    """Resolved CellProfiler pipeline source."""
+
+    path: Path
+    reference_url: str | None = None
+
+
+def resolve_cppipe_source(
+    request: CPPipeSourceRequest,
+    *,
+    materialize_reference: CPPipeReferenceMaterializer | None = None,
+) -> CPPipeSourceResolution:
+    """Resolve a local or dataset-owned .cppipe path."""
+    materializer = materialize_reference or materialize_cppipe_reference
+    if request.cppipe_path is not None:
+        if not request.cppipe_path.exists():
+            raise ToolExecutionError(f".cppipe file not found: {request.cppipe_path}")
+        return CPPipeSourceResolution(request.cppipe_path)
+
+    reference_url = request.reference_url
+    if reference_url is None and request.reference_index is not None:
+        reference_url = reference_cppipe_url(
+            request.dataset_id,
+            request.reference_index,
+        )
+    if reference_url is None:
+        raise ToolExecutionError(
+            "CellProfiler pipeline execution requires cppipe_path, cppipe_file, "
+            "cppipe_reference_url, or cppipe_reference_index."
+        )
+
+    return CPPipeSourceResolution(
+        materializer(reference_url, request.output_dir / "cppipe_references"),
+        reference_url=reference_url,
+    )
+
+
+def reference_cppipe_url(dataset_id: str, reference_index: int) -> str:
+    """Resolve one canonical .cppipe URL from the dataset registry."""
+    try:
+        dataset_spec = get_dataset_spec(dataset_id)
+    except KeyError as exc:
+        raise ToolExecutionError(
+            f"Unknown dataset id {dataset_id!r} for cppipe reference lookup."
+        ) from exc
+    try:
+        return dataset_spec.reference_cppipe_urls[reference_index]
+    except IndexError as exc:
+        raise ToolExecutionError(
+            f"Dataset {dataset_id!r} exposes "
+            f"{len(dataset_spec.reference_cppipe_urls)} cppipe references; "
+            f"index {reference_index} is out of range."
+        ) from exc
+
+
+def materialize_cppipe_reference(
+    reference_url: str,
+    target_dir: Path,
+) -> Path:
+    """Download one canonical .cppipe file into a stable local path."""
+    target_dir.mkdir(parents=True, exist_ok=True)
+    parsed = urlparse(reference_url)
+    filename = Path(parsed.path).name or "reference.cppipe"
+    target_path = target_dir / filename
+    if target_path.exists():
+        return target_path
+    with urlopen(reference_url) as response:  # noqa: S310
+        target_path.write_bytes(response.read())
+    return target_path