Task parsing and validation

Validate task scripts, expand configured grid targets, and resolve named poses. Device readiness, planning, and execution occur later.

beambot.core.task_script · Source

Task execution · Guide · Symbol index

How it is used

The orchestrator calls parse_task_script() before setting up execution. Supply JSON text and the beamline’s gripper, pose-registry, and vision-target settings; it returns (start_gripper, tasks, poses_json). The input dry_run flag controls validation and is not part of the return value.

expand_grid_target() is shared by parser pickup macros and MCP grid-target builders. The motion validators check MoveTo and flange-offset fields at other entry points. These functions do not check reachability, collisions, hardware readiness, or every task-specific field.

Start here

Callable	Purpose
parse_task_script	Parse and validate a task document, expand macros, and fill missing named poses from the registry. Validate start/explicit end-effector names, supported task types, MoveTo/flange offsets, and dry-run restrictions. Unresolved pose names remain for downstream resolution.
expand_grid_target	Expand a configured grid target into optional scan motion, marker approach, and relative moves. Return the expanded tasks plus the selected element and row/column displacement metadata.
moveto_goal_error	Check that exactly one named, relative or Cartesian target is selected and that planner, direction, distance and Cartesian values satisfy the MoveTo contract.
flange_offset_error	Validate either an absent flange-frame offset or a supported direction with a finite positive distance. Booleans are rejected as distances.

Example

This example parses an inline pose without reading a registry or starting ROS. The zero-valued pose illustrates the document structure; it is not an operating pose recommendation. Parsing does not plan or execute motion.

import json
from beambot.core.task_script import parse_task_script

document = {
    "start_gripper": "none",
    "poses": {"example_pose": [0, 0, 0, 0, 0, 0]},
    "tasks": [{"task_type": "moveto", "target": "example_pose"}],
}
gripper, tasks, poses_json = parse_task_script(
    json.dumps(document),
    dry_run=True,
    grippers={"none": {}},
    poses_file="unused.yaml",
    vision_targets={},
)
assert gripper == "none"
assert tasks == document["tasks"]
assert json.loads(poses_json) == document["poses"]

In the orchestrator, the gripper mapping and registry path come from the selected beamline configuration.

Functions

`parse_task_script()`

Parse and validate a task document, expand macros, and fill missing named poses from the registry. Validate start/explicit end-effector names, supported task types, MoveTo/flange offsets, and dry-run restrictions. Unresolved pose names remain for downstream resolution.

Entry point · Source

parse_task_script(
    full_json: str,
    *,
    dry_run: bool,
    grippers: Mapping[str, Any],
    poses_file: str,
    vision_targets: Mapping[str, Any],
    on_info: Callable[[str], None] | None = None,
    on_warning: Callable[[str], None] | None = None,
) -> tuple[str, list[dict[str, Any]], str]

Parameter	Meaning
`full_json`	JSON text containing start_gripper, nonempty tasks and optional poses.
`dry_run`	Restrict expanded tasks to moveto and end_effector when true; used for validation only and not returned.
`grippers`	Mapping of accepted start_gripper and explicitly selected end_effector_type names.
`poses_file`	YAML registry filename consulted only when referenced names are absent from inline poses.
`vision_targets`	Mapping of vision target configurations used for pickup macros; must be a mapping even without macros.
`on_info`	Optional callback announcing resolved registry poses.
`on_warning`	Optional callback for registry-read failures.

Returns: (start_gripper, expanded_tasks, poses_json). The caller retains its requested dry_run flag separately.

Errors: Raises ValueError for JSON/envelope/task/motion/macro errors, unknown or non-string explicit end_effector_type values, and unsupported dry-run tasks. Pose container shapes and many task-specific fields are not fully validated and may raise downstream Python errors. Callback exceptions propagate.

Side effects: May read the poses registry and invoke logging callbacks; modifies only the newly parsed inline pose mapping.

Referenced in source (1)

