[3/11] Separate protocol operations from targets and preserve semantic identities

src/lab/compiler.py

@@ -11,6 +11,7 @@
 
 import lab.documents as documents
 from lab._version import __version__
+from lab.artifacts import write_bundle
 from lab.deck import Deck
 from lab.experiments.cloning.stages.assembly import build_assembly
 from lab.experiments.cloning.stages.plating import build_plating
@@ -22,13 +23,15 @@
     Transformation,
     TransformationRequest,
 )
-from lab.model import Distribute, Mix, RecordedProtocol, TargetPlan, Transfer, encode
+from lab.model import RecordedProtocol, encode
+from lab.operations import Distribute, Mix, Transfer
 from lab.protocol import Protocol
 from lab.samples import Location, OutputManifest
+from lab.target import TargetPlan
 from lab.targets.liquid_handler import LiquidHandler
 from lab.targets.lower import lower_deck
 from lab.targets.manual import Manual
-from lab.validation import CompileError, logical_bindings, validate
+from lab.validation import CompileError, count_trace, logical_bindings, validate
 
 
 class Target(Interface):
@@ -69,6 +72,10 @@ def plan_json(self) -> str:
                     {"location": encode(location), "volume": encode(volume)}
                     for location, volume in self.final_volumes
                 ],
+                "final_counts": [
+                    {"location": encode(location), "count": count}
+                    for location, count in count_trace(self.protocol)[-1].items()
+                ],
                 "source_sha256": (
                     hashlib.sha256(self.target.source.encode()).hexdigest()
                     if self.target.source is not None
@@ -81,9 +88,44 @@ def plan_json(self) -> str:
     def digest(self) -> str:
         return hashlib.sha256(self.plan_json.encode()).hexdigest()
 
+    @property
+    def source_map(self) -> tuple[dict[str, object], ...]:
+        """One-based inclusive generated line spans for semantic operations."""
+        source = self.target.source
+        if source is None:
+            return ()
+        lines = source.splitlines()
+        markers = [
+            (index + 1, line.strip().removeprefix("# lab:step "))
+            for index, line in enumerate(lines)
+            if line.strip().startswith("# lab:step ")
+        ]
+        ends = {
+            line.strip().removeprefix("# lab:end "): index + 1
+            for index, line in enumerate(lines)
+            if line.strip().startswith("# lab:end ")
+        }
+        expected = tuple(step.identity for step in self.protocol.steps)
+        if tuple(identity for _, identity in markers) != expected or set(ends) != set(expected):
+            raise CompileError("Generated source does not map every protocol step exactly once")
+        return tuple(
+            {
+                "step": identity,
+                "file": "protocol.py",
+                "start_line": start,
+                "end_line": ends[identity],
+            }
+            for start, identity in markers
+        )
+
     @property
     def files(self) -> dict[str, str]:
-        result = {"plan.json": self.plan_json, "protocol.html": documents.render(self)}
+        result = {
+            "plan.json": self.plan_json,
+            "protocol.html": documents.render(self),
+            "protocol.json": self.protocol.semantic_json,
+            "source-map.json": canonical_json(self.source_map),
+        }
         if self.protocol.output_sample_ids:
             result["manifest.json"] = canonical_json(self.manifest.to_dict())
         if self.target.source is not None:
@@ -92,24 +134,75 @@ def files(self) -> dict[str, str]:
 
     def write(self, directory: str | Path) -> Path:
         """Write a bundle. Refuse to replace any different existing artifact."""
-        directory = Path(directory)
-        files = self.files
-        for name, text in files.items():
-            path = directory / name
-            if path.exists() and path.read_text(encoding="utf-8") != text:
-                raise FileExistsError(f"{path} already contains a different artifact")
-        directory.mkdir(parents=True, exist_ok=True)
-        for name, text in files.items():
-            (directory / name).write_text(text, encoding="utf-8")
-        return directory
+        return write_bundle(directory, self.files)
+
+
+def _compile_protocol(
+    protocol: Protocol | RecordedProtocol,
+    hardware: Target | Deck,
+    *,
+    liquid_handler: LiquidHandler | None = None,
+) -> Compilation:
+    """Compile offline for one piece of hardware.
+
+    A ``Deck`` contains shared requirements and optional Lab-owned layouts. The
+    selected backend validates and translates its layout or supported preset.
+    Concrete backend targets are also accepted for low-level integrations.
+    A document target such as ``Manual()`` has no robot, so ``liquid_handler`` is omitted.
+    """
+    recorded = protocol.snapshot() if isinstance(protocol, Protocol) else protocol
+    authored_deck = hardware if isinstance(hardware, Deck) else None
+    if isinstance(hardware, Deck):
+        if not isinstance(liquid_handler, LiquidHandler):
+            raise TypeError(
+                "Pass liquid_handler=LiquidHandler.OT2, LiquidHandler.FLEX, or LiquidHandler.STAR."
+            )
+        liquid = tuple(
+            step.volume for step in recorded.steps if isinstance(step, (Transfer, Mix, Distribute))
+        )
+        requirements = {container.id: container.labware for container in hardware.containers}
+        for resource in recorded.resources:
+            spec = requirements.get(resource.name)
+            if spec is None or (spec.rows, spec.columns) != (resource.rows, resource.columns):
+                raise CompileError(
+                    f"Deck requirements must match the protocol geometry for {resource.name}."
+                )
+            if resource.capacity > spec.capacity_ul:
+                raise CompileError(f"Protocol capacity exceeds the deck limit for {resource.name}.")
+        hardware = lower_deck(hardware, liquid_handler, liquid)
+    declared = getattr(hardware, "liquid_handler", None)
+    if isinstance(declared, LiquidHandler) and liquid_handler != declared:
+        raise TypeError(
+            f"This hardware is LiquidHandler.{declared.name}. "
+            f"Pass liquid_handler=LiquidHandler.{declared.name}."
+        )
+    if liquid_handler is not None and not isinstance(declared, LiquidHandler):
+        raise TypeError("This hardware does not name a LiquidHandler.")
+    if liquid_handler is not None and not isinstance(liquid_handler, LiquidHandler):
+        raise TypeError("Pass LiquidHandler.OT2, LiquidHandler.FLEX, or LiquidHandler.STAR.")
+    validate(recorded, logical_bindings(recorded))
+    prepared = hardware.prepare(recorded)
+    if authored_deck is not None:
+        configuration = json.loads(prepared.configuration_json)
+        configuration["lab_deck"] = encode(authored_deck)
+        prepared = replace(prepared, configuration_json=json.dumps(configuration, sort_keys=True))
+    volumes = validate(recorded, prepared.bindings)
+    return Compilation(recorded, prepared, tuple(volumes.items()))
 
 
 class _DefaultOutput:
     """Marks compile's default output directory."""
 
 
 _DEFAULT_OUTPUT = _DefaultOutput()
-_BUNDLE_FILES = ("protocol.html", "plan.json", "manifest.json", "protocol.py")
+_BUNDLE_FILES = (
+    "protocol.html",
+    "plan.json",
+    "manifest.json",
+    "protocol.py",
+    "protocol.json",
+    "source-map.json",
+)
 
 
 def _segment(value: str) -> str:
@@ -145,7 +238,7 @@ def _write_output(directory: Path, files: dict[str, str]) -> None:
 
 @overload
 def compile(
-    protocol: Protocol | Assembly | AssemblyRequest,
+    protocol: Protocol | RecordedProtocol | Assembly | AssemblyRequest,
     target: Target | None = None,
     *,
     deck: Deck | None = None,
@@ -182,6 +275,7 @@ def compile(
 def compile(
     protocol: (
         Protocol
+        | RecordedProtocol
         | Assembly
         | AssemblyRequest
         | Transformation
@@ -208,20 +302,18 @@ def compile(
     backend validates and translates its layout or supported preset. Concrete
     backend targets are also accepted for low-level integrations.
     """
+    if inputs is not None and not isinstance(
+        protocol, (Transformation, TransformationRequest, PlatingRequest)
+    ):
+        raise TypeError("Pass inputs with a transformation or plating request.")
     if isinstance(target, Deck):
         raise TypeError("Pass a deck with deck=.")
     if deck is not None and target is not None:
         raise TypeError("Pass a target or a deck.")
     if deck is None and target is None and liquid_handler is not None:
         raise TypeError("Pass deck= with liquid_handler.")
-    hardware: Target | Deck = (
-        deck if deck is not None else target if target is not None else Manual()
-    )
-    if inputs is not None and not isinstance(
-        protocol, (Transformation, TransformationRequest, PlatingRequest)
-    ):
-        raise TypeError("Pass inputs with a transformation or plating request.")
-    work: Protocol
+    hardware = deck if deck is not None else target if target is not None else Manual()
+    work: Protocol | RecordedProtocol
     if isinstance(protocol, Assembly):
         work = build_assembly(AssemblyRequest(id=protocol.id, assemblies=(protocol,)))
     elif isinstance(protocol, AssemblyRequest):
@@ -237,44 +329,7 @@ def compile(
         work = build_plating(protocol, inputs=inputs)
     else:
         work = protocol
-    recorded = work.snapshot()
-    authored_deck = hardware if isinstance(hardware, Deck) else None
-    if isinstance(hardware, Deck):
-        if not isinstance(liquid_handler, LiquidHandler):
-            raise TypeError(
-                "Pass liquid_handler=LiquidHandler.OT2, LiquidHandler.FLEX, or LiquidHandler.STAR."
-            )
-        liquid = tuple(
-            step.volume for step in recorded.steps if isinstance(step, (Transfer, Mix, Distribute))
-        )
-        requirements = {container.id: container.labware for container in hardware.containers}
-        for resource in recorded.resources:
-            spec = requirements.get(resource.name)
-            if spec is None or (spec.rows, spec.columns) != (resource.rows, resource.columns):
-                raise CompileError(
-                    f"Deck requirements must match the protocol geometry for {resource.name}."
-                )
-            if resource.capacity > spec.capacity_ul:
-                raise CompileError(f"Protocol capacity exceeds the deck limit for {resource.name}.")
-        hardware = lower_deck(hardware, liquid_handler, liquid)
-    declared = getattr(hardware, "liquid_handler", None)
-    if isinstance(declared, LiquidHandler) and liquid_handler != declared:
-        raise TypeError(
-            f"This hardware is LiquidHandler.{declared.name}. "
-            f"Pass liquid_handler=LiquidHandler.{declared.name}."
-        )
-    if liquid_handler is not None and not isinstance(declared, LiquidHandler):
-        raise TypeError("This hardware does not name a LiquidHandler.")
-    if liquid_handler is not None and not isinstance(liquid_handler, LiquidHandler):
-        raise TypeError("Pass LiquidHandler.OT2, LiquidHandler.FLEX, or LiquidHandler.STAR.")
-    validate(recorded, logical_bindings(recorded))
-    prepared = hardware.prepare(recorded)
-    if authored_deck is not None:
-        configuration = json.loads(prepared.configuration_json)
-        configuration["lab_deck"] = encode(authored_deck)
-        prepared = replace(prepared, configuration_json=json.dumps(configuration, sort_keys=True))
-    volumes = validate(recorded, prepared.bindings)
-    compilation = Compilation(recorded, prepared, tuple(volumes.items()))
+    compilation = _compile_protocol(work, hardware, liquid_handler=liquid_handler)
     directory = _output_directory(compilation, to)
     if directory is not None:
         _write_output(directory, compilation.files)

src/lab/documents.py

@@ -3,8 +3,9 @@
 from html import escape
 from typing import Any
 
-from lab.model import (
+from lab.operations import (
     Distribute,
+    ExternalPreparation,
     ManualInstruction,
     Mix,
     SetTemperature,
@@ -18,8 +19,9 @@
 
 def describe(step: Step) -> str:
     match step:
-        case Transfer(source, destination, volume, _):
-            return f"Transfer {number(volume)} µL from {source} to {destination}."
+        case Transfer(source, destination, volume, _, height):
+            placement = "" if height is None else f" at {number(height)} mm above the well bottom"
+            return f"Transfer {number(volume)} µL from {source} to {destination}{placement}."
         case Distribute(source, destinations, volume, air_gap, _):
             wells = ", ".join(str(destination) for destination in destinations)
             gap = f" Air gap {number(air_gap)} µL." if air_gap is not None else ""
@@ -48,18 +50,46 @@ def describe(step: Step) -> str:
             return f"Hold {resource} at {number(celsius)} °C."
         case ManualInstruction(text, _):
             return f"Operator: {text}"
+        case ExternalPreparation(procedure, instructions, inputs, outputs, _):
+
+            def describe_ports(ports: tuple) -> str:
+                return "; ".join(
+                    f"{port.count} unit(s) at {port.location}"
+                    if port.count
+                    else f"{number(port.volume_ul)} µL at {port.location}"
+                    for port in ports
+                )
+
+            return (
+                f"External procedure <{procedure}>: {instructions} "
+                f"Consume {describe_ports(inputs)}. Expected output: {describe_ports(outputs)}. "
+                "Record completion and measured output before treating material "
+                "as available inventory."
+            )
         case _:
             raise TypeError(f"Unsupported step: {type(step).__name__}")
 
 
 def render(compilation: Any) -> str:
     p, target = compilation.protocol, compilation.target
     bindings = {binding.location: binding for binding in target.bindings}
+    samples = {sample.id: sample for sample in p.samples}
     resources = []
     for resource in p.resources:
         fills = (
             "; ".join(
-                f"{fill.well}: {fill.material}, {number(fill.volume)} µL" for fill in resource.fills
+                [
+                    f"{fill.well}: {fill.material}, {number(fill.volume)} µL"
+                    for fill in resource.fills
+                ]
+                + [
+                    f"{place.location.well}: {samples[place.sample_id].label}, "
+                    f"{samples[place.sample_id].count} unit(s)"
+                    for place in p.placements
+                    if place.location.resource == resource.name
+                    and place.sample_id in p.input_sample_ids
+                    and samples[place.sample_id].count is not None
+                ]
             )
             or "Initially empty"
         )

src/lab/equipment.py

@@ -37,6 +37,7 @@ class TipRackModel(Enum):
     OPENTRONS_20_UL = "opentrons_20_ul"
     OPENTRONS_300_UL = "opentrons_300_ul"
     FLEX_200_UL = "flex_200_ul"
+    FLEX_50_UL = "flex_50_ul"
     HAMILTON_50_UL = "hamilton_50_ul"
     HAMILTON_300_UL = "hamilton_300_ul"
 
@@ -45,6 +46,7 @@ class PipetteModel(Enum):
     P20_SINGLE_GEN2 = "p20_single_gen2"
     P300_SINGLE_GEN2 = "p300_single_gen2"
     FLEX_1CHANNEL_1000 = "flex_1channel_1000"
+    FLEX_1CHANNEL_50 = "flex_1channel_50"
 
 
 class Mount(Enum):