Batch & evaluation

Batch execution and manifest types.

BatchRunner

Execute jobs with resume-aware manifest updates.

The worker must be pickleable when max_workers is greater than one.

Methods:

Name	Description
run	Run jobs, skip existing outputs, and update a manifest incrementally.

Functions

run

run(
    jobs: Iterable[BatchJob],
    worker: BatchWorker,
    *,
    manifest_path: str | Path,
    max_workers: int = 1,
    force: bool = False,
    input_dir: Path | None = None,
    output_dir: Path | None = None,
    pattern: str = "",
) -> BatchManifest

Run jobs, skip existing outputs, and update a manifest incrementally.

Loads an existing manifest at manifest_path when present so reruns can resume. Jobs whose output already exists are marked skipped unless force is true. After each completed (or skipped) job the manifest is rewritten so partial progress survives interruption.

Parameters:

Name	Type	Description	Default
jobs	Iterable[BatchJob]	Iterable of retargeting jobs to execute.	required
worker	BatchWorker	Pickleable callable invoked per pending job; must accept a :class:BatchJob and return optional provenance on success.	required
manifest_path	str | Path	JSON manifest path updated after each job finishes.	required
max_workers	int	Process pool size; 1 runs jobs sequentially in-process.	1
force	bool	When true, rerun jobs even if output already exists.	False
input_dir	Path | None	Optional batch input root stored on the manifest.	None
output_dir	Path | None	Optional batch output root stored on the manifest.	None
pattern	str	Optional glob or filter label stored on the manifest.	''

Returns:

Type	Description
BatchManifest	Final manifest aggregating all job records and summary counts.

Raises:

Type	Description
ValueError	If max_workers is less than 1.

BatchJob

Bases: BaseModel

One retargeting job in a batch run.

Attributes:

Name	Type	Description
id	str	Stable job identifier used in manifests and resume logic.
motion	Path	Input motion artifact path for the worker.
output	Path	Expected output artifact path; existing files may be skipped.
name	str | None	Optional display name; defaults to id in callers.
motion_format	MotionFormatKind | None	Typed source motion format override.
robot	RobotKind | None	Typed robot registry override.
config_path	Path | None	Optional declarative experiment config.
provenance	dict[str, Any]	Origin and scheduling history for the job.

Attributes

model_config class-attribute instance-attribute

model_config = ConfigDict(arbitrary_types_allowed=True)

id instance-attribute

id: str

motion instance-attribute

motion: Path

output instance-attribute

output: Path

name class-attribute instance-attribute

name: str | None = None

motion_format class-attribute instance-attribute

motion_format: MotionFormatKind | None = None

robot class-attribute instance-attribute

robot: RobotKind | None = None

config_path class-attribute instance-attribute

config_path: Path | None = None

provenance class-attribute instance-attribute

provenance: dict[str, Any] = Field(default_factory=dict)

BatchManifest

Bases: BaseModel

Manifest written by resumable batch runs.

Attributes:

Name	Type	Description
schema_version	int	Manifest format version for forward compatibility.
created_at	datetime	UTC time when the batch run first started.
updated_at	datetime	UTC time of the most recent manifest write.
input_dir	Path | None	Root directory scanned for input motions, if known.
output_dir	Path | None	Root directory for written outputs, if known.
pattern	str	Glob or filter pattern used to build the job list.
total	int	Number of jobs in the batch definition.
success_count	int	Jobs that completed with :attr:~RunStatus.SUCCESS.
skipped_count	int	Jobs skipped because outputs already existed.
failed_count	int	Jobs that raised during execution.
records	tuple[BatchRunRecord, ...]	Per-job outcomes in submission order.

Methods:

Name	Description
from_records	Build a manifest and derive summary counts from records.
load	Load a manifest JSON file.
save_json	Save manifest JSON.
records_by_id	Return records keyed by job id.

Attributes

model_config class-attribute instance-attribute

model_config = ConfigDict(arbitrary_types_allowed=True)

schema_version class-attribute instance-attribute

schema_version: int = 1

created_at instance-attribute

created_at: datetime

updated_at instance-attribute

updated_at: datetime

input_dir class-attribute instance-attribute

input_dir: Path | None = None

output_dir class-attribute instance-attribute

output_dir: Path | None = None

pattern class-attribute instance-attribute

pattern: str = ''

total class-attribute instance-attribute

total: int = 0

success_count class-attribute instance-attribute

success_count: int = 0

skipped_count class-attribute instance-attribute

skipped_count: int = 0

failed_count class-attribute instance-attribute

failed_count: int = 0

records class-attribute instance-attribute

records: tuple[BatchRunRecord, ...] = ()

Functions

from_records classmethod

from_records(
    *,
    records: Iterable[BatchRunRecord],
    total: int,
    created_at: datetime,
    input_dir: Path | None = None,
    output_dir: Path | None = None,
    pattern: str = "",
) -> BatchManifest

Build a manifest and derive summary counts from records.

load classmethod

load(path: str | Path) -> BatchManifest

Load a manifest JSON file.

save_json

save_json(path: str | Path) -> Path

Save manifest JSON.

records_by_id

records_by_id() -> dict[str, BatchRunRecord]

Return records keyed by job id.

BatchRunRecord

Bases: BaseModel

Recorded outcome for one batch job.

Attributes:

Name	Type	Description
job_id	str	Identifier matching :attr:BatchJob.id.
motion	Path	Input path from the job definition.
output	Path	Output path from the job definition.
status	RunStatus	Terminal status (success, skipped, or failed).
started_at	datetime	UTC timestamp when execution began.
finished_at	datetime	UTC timestamp when execution completed.
message	str	Human-readable summary or error text.
error_type	str | None	Exception class name when status is failed.
provenance	dict[str, Any]	Worker-returned processing history or resume information.

Attributes

model_config class-attribute instance-attribute

model_config = ConfigDict(arbitrary_types_allowed=True)

job_id instance-attribute

job_id: str

motion instance-attribute

motion: Path

output instance-attribute

output: Path

status instance-attribute

status: RunStatus

started_at instance-attribute

started_at: datetime

finished_at instance-attribute

finished_at: datetime

message class-attribute instance-attribute

message: str = ''

error_type class-attribute instance-attribute

error_type: str | None = None

provenance class-attribute instance-attribute

provenance: dict[str, Any] = Field(default_factory=dict)

duration_s property

duration_s: float

Wall-clock duration in seconds.