MTCOrchestratorServer._execute — beambot.action_servers.orchestrator

`expand_grid_target()`

Expand a configured grid target into optional scan motion, marker approach, and relative moves. Return the expanded tasks plus the selected element and row/column displacement metadata.

Function · Source

expand_grid_target(
    target_name: str,
    vision_targets: Mapping[str, Any],
    task: Mapping[str, Any],
    task_path: str = 'task',
) -> tuple[list[dict[str, Any]], dict[str, Any]]

Parameter	Meaning
`target_name`	Grid target key in vision_targets, such as tip_rack or vial_rack.
`vision_targets`	Mapping of target names to grid configurations.
`task`	Row/col or element_index selection with optional config overrides. task_type=pickup_vial also enables optional SUCK, EXPEL, or RINSE insertion.
`task_path`	Label used in validation errors, default task; parser callers pass tasks[index]. It is not a filesystem path.

Returns: (expanded_tasks, element). element contains zero-based row, col, row-major index, and moves mapping col/row to (direction, distance_metres). Axis offsets no larger than 1 micrometre are omitted from the task list but remain in metadata.

Errors: Raises ValueError for missing/invalid grid configuration, marker ID, addresses, directions, pitches, offsets, move lists, or vial-operation fields. Pitch defaults to 0.009 m; increasing indices default opposite the configured column/row directions. Vial operations require a final retreat move; RINSE emits SUCK/EXPEL pairs followed by a final SUCK.

Side effects: Builds new task and metadata containers without file access, ROS initialization, or motion. The original target/task mappings are not modified.

Example

from beambot.core.task_script import expand_grid_target

targets = {"rack": {
    "mode": "grid", "marker_id": 4,
    "grid": {"rows": 2, "cols": 3},
    "moves": ["column_offset", "row_offset"],
}}
tasks, element = expand_grid_target("rack", targets, {"row": 1, "col": 1})
assert tasks[0] == {"task_type": "vision_moveto", "tag_id": 4}
assert element["index"] == 4
assert element["moves"] == {"col": ("right", 0.009), "row": ("down", 0.009)}

This example builds task data only. Its grid geometry is illustrative.

Referenced in source (2)

_expand_task_macros — beambot.core.task_script
_build_vision_target_tasks — src.beambot.mcp_server.beambot_mcp_server

`moveto_goal_error()`

Check that exactly one named, relative or Cartesian target is selected and that planner, direction, distance and Cartesian values satisfy the MoveTo contract.

Function · Source

moveto_goal_error(
    *,
    target: Any = '',
    direction: Any = '',
    distance: Any = 0.0,
    cartesian_target: Any = (),
    planning_type: Any = '',
) -> str | None

Parameter	Meaning
`target`	Named target string; empty selects no named goal.
`direction`	Relative direction alias; empty selects no relative direction.
`distance`	Relative travel in metres; must be finite and positive when relative mode is selected.
`cartesian_target`	Sequence of exactly three XYZ metre values or six XYZ metres plus roll/pitch/yaw degrees; empty selects no Cartesian goal.
`planning_type`	Planner selector: empty, auto, joint, cartesian, pilz or pilz_ptp.

Returns: None for a valid goal; otherwise the first explanatory error string.

Errors: Ordinary invalid scalar/sequence inputs produce messages, not exceptions. Objects with ambiguous truth values or unsupported comparisons can still raise.

Example

from beambot.core.task_script import moveto_goal_error
assert moveto_goal_error(direction="up", distance=0.05) is None
assert "exactly one" in moveto_goal_error()

Referenced in source (2)

parse_task_script — beambot.core.task_script
MoveToStages.add_to_task — beambot.stages.move_to_stages

`flange_offset_error()`

Validate either an absent flange-frame offset or a supported direction with a finite positive distance. Booleans are rejected as distances.

Function · Source

flange_offset_error(*, direction: Any='', distance: Any=0.0) -> str | None

