Skills › Scientific Agent Skills › pylabrobot › _common.py
_common.py
742 lines · skills/pylabrobot/scripts/_common.py at 5f2bc03
1"""Dependency-free validation and planning helpers for the PyLabRobot skill.
3These helpers never import PyLabRobot, open sockets, enumerate hardware, or access
4serial/USB devices. All file inputs are bounded, local to the current working
5directory, regular files, and non-symlinks.
6"""
8from __future__ import annotations
10import csv
11import json
12import math
13import re
14import sys
15from pathlib import Path
16from typing import Any, Iterable, Sequence
18SCHEMA_VERSION = "1.0"
19PYLABROBOT_VERSION = "0.2.2"
20MAX_FILE_BYTES = 2_000_000
21MAX_JSON_DEPTH = 64
22MAX_RESOURCES = 256
23MAX_TRANSFERS = 10_000
24MAX_DIMENSION_MM = 5_000.0
25MAX_VOLUME_UL = 1_000_000.0
26MAX_RATE_UL_S = 10_000.0
27MAX_CHANNELS = 96
29NAME_RE = re.compile(r"^[A-Za-z][A-Za-z0-9_-]{0,63}$")
30WELL_RE = re.compile(r"^([A-Z]{1,2})([1-9][0-9]{0,2})$")
31INT_TEXT_RE = re.compile(r"^(0|[1-9][0-9]*)$")
33RESOURCE_KINDS = frozenset({"plate", "reservoir", "tube_rack", "tip_rack", "waste"})
34LIQUID_KINDS = frozenset({"plate", "reservoir", "tube_rack"})
35TRANSFER_HEADERS = (
36 "transfer_id",
37 "source",
38 "destination",
39 "volume_uL",
40 "tip_policy",
41 "tip_type",
42 "liquid_class",
43 "aspiration_height_mm",
44 "dispense_height_mm",
45 "aspiration_rate_uL_s",
46 "dispense_rate_uL_s",
47 "channel",
48)
51class ValidationError(ValueError):
52 """A deterministic, user-correctable validation failure."""
55def _reject_constant(value: str) -> None:
56 raise ValidationError(f"non-finite JSON number is forbidden: {value}")
59def _unique_object(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
60 result: dict[str, Any] = {}
61 for key, value in pairs:
62 if key in result:
63 raise ValidationError(f"duplicate JSON key: {key}")
64 result[key] = value
65 return result
68def safe_input_path(raw_path: str, allowed_suffixes: Sequence[str]) -> Path:
69 """Resolve a bounded local input path and reject symlinks/path traversal."""
70 base = Path.cwd().resolve()
71 candidate = Path(raw_path)
72 unresolved = candidate if candidate.is_absolute() else base / candidate
73 try:
74 lexical_relative = unresolved.relative_to(base)
75 except ValueError as exc:
76 raise ValidationError("input path must stay inside the current working directory") from exc
77 if ".." in lexical_relative.parts:
78 raise ValidationError("parent-directory traversal is forbidden")
80 cursor = base
81 for part in lexical_relative.parts:
82 cursor = cursor / part
83 if cursor.is_symlink():
84 raise ValidationError(f"symlink paths are forbidden: {raw_path}")
86 resolved = unresolved.resolve(strict=False)
87 try:
88 resolved.relative_to(base)
89 except ValueError as exc:
90 raise ValidationError("input path must stay inside the current working directory") from exc
92 if resolved.suffix.lower() not in {suffix.lower() for suffix in allowed_suffixes}:
93 allowed = ", ".join(sorted(allowed_suffixes))
94 raise ValidationError(f"input must use one of these suffixes: {allowed}")
95 if not resolved.exists() or not resolved.is_file():
96 raise ValidationError(f"input is not a regular file: {raw_path}")
97 size = resolved.stat().st_size
98 if size > MAX_FILE_BYTES:
99 raise ValidationError(f"input exceeds {MAX_FILE_BYTES} bytes")
100 return resolved
103def load_json(raw_path: str) -> dict[str, Any]:
104 path = safe_input_path(raw_path, (".json",))
105 try:
106 text = path.read_text(encoding="utf-8")
107 except UnicodeDecodeError as exc:
108 raise ValidationError("JSON input must be UTF-8") from exc
109 try:
110 data = json.loads(
111 text,
112 object_pairs_hook=_unique_object,
113 parse_constant=_reject_constant,
114 )
115 except json.JSONDecodeError as exc:
116 raise ValidationError(f"invalid JSON at line {exc.lineno}, column {exc.colno}") from exc
117 except (RecursionError, ValueError) as exc:
118 if isinstance(exc, ValidationError):
119 raise
120 raise ValidationError("JSON nesting or numeric representation exceeds supported limits") from exc
121 if not isinstance(data, dict):
122 raise ValidationError("JSON root must be an object")
123 pending = [(data, 1)]
124 while pending:
125 value, depth = pending.pop()
126 if depth > MAX_JSON_DEPTH:
127 raise ValidationError(f"JSON nesting exceeds {MAX_JSON_DEPTH} levels")
128 if isinstance(value, dict):
129 pending.extend((item, depth + 1) for item in value.values())
130 elif isinstance(value, list):
131 pending.extend((item, depth + 1) for item in value)
132 return data
135def load_csv(raw_path: str) -> list[dict[str, str]]:
136 path = safe_input_path(raw_path, (".csv",))
137 try:
138 with path.open("r", encoding="utf-8", newline="") as handle:
139 reader = csv.DictReader(handle)
140 if tuple(reader.fieldnames or ()) != TRANSFER_HEADERS:
141 raise ValidationError(
142 "CSV header must exactly match: " + ",".join(TRANSFER_HEADERS)
143 )
144 rows: list[dict[str, str]] = []
145 for line_number, row in enumerate(reader, start=2):
146 if None in row:
147 raise ValidationError(f"CSV line {line_number} has extra columns")
148 normalized = {key: (value or "").strip() for key, value in row.items()}
149 if any(value == "" for value in normalized.values()):
150 raise ValidationError(f"CSV line {line_number} contains an empty field")
151 rows.append(normalized)
152 if len(rows) > MAX_TRANSFERS:
153 raise ValidationError(f"CSV exceeds {MAX_TRANSFERS} transfer rows")
154 except UnicodeDecodeError as exc:
155 raise ValidationError("CSV input must be UTF-8") from exc
156 except csv.Error as exc:
157 raise ValidationError("CSV structure or field length exceeds supported limits") from exc
158 if not rows:
159 raise ValidationError("CSV must contain at least one transfer row")
160 return rows
163def _exact_keys(
164 obj: Any,
165 required: Iterable[str],
166 optional: Iterable[str],
167 where: str,
168) -> dict[str, Any]:
169 if not isinstance(obj, dict):
170 raise ValidationError(f"{where} must be an object")
171 required_set = set(required)
172 allowed = required_set | set(optional)
173 missing = sorted(required_set - set(obj))
174 unknown = sorted(set(obj) - allowed)
175 if missing:
176 raise ValidationError(f"{where} missing keys: {', '.join(missing)}")
177 if unknown:
178 raise ValidationError(f"{where} has unknown keys: {', '.join(unknown)}")
179 return obj
182def _name(value: Any, where: str) -> str:
183 if not isinstance(value, str) or NAME_RE.fullmatch(value) is None:
184 raise ValidationError(f"{where} must match {NAME_RE.pattern}")
185 return value
188def _number(
189 value: Any,
190 where: str,
191 *,
192 minimum: float = 0.0,
193 maximum: float,
194 minimum_inclusive: bool = True,
195) -> float:
196 if isinstance(value, bool) or not isinstance(value, (int, float)):
197 raise ValidationError(f"{where} must be a number")
198 try:
199 result = float(value)
200 except OverflowError as exc:
201 raise ValidationError(f"{where} must be a finite bounded number") from exc
202 if not math.isfinite(result):
203 raise ValidationError(f"{where} must be finite")
204 too_small = result < minimum if minimum_inclusive else result <= minimum
205 if too_small or result > maximum:
206 bracket = "[" if minimum_inclusive else "("
207 raise ValidationError(f"{where} must be in {bracket}{minimum}, {maximum}]")
208 return result
211def _integer(value: Any, where: str, minimum: int, maximum: int) -> int:
212 if isinstance(value, bool) or not isinstance(value, int):
213 raise ValidationError(f"{where} must be an integer")
214 if value < minimum or value > maximum:
215 raise ValidationError(f"{where} must be in [{minimum}, {maximum}]")
216 return value
219def _string_list(value: Any, where: str, maximum_items: int = 256) -> list[str]:
220 if not isinstance(value, list) or not value or len(value) > maximum_items:
221 raise ValidationError(f"{where} must be a non-empty list with at most {maximum_items} items")
222 result: list[str] = []
223 for index, item in enumerate(value):
224 result.append(_name(item, f"{where}[{index}]"))
225 if len(set(result)) != len(result):
226 raise ValidationError(f"{where} must not contain duplicates")
227 return result
230def _xyz(value: Any, where: str, *, positive: bool) -> dict[str, float]:
231 obj = _exact_keys(value, ("x", "y", "z"), (), where)
232 minimum_inclusive = not positive
233 return {
234 axis: _number(
235 obj[axis],
236 f"{where}.{axis}",
237 minimum=0.0,
238 maximum=MAX_DIMENSION_MM,
239 minimum_inclusive=minimum_inclusive,
240 )
241 for axis in ("x", "y", "z")
242 }
245def row_index(row_letters: str) -> int:
246 index = 0
247 for character in row_letters:
248 index = index * 26 + ord(character) - ord("A") + 1
249 return index - 1
252def validate_well(well: str, rows: int, columns: int, where: str) -> str:
253 match = WELL_RE.fullmatch(well)
254 if match is None:
255 raise ValidationError(f"{where} must be an uppercase well such as A1")
256 row, column_text = match.groups()
257 if row_index(row) >= rows or int(column_text) > columns:
258 raise ValidationError(f"{where}={well} lies outside the declared grid")
259 return well
262def _validate_grid(value: Any, kind: str, where: str) -> dict[str, Any]:
263 if kind == "tip_rack":
264 obj = _exact_keys(value, ("rows", "columns", "tip_capacity_uL"), (), where)
265 return {
266 "rows": _integer(obj["rows"], f"{where}.rows", 1, 384),
267 "columns": _integer(obj["columns"], f"{where}.columns", 1, 384),
268 "tip_capacity_uL": _number(
269 obj["tip_capacity_uL"],
270 f"{where}.tip_capacity_uL",
271 maximum=MAX_VOLUME_UL,
272 minimum_inclusive=False,
273 ),
274 }
275 obj = _exact_keys(
276 value,
277 ("rows", "columns", "well_capacity_uL", "dead_volume_uL", "well_depth_mm"),
278 (),
279 where,
280 )
281 capacity = _number(
282 obj["well_capacity_uL"],
283 f"{where}.well_capacity_uL",
284 maximum=MAX_VOLUME_UL,
285 minimum_inclusive=False,
286 )
287 dead = _number(
288 obj["dead_volume_uL"],
289 f"{where}.dead_volume_uL",
290 maximum=capacity,
291 )
292 if dead >= capacity:
293 raise ValidationError(f"{where}.dead_volume_uL must be below well capacity")
294 return {
295 "rows": _integer(obj["rows"], f"{where}.rows", 1, 384),
296 "columns": _integer(obj["columns"], f"{where}.columns", 1, 384),
297 "well_capacity_uL": capacity,
298 "dead_volume_uL": dead,
299 "well_depth_mm": _number(
300 obj["well_depth_mm"],
301 f"{where}.well_depth_mm",
302 maximum=MAX_DIMENSION_MM,
303 minimum_inclusive=False,
304 ),
305 }
308def validate_manifest(data: dict[str, Any]) -> dict[str, Any]:
309 """Validate and normalize protocol manifest schema v1.0."""
310 top = _exact_keys(
311 data,
312 ("schema_version", "protocol_id", "mode", "units", "deck", "resources", "constraints"),
313 (),
314 "manifest",
315 )
316 if top["schema_version"] != SCHEMA_VERSION:
317 raise ValidationError(f"schema_version must be {SCHEMA_VERSION!r}")
318 protocol_id = _name(top["protocol_id"], "protocol_id")
319 if top["mode"] != "offline":
320 raise ValidationError("mode must be 'offline'; live execution is not supported")
322 units = _exact_keys(top["units"], ("volume", "length", "rate", "time"), (), "units")
323 expected_units = {"volume": "uL", "length": "mm", "rate": "uL/s", "time": "s"}
324 if units != expected_units:
325 raise ValidationError(f"units must exactly equal {expected_units}")
327 deck_obj = _exact_keys(top["deck"], ("name", "size_mm"), (), "deck")
328 deck = {
329 "name": _name(deck_obj["name"], "deck.name"),
330 "size_mm": _xyz(deck_obj["size_mm"], "deck.size_mm", positive=True),
331 }
333 resources_value = top["resources"]
334 if (
335 not isinstance(resources_value, list)
336 or not resources_value
337 or len(resources_value) > MAX_RESOURCES
338 ):
339 raise ValidationError(f"resources must contain 1 to {MAX_RESOURCES} objects")
341 resources: list[dict[str, Any]] = []
342 resource_names: set[str] = set()
343 for index, raw_resource in enumerate(resources_value):
344 where = f"resources[{index}]"
345 base = _exact_keys(
346 raw_resource,
347 ("name", "kind", "location_mm", "size_mm"),
348 ("grid", "initial_volumes_uL", "tip_type", "tips_available"),
349 where,
350 )
351 name = _name(base["name"], f"{where}.name")
352 if name in resource_names:
353 raise ValidationError(f"duplicate resource name: {name}")
354 resource_names.add(name)
355 kind = base["kind"]
356 if not isinstance(kind, str) or kind not in RESOURCE_KINDS:
357 raise ValidationError(f"{where}.kind must be one of {sorted(RESOURCE_KINDS)}")
358 normalized: dict[str, Any] = {
359 "name": name,
360 "kind": kind,
361 "location_mm": _xyz(base["location_mm"], f"{where}.location_mm", positive=False),
362 "size_mm": _xyz(base["size_mm"], f"{where}.size_mm", positive=True),
363 }
365 if kind == "waste":
366 forbidden = {"grid", "initial_volumes_uL", "tip_type", "tips_available"} & set(base)
367 if forbidden:
368 raise ValidationError(f"{where} waste has forbidden keys: {', '.join(sorted(forbidden))}")
369 elif kind == "tip_rack":
370 if "grid" not in base or "tip_type" not in base or "tips_available" not in base:
371 raise ValidationError(f"{where} tip_rack requires grid, tip_type, and tips_available")
372 if "initial_volumes_uL" in base:
373 raise ValidationError(f"{where} tip_rack cannot define initial volumes")
374 grid = _validate_grid(base["grid"], kind, f"{where}.grid")
375 tips = _string_list(base["tips_available"], f"{where}.tips_available", 384 * 384)
376 for tip_index, tip in enumerate(tips):
377 validate_well(tip, grid["rows"], grid["columns"], f"{where}.tips_available[{tip_index}]")
378 normalized.update(
379 grid=grid,
380 tip_type=_name(base["tip_type"], f"{where}.tip_type"),
381 tips_available=tips,
382 )
383 else:
384 if "grid" not in base or "initial_volumes_uL" not in base:
385 raise ValidationError(f"{where} {kind} requires grid and initial_volumes_uL")
386 if "tip_type" in base or "tips_available" in base:
387 raise ValidationError(f"{where} {kind} cannot define tip fields")
388 grid = _validate_grid(base["grid"], kind, f"{where}.grid")
389 volumes = base["initial_volumes_uL"]
390 if not isinstance(volumes, dict):
391 raise ValidationError(f"{where}.initial_volumes_uL must be an object")
392 normalized_volumes: dict[str, float] = {}
393 for well, volume in volumes.items():
394 validate_well(well, grid["rows"], grid["columns"], f"{where}.initial_volumes_uL key")
395 normalized_volumes[well] = _number(
396 volume,
397 f"{where}.initial_volumes_uL.{well}",
398 maximum=grid["well_capacity_uL"],
399 )
400 normalized.update(grid=grid, initial_volumes_uL=normalized_volumes)
401 resources.append(normalized)
403 if not any(resource["kind"] == "waste" for resource in resources):
404 raise ValidationError("manifest must declare at least one waste resource")
406 constraints_obj = _exact_keys(
407 top["constraints"],
408 (
409 "allowed_liquid_classes",
410 "allowed_tip_types",
411 "channels",
412 "requires_human_confirmation",
413 "max_transfer_uL",
414 "max_rate_uL_s",
415 ),
416 (),
417 "constraints",
418 )
419 if constraints_obj["requires_human_confirmation"] is not True:
420 raise ValidationError("constraints.requires_human_confirmation must be true")
421 constraints = {
422 "allowed_liquid_classes": _string_list(
423 constraints_obj["allowed_liquid_classes"], "constraints.allowed_liquid_classes"
424 ),
425 "allowed_tip_types": _string_list(
426 constraints_obj["allowed_tip_types"], "constraints.allowed_tip_types"
427 ),
428 "channels": _integer(
429 constraints_obj["channels"], "constraints.channels", 1, MAX_CHANNELS
430 ),
431 "requires_human_confirmation": True,
432 "max_transfer_uL": _number(
433 constraints_obj["max_transfer_uL"],
434 "constraints.max_transfer_uL",
435 maximum=MAX_VOLUME_UL,
436 minimum_inclusive=False,
437 ),
438 "max_rate_uL_s": _number(
439 constraints_obj["max_rate_uL_s"],
440 "constraints.max_rate_uL_s",
441 maximum=MAX_RATE_UL_S,
442 minimum_inclusive=False,
443 ),
444 }
446 for resource in resources:
447 if resource["kind"] == "tip_rack" and resource["tip_type"] not in constraints["allowed_tip_types"]:
448 raise ValidationError(
449 f"tip rack {resource['name']} uses a tip_type outside constraints.allowed_tip_types"
450 )
452 return {
453 "schema_version": SCHEMA_VERSION,
454 "protocol_id": protocol_id,
455 "mode": "offline",
456 "units": expected_units,
457 "deck": deck,
458 "resources": resources,
459 "constraints": constraints,
460 }
463def _parse_float_text(
464 value: str,
465 where: str,
466 *,
467 maximum: float,
468 minimum: float = 0.0,
469 minimum_inclusive: bool = True,
470) -> float:
471 try:
472 parsed = float(value)
473 except ValueError as exc:
474 raise ValidationError(f"{where} must be a decimal number") from exc
475 return _number(
476 parsed,
477 where,
478 minimum=minimum,
479 maximum=maximum,
480 minimum_inclusive=minimum_inclusive,
481 )
484def _parse_int_text(value: str, where: str, minimum: int, maximum: int) -> int:
485 if INT_TEXT_RE.fullmatch(value) is None:
486 raise ValidationError(f"{where} must be an unsigned base-10 integer")
487 try:
488 parsed = int(value)
489 except ValueError as exc:
490 raise ValidationError(f"{where} integer exceeds supported limits") from exc
491 if parsed < minimum or parsed > maximum:
492 raise ValidationError(f"{where} must be in [{minimum}, {maximum}]")
493 return parsed
496def _endpoint(value: str, where: str) -> tuple[str, str]:
497 if value.count(":") != 1:
498 raise ValidationError(f"{where} must use resource:WELL syntax")
499 resource, well = value.split(":")
500 _name(resource, f"{where} resource")
501 if WELL_RE.fullmatch(well) is None:
502 raise ValidationError(f"{where} well must be uppercase, for example A1")
503 return resource, well
506def validate_transfers(rows: list[dict[str, str]]) -> list[dict[str, Any]]:
507 normalized: list[dict[str, Any]] = []
508 identifiers: set[str] = set()
509 for index, row in enumerate(rows):
510 where = f"transfers[{index}]"
511 transfer_id = _name(row["transfer_id"], f"{where}.transfer_id")
512 if transfer_id in identifiers:
513 raise ValidationError(f"duplicate transfer_id: {transfer_id}")
514 identifiers.add(transfer_id)
515 if row["tip_policy"] != "new":
516 raise ValidationError(f"{where}.tip_policy must be 'new' for contamination safety")
517 normalized.append(
518 {
519 "transfer_id": transfer_id,
520 "source": _endpoint(row["source"], f"{where}.source"),
521 "destination": _endpoint(row["destination"], f"{where}.destination"),
522 "volume_uL": _parse_float_text(
523 row["volume_uL"],
524 f"{where}.volume_uL",
525 maximum=MAX_VOLUME_UL,
526 minimum_inclusive=False,
527 ),
528 "tip_policy": "new",
529 "tip_type": _name(row["tip_type"], f"{where}.tip_type"),
530 "liquid_class": _name(row["liquid_class"], f"{where}.liquid_class"),
531 "aspiration_height_mm": _parse_float_text(
532 row["aspiration_height_mm"],
533 f"{where}.aspiration_height_mm",
534 maximum=MAX_DIMENSION_MM,
535 ),
536 "dispense_height_mm": _parse_float_text(
537 row["dispense_height_mm"],
538 f"{where}.dispense_height_mm",
539 maximum=MAX_DIMENSION_MM,
540 ),
541 "aspiration_rate_uL_s": _parse_float_text(
542 row["aspiration_rate_uL_s"],
543 f"{where}.aspiration_rate_uL_s",
544 maximum=MAX_RATE_UL_S,
545 minimum_inclusive=False,
546 ),
547 "dispense_rate_uL_s": _parse_float_text(
548 row["dispense_rate_uL_s"],
549 f"{where}.dispense_rate_uL_s",
550 maximum=MAX_RATE_UL_S,
551 minimum_inclusive=False,
552 ),
553 "channel": _parse_int_text(
554 row["channel"], f"{where}.channel", 0, MAX_CHANNELS - 1
555 ),
556 }
557 )
558 return normalized
561def geometry_report(manifest: dict[str, Any]) -> dict[str, Any]:
562 deck_size = manifest["deck"]["size_mm"]
563 resources = manifest["resources"]
564 out_of_bounds: list[dict[str, Any]] = []
565 for resource in resources:
566 axes: list[str] = []
567 for axis in ("x", "y", "z"):
568 start = resource["location_mm"][axis]
569 end = start + resource["size_mm"][axis]
570 if start < 0 or end > deck_size[axis]:
571 axes.append(axis)
572 if axes:
573 out_of_bounds.append({"resource": resource["name"], "axes": axes})
575 collisions: list[dict[str, str]] = []
576 for left_index, left in enumerate(resources):
577 for right in resources[left_index + 1 :]:
578 overlaps = all(
579 min(
580 left["location_mm"][axis] + left["size_mm"][axis],
581 right["location_mm"][axis] + right["size_mm"][axis],
582 )
583 > max(left["location_mm"][axis], right["location_mm"][axis])
584 for axis in ("x", "y", "z")
585 )
586 if overlaps:
587 collisions.append({"resource_a": left["name"], "resource_b": right["name"]})
589 return {
590 "ok": not out_of_bounds and not collisions,
591 "out_of_bounds": out_of_bounds,
592 "collisions": collisions,
593 "model": "axis-aligned static bounding boxes in millimetres",
594 "limitations": [
595 "does not model rotations, gripper/channel motion envelopes, lids, tubing, cables, or tolerances",
596 "passing this check is not evidence that a physical run is collision-free",
597 ],
598 }
601def plan_transfers(
602 manifest: dict[str, Any],
603 transfers: list[dict[str, Any]],
604) -> dict[str, Any]:
605 resources = {resource["name"]: resource for resource in manifest["resources"]}
606 constraints = manifest["constraints"]
607 liquid_classes = set(constraints["allowed_liquid_classes"])
608 tip_types = set(constraints["allowed_tip_types"])
609 waste = next(resource["name"] for resource in manifest["resources"] if resource["kind"] == "waste")
611 available_tips: dict[str, list[tuple[str, str, float]]] = {tip_type: [] for tip_type in tip_types}
612 for resource in manifest["resources"]:
613 if resource["kind"] == "tip_rack":
614 for well in resource["tips_available"]:
615 available_tips[resource["tip_type"]].append(
616 (resource["name"], well, resource["grid"]["tip_capacity_uL"])
617 )
619 ledger: dict[tuple[str, str], float] = {}
620 known_liquid_state: set[tuple[str, str]] = set()
621 for resource in manifest["resources"]:
622 if resource["kind"] in LIQUID_KINDS:
623 for well, volume in resource["initial_volumes_uL"].items():
624 ledger[(resource["name"], well)] = volume
625 known_liquid_state.add((resource["name"], well))
627 operations: list[dict[str, Any]] = []
628 tip_cursor: dict[str, int] = {tip_type: 0 for tip_type in tip_types}
629 for index, transfer in enumerate(transfers):
630 where = f"transfers[{index}]"
631 source_name, source_well = transfer["source"]
632 destination_name, destination_well = transfer["destination"]
633 if source_name not in resources or destination_name not in resources:
634 raise ValidationError(f"{where} references an unknown resource")
635 source_resource = resources[source_name]
636 destination_resource = resources[destination_name]
637 if source_resource["kind"] not in LIQUID_KINDS:
638 raise ValidationError(f"{where}.source must reference liquid-holding labware")
639 if destination_resource["kind"] not in LIQUID_KINDS:
640 raise ValidationError(f"{where}.destination must reference liquid-holding labware")
641 validate_well(
642 source_well,
643 source_resource["grid"]["rows"],
644 source_resource["grid"]["columns"],
645 f"{where}.source",
646 )
647 validate_well(
648 destination_well,
649 destination_resource["grid"]["rows"],
650 destination_resource["grid"]["columns"],
651 f"{where}.destination",
652 )
653 source_key = (source_name, source_well)
654 destination_key = (destination_name, destination_well)
655 if source_key == destination_key:
656 raise ValidationError(f"{where} source and destination must differ")
657 if source_key not in known_liquid_state:
658 raise ValidationError(f"{where}.source requires declared starting volume")
659 if destination_key not in known_liquid_state:
660 raise ValidationError(f"{where}.destination requires declared starting volume, including zero")
662 volume = transfer["volume_uL"]
663 if volume > constraints["max_transfer_uL"]:
664 raise ValidationError(f"{where}.volume_uL exceeds constraints.max_transfer_uL")
665 if transfer["liquid_class"] not in liquid_classes:
666 raise ValidationError(f"{where}.liquid_class is not allowlisted")
667 if transfer["tip_type"] not in tip_types:
668 raise ValidationError(f"{where}.tip_type is not allowlisted")
669 if transfer["channel"] >= constraints["channels"]:
670 raise ValidationError(f"{where}.channel exceeds the declared channel count")
671 if (
672 transfer["aspiration_rate_uL_s"] > constraints["max_rate_uL_s"]
673 or transfer["dispense_rate_uL_s"] > constraints["max_rate_uL_s"]
674 ):
675 raise ValidationError(f"{where} rate exceeds constraints.max_rate_uL_s")
676 if transfer["aspiration_height_mm"] > source_resource["grid"]["well_depth_mm"]:
677 raise ValidationError(f"{where}.aspiration_height_mm exceeds source well depth")
678 if transfer["dispense_height_mm"] > destination_resource["grid"]["well_depth_mm"]:
679 raise ValidationError(f"{where}.dispense_height_mm exceeds destination well depth")
681 remaining = ledger[source_key] - volume
682 if remaining < source_resource["grid"]["dead_volume_uL"]:
683 raise ValidationError(f"{where} would aspirate below source dead volume")
684 destination_volume = ledger[destination_key] + volume
685 if destination_volume > destination_resource["grid"]["well_capacity_uL"]:
686 raise ValidationError(f"{where} would exceed destination well capacity")
688 tips = available_tips.get(transfer["tip_type"], [])
689 cursor = tip_cursor[transfer["tip_type"]]
690 if cursor >= len(tips):
691 raise ValidationError(f"{where} has no unused compatible tip")
692 tip_rack, tip_well, tip_capacity = tips[cursor]
693 if volume > tip_capacity:
694 raise ValidationError(f"{where}.volume_uL exceeds selected tip capacity")
695 tip_cursor[transfer["tip_type"]] += 1
697 ledger[source_key] = remaining
698 ledger[destination_key] = destination_volume
699 known_liquid_state.add(destination_key)
700 operations.append(
701 {
702 **transfer,
703 "source": f"{source_name}:{source_well}",
704 "destination": f"{destination_name}:{destination_well}",
705 "assigned_tip": f"{tip_rack}:{tip_well}",
706 "drop_tip_at": waste,
707 }
708 )
710 final_ledger = [
711 {"location": f"{resource}:{well}", "volume_uL": round(volume, 9)}
712 for (resource, well), volume in sorted(ledger.items())
713 ]
714 used_tips = sum(tip_cursor.values())
715 return {
716 "ok": True,
717 "protocol_id": manifest["protocol_id"],
718 "operations": operations,
719 "final_volume_ledger": final_ledger,
720 "tip_summary": {
721 "policy": "one new tip per transfer",
722 "used": used_tips,
723 "remaining": sum(len(available_tips[key]) - tip_cursor[key] for key in tip_cursor),
724 },
725 "bookkeeping_warning": (
726 "Volumes and tips are planned state only; they do not detect physical liquid or tip presence."
727 ),
728 }
731def emit_json(payload: Any, *, stream: Any = None) -> None:
732 if stream is None:
733 stream = sys.stdout
734 print(json.dumps(payload, indent=2, sort_keys=True, allow_nan=False), file=stream)
737def emit_error(error: Exception) -> int:
738 emit_json(
739 {"ok": False, "error": type(error).__name__, "message": str(error)},
740 stream=sys.stderr,
741 )
742 return 2