Mapping capture sequence

Sequence individual orchestrator goals and optional Zivid captures for a separately running Octomap pipeline.

src.beambot.scripts.capture_octomap · Source

How it is used

Use plan_capture_steps() to inspect how a task document becomes individual orchestrator goals and which named poses request a camera capture. This function only builds JSON and capture flags. The CLI main() calls run() for actual motion and capture, or runs offline assertions with --self-check.

Normal execution needs robot/camera bringup and the Octomap pipeline already running. Startup and movement failures stop the run. During the sequence, unsuccessful capture responses and map-save failures are best effort, so completion can leave an incomplete map.

Start here

Callable	Purpose
plan_capture_steps	Build one goal JSON string per task and mark matching named MoveTo steps for a later capture. Preserve `start_gripper`, `poses`, and `run_name`; other top-level metadata is omitted. No goal is sent.
main	Parse capture-run options and route –self-check to pure assertions or an ordinary run to ROS execution.
run	Read a run, send each planned single-task goal, capture after matching poses and wait after each capture attempt. Stop on movement failure, continue after capture failure, then optionally request map saving.

Example

This function-level excerpt assumes plan_capture_steps is available from capture_octomap.py. It shows the returned data without submitting the planned goals:

import json

run = {
    "start_gripper": "hande",
    "tasks": [{"task_type": "moveto", "target": "oscan1"}],
}
steps = plan_capture_steps(run, capture_prefix="oscan")
goal_json, capture_after = steps[0]
print(json.loads(goal_json)["tasks"])  # [{"task_type": "moveto", "target": "oscan1"}]
print(capture_after)                 # True

Functions

`plan_capture_steps()`

Build one goal JSON string per task and mark matching named MoveTo steps for a later capture. Preserve start_gripper, poses, and run_name; other top-level metadata is omitted. No goal is sent.

Function · Source

plan_capture_steps(run: dict, capture_prefix: str)

Parameter	Meaning
`run`	Run mapping with optional tasks, poses, start_gripper and run_name; missing tasks yields no steps.
`capture_prefix`	Case-sensitive target prefix; empty prefix matches every moveto target string.

Returns: Ordered list of (goal_json_string, do_capture_boolean) pairs. Each goal contains exactly one task and defaults run_name to capture_run.

Errors: Malformed task mappings/targets or nonserializable values propagate; no task-schema validation.

Side effects: JSON construction only; leaves input mappings unchanged.

Example

steps = plan_capture_steps({"start_gripper": "hande", "tasks": [{"task_type": "moveto", "target": "oscan1"}]}, "oscan")
assert len(steps) == 1 and steps[0][1]
assert json.loads(steps[0][0])["run_name"] == "capture_run"

Referenced in source (2)

_self_check — src.beambot.scripts.capture_octomap
run — src.beambot.scripts.capture_octomap

`main()`

Parse capture-run options and route –self-check to pure assertions or an ordinary run to ROS execution.

Entry point · Source

main() -> int

Returns: 0 for passing self-check; otherwise run() status.

Errors: Argparse errors for absent run_json; timeout/settle signs are not validated. Downstream input/runtime errors propagate.

Side effects: Self-check prints only; ordinary run reads JSON and can move hardware, acquire clouds and save a map.

`run()`

Read a run, send each planned single-task goal, capture after matching poses and wait after each capture attempt. Stop on movement failure, continue after capture failure, then optionally request map saving.

Function · Source

run(args) -> int

Parameter	Meaning
`args`	Namespace with run_json, capture_prefix, capture_service, capture_timeout (seconds), settle (ROS-clock seconds), no_save and save.

Returns: 0 when all movement goals succeeded even if captures/save failed; 1 for empty tasks, missing endpoints or a movement failure.

Errors: File/JSON/ROS errors propagate. After successful ROS initialization, finally destroys a constructed node and calls rclpy.try_shutdown(), including endpoint early returns and failed runs.

Side effects: Reads run JSON, initializes ROS, commands robot movement and camera capture, spins during integration waits and cleans up ROS; optionally writes an octomap via subprocess after normal movement completion.

Referenced in source (1)

main — src.beambot.scripts.capture_octomap

Internal callbacks and helpers

`_send_goal()`

Send one MTCExecution goal and spin until acceptance and final action status.

Internal helper · Source

_send_goal(node, action_client, full_json: str)

Parameter	Meaning
`node`	ROS node used by spin_until_future_complete.
`action_client`	MTCExecution action client for the orchestrator.
`full_json`	Full task JSON string copied into the goal.

Returns: (ok, message), with ok true only for STATUS_SUCCEEDED and message from result.error_message; rejected/None handle returns (False, “goal rejected”).

Errors: No acceptance/result timeout or exception catch; absent/malformed final result can raise.

Side effects: Submits robot tasks and spins ROS; no cancellation on timeout is implemented.

Referenced in source (1)

run — src.beambot.scripts.capture_octomap

`_capture()`

Call a Trigger capture service and spin until its result or the requested timeout.

Internal helper · Source

_capture(node, trigger_client, timeout: float)

Parameter	Meaning
`node`	ROS node used to process the service future.
`trigger_client`	std_srvs/Trigger client for the camera capture service.
`timeout`	Maximum service-result wait in seconds.

Returns: (response.success, response.message), or (False, timeout_message) if future remains pending.

Errors: Future/None-response exceptions propagate. Does not verify reception of an image/cloud or cancel a timed-out service request.

Side effects: Triggers acquisition and processes ROS callbacks.

Referenced in source (1)

run — src.beambot.scripts.capture_octomap

`_save_map()`

Run octomap_saver_node with octomap_path set to the destination and report success or a manual retry command.

Internal helper · Source

_save_map(path: str)

Parameter	Meaning
`path`	Destination octomap path passed as a ROS parameter, normally a .bt file.

Returns: None regardless of save success.

Errors: All Exceptions including nonzero exit and the 30-second timeout are printed to stderr and swallowed.

Side effects: Spawns ROS saver process that reads the live map and writes the destination.

Referenced in source (1)

run — src.beambot.scripts.capture_octomap

`_sleep()`

Advance ROS callback processing until the node clock reaches the requested interval or ROS shuts down.

Internal helper · Source

_sleep(node, seconds: float)

Parameter	Meaning
`node`	ROS node whose clock defines elapsed time.
`seconds`	Duration in node-clock seconds; nonpositive values return immediately. Paused simulated time can keep the loop waiting.

Returns: None.

Errors: Clock/spin errors propagate.

Side effects: Repeatedly calls spin_once with 0.05-second wall timeout.

Referenced in source (1)

run — src.beambot.scripts.capture_octomap

`_self_check()`

Assert one-task splitting, prefix selection and preservation of gripper/poses/run_name, including the empty-run case.

Internal helper · Source

_self_check()

Returns: None if all assertions pass.

Errors: AssertionError on planner regressions.

Side effects: Prints self-check OK; uses only Python standard library and performs no ROS/hardware work.

Referenced in source (1)

main — src.beambot.scripts.capture_octomap

Mapping capture sequence

How it is used

Start here

Example

Functions

plan_capture_steps()

main()

run()

Internal callbacks and helpers

_send_goal()

_capture()

_save_map()

_sleep()

_self_check()

`plan_capture_steps()`

`main()`

`run()`

`_send_goal()`

`_capture()`

`_save_map()`

`_sleep()`

`_self_check()`