Parameter	Meaning
`direction`	Direction alias; empty is valid only with zero distance.
`distance`	Offset distance in metres; zero denotes absence when direction is empty.

Returns: None when absent or valid; otherwise the first explanatory error string.

Errors: No explicit exceptions for normal malformed inputs; reports them as strings.

Referenced in source (4)

parse_task_script — beambot.core.task_script
PickSampleStages.run — beambot.stages.pick_sample_stages
PlaceSampleStages.run — beambot.stages.place_sample_stages
VisionTaskStages.run — beambot.stages.vision_task_stages

Internal callbacks and helpers

`_load_poses_registry()`

Load an existing YAML pose registry, treating unavailable or unreadable content as a nonfatal lookup failure.

Internal helper · Source

_load_poses_registry(
    poses_file: str,
    on_warning: Callable[[str], None] | None,
) -> dict[str, Any]

Parameter	Meaning
`poses_file`	YAML filename; a nonexistent regular file yields an empty mapping.
`on_warning`	Optional callback receiving a warning when opening or parsing fails.

Returns: Parsed truthy YAML value, or {} for a missing, empty or failed registry. The root type is not checked here.

Errors: Catches read/parse exceptions; a warning callback that raises still propagates.

Side effects: Checks and reads a file; may invoke the warning callback.

Referenced in source (1)

parse_task_script — beambot.core.task_script

`_finite_number()`

Reject booleans, non-real values, infinities, NaN and, optionally, zero or negatives before float conversion.

Internal helper · Source

_finite_number(value: Any, path: str, *, positive: bool=False) -> float

Parameter	Meaning
`value`	Value to validate.
`path`	Human-readable field path used in the error.
`positive`	Require greater than zero when true.

Returns: Validated value converted to float.

Errors: Raises ValueError naming the field when validation fails.

Referenced in source (1)

expand_grid_target — beambot.core.task_script

`_positive_int()`

Require a Python integer greater than zero, explicitly excluding booleans.

Internal helper · Source

_positive_int(value: Any, path: str) -> int

Parameter	Meaning
`value`	Value to validate.
`path`	Human-readable field path used in the error.

Returns: The original positive integer.

Errors: Raises ValueError for zero, negatives, booleans and non-integers.

Referenced in source (1)

expand_grid_target — beambot.core.task_script

`_grid_direction()`

Validate the six word-based grid directions used by pickup macro moves.

Internal helper · Source

_grid_direction(value: Any, path: str) -> str

Parameter	Meaning
`value`	Expected forward, backward, left, right, up or down.
`path`	Human-readable field path used in the error.

Returns: The accepted direction string.

Errors: Raises ValueError listing allowed directions for invalid values.

Referenced in source (1)

expand_grid_target — beambot.core.task_script

`_expand_task_macros()`

Expand recognized pickup macros in order and shallow-copy all other task dictionaries.

Internal helper · Source

_expand_task_macros(
    tasks: Sequence[Mapping[str, Any]],
    vision_targets: Mapping[str, Any],
) -> list[dict[str, Any]]

Parameter	Meaning
`tasks`	Validated task mappings with task_type fields.
`vision_targets`	Rack configurations forwarded to pickup expansion.

Returns: Flat list of tasks with macros replaced by their executable steps.

Errors: Macro validation errors propagate; missing task_type raises KeyError.

Side effects: None; nested values in copied non-macro tasks remain shared.

Referenced in source (1)

parse_task_script — beambot.core.task_script

Task parsing and validation

How it is used

Start here

Example

Functions

parse_task_script()

expand_grid_target()

moveto_goal_error()

flange_offset_error()

Internal callbacks and helpers

_load_poses_registry()

_finite_number()

_positive_int()

_grid_direction()

_expand_task_macros()

`parse_task_script()`

`expand_grid_target()`

`moveto_goal_error()`

`flange_offset_error()`

`_load_poses_registry()`

`_finite_number()`

`_positive_int()`

`_grid_direction()`

`_expand_task_macros()`