Skip to content

Batch Processing

batch

Batch file loading and numpy export utilities for BVH datasets.

HarmonizeReport(kept_indices: list[int] = list(), kept_sources: list[str | None] = list(), dropped_indices: list[int] = list(), dropped_sources: list[str | None] = list(), drop_reasons: list[str] = list(), applied_stages: list[dict[str, str]] = list()) dataclass

Per-call summary of what :func:harmonize did to each clip.

All fields use JSON-native types so the report can be serialized directly with json.dumps(dataclasses.asdict(report)) and embedded as audit metadata alongside a preprocessed dataset.

Attributes:

Name Type Description
kept_indices list of int

Indices (into the input clips list) of clips that survived the topology gate.

kept_sources list of str or None

source_path of each kept clip, aligned with kept_indices.

dropped_indices list of int

Indices of clips that were dropped by the topology gate.

dropped_sources list of str or None

source_path of each dropped clip, aligned with dropped_indices.

drop_reasons list of str

One human-readable reason per dropped clip, aligned with dropped_indices.

applied_stages list of dict

One dict per kept clip, aligned with kept_indices. Each dict records which harmonization stages ran for that clip, with before→after where meaningful. Possible keys: "retarget", "resample", "world_up", "rest_up", "rest_forward", "euler_order". Empty dict means the clip passed the gate without needing any transformation.

read_bvh_directory(dirpath: str | Path, pattern: str = '*.bvh', sort: bool = True, parallel: bool = False, max_workers: int | None = None, world_up: str = 'auto', lr_mapping: dict[str, str] | None = None, skip_errors: bool = False) -> list[Bvh]

Load all BVH files from a directory.

Parameters:

Name Type Description Default
dirpath str or Path

Directory to search for BVH files.

required
pattern str

Glob pattern to filter files (default "*.bvh").

'*.bvh'
sort bool

If True (default), sort files alphabetically for deterministic ordering.

True
parallel bool

If True, load files in parallel using threads.

False
max_workers int or None

Maximum number of threads when parallel=True. None defers to the ThreadPoolExecutor default.

None
world_up str

World vertical axis applied to every loaded file. "auto" (default) auto-detects per file. Pass e.g. "+y" to override all files uniformly.

'auto'
lr_mapping dict or None

Explicit left/right joint pair mapping applied to every loaded file. Useful when a whole dataset shares an unusual naming convention the auto-detect heuristic can't parse.

None
skip_errors bool

If True, files that fail to load emit a UserWarning and are skipped. If False (default), the first failure propagates as the original exception. Use True when robustness against occasional corrupt files matters more than strict verification.

False

Returns:

Type Description
list of Bvh

One Bvh object per successfully loaded file. Shorter than the set of matched files when skip_errors=True and some failed.

Raises:

Type Description
FileNotFoundError

If dirpath does not exist.

harmonize(clips: list[Bvh], *, reference: Bvh | None = None, target_fps: float | None = None, target_world_up: str | None = None, target_rest_up: str | None = None, target_rest_forward: str | None = None, target_euler_order: str | None = None, on_incompatible: Literal['drop', 'raise'] = 'drop', verbose: bool = True, return_report: bool = False) -> list[Bvh] | tuple[list[Bvh], HarmonizeReport]

harmonize(clips: list[Bvh], *, reference: Bvh | None = ..., target_fps: float | None = ..., target_world_up: str | None = ..., target_rest_up: str | None = ..., target_rest_forward: str | None = ..., target_euler_order: str | None = ..., on_incompatible: Literal['drop', 'raise'] = ..., verbose: bool = ..., return_report: Literal[False] = ...) -> list[Bvh]
harmonize(clips: list[Bvh], *, reference: Bvh | None = ..., target_fps: float | None = ..., target_world_up: str | None = ..., target_rest_up: str | None = ..., target_rest_forward: str | None = ..., target_euler_order: str | None = ..., on_incompatible: Literal['drop', 'raise'] = ..., verbose: bool = ..., return_report: Literal[True]) -> tuple[list[Bvh], HarmonizeReport]

Apply dataset-level harmonization to a list of clips.

For each clip, applies in order:

  1. Topology check vs reference (if provided). On mismatch, the clip is dropped or raises per on_incompatible.
  2. Bone-proportion retargeting to reference (if provided).
  3. Frame-rate resampling to target_fps (if provided and the clip's current fps differs by more than 0.01).
  4. World-up reorientation to target_world_up (if provided and bvh.world_up != target_world_up). Rotates the entire scene — affects offsets, root_pos, and the world_up flag.
  5. Rest-up reorientation to target_rest_up (if provided and bvh.rest_up != target_rest_up). Rotates only the rest-pose offsets and compensates joint rotations so FK positions are unchanged; the world frame is untouched.
  6. Rest-forward reorientation to target_rest_forward (if provided and bvh.rest_forward != target_rest_forward). Rotates rest-pose offsets around the vertical axis so the skeleton's rest-pose facing matches.
  7. Euler-order re-expression to target_euler_order (if provided and any per-joint Euler order differs). Re-expresses each joint's stored Euler angles in the target order while preserving the underlying rotations.

The ordering matters: world-up is "heaviest" (touches everything), rest-up modifies only rest-pose offsets (leaving world frame intact), and rest-forward is a further rotation of those same offsets around the vertical. Euler-order re-expression runs last because it only rewrites channel layout, not geometry.

Any of the reference / target_* kwargs may be None to skip that stage. Passing all as None returns a shallow copy of clips (no-op).

Parameters:

Name Type Description Default
clips list of Bvh

Input clips.

required
reference Bvh or None

If provided, every clip must match reference.matches_hierarchy (same joints, same parent structure — rest offsets are allowed to differ since retargeting will overwrite them next). Kept clips are then retargeted to reference's bone offsets.

None
target_fps float or None

Target frame rate in Hz. Clips whose fps differs by more than 0.01 are resampled via quaternion SLERP.

None
target_world_up str or None

Signed-axis string ('+y', '-z', ...). Clips whose world_up differs are rotated via reorient_world_up.

None
target_rest_up str or None

Signed-axis string. Clips whose rest_up differs are corrected via reorient_rest_up. Typically used to fix files whose rest pose and animation disagree on the up axis.

None
target_rest_forward str or None

Signed-axis string. Clips whose rest_forward differs are rotated via reorient_rest_forward so the rest pose faces a consistent direction across the dataset.

None
target_euler_order str or None

Three-character order like 'XYZ' / 'ZYX'. When set, clips with any joint whose Euler order differs are re-expressed in the target order via :meth:Bvh.change_euler_order. This is orientation-preserving: underlying rotations are unchanged; only the channel layout is rewritten. Numerical drift can occur on gimbal-lock-adjacent rotations — for bit-exact round-trips across the conversion, prefer rotation-invariant representations ('6d' / 'quaternion') downstream.

None
on_incompatible ('drop', 'raise')

Behavior on topology mismatch with reference. "drop" (default) skips the clip; "raise" raises ValueError at the first mismatch.

"drop"
verbose bool

If True (default), emit a single UserWarning at end of call when one or more clips were dropped, summarizing how many were dropped and identifying the first few. Set to False to silence the summary entirely.

True
return_report bool

If True, return (clips, report) where report is a :class:HarmonizeReport describing every stage applied to every kept clip plus per-clip drop reasons. Default False keeps the return type as a plain list[Bvh].

False

Returns:

Type Description
list of Bvh, or (list of Bvh, HarmonizeReport)

Harmonized clips (and optional report). The list may be shorter than clips if any were dropped.

Raises:

Type Description
ValueError

If on_incompatible is not one of the accepted values, or if on_incompatible='raise' and a clip mismatches reference.

batch_to_numpy(bvh_list: list[Bvh], representation: str = 'euler', include_root_pos: bool = True, pad: bool = False, pad_value: float = 0.0) -> npt.NDArray[np.float64] | list[npt.NDArray[np.float64]]

Convert a list of Bvh objects to NumPy arrays.

All Bvh objects must share the same skeleton hierarchy. For representations whose channel layout depends on the source Euler order ('euler', 'axisangle'), all clips must additionally share the same per-joint Euler orders. For rotation-invariant representations ('6d', 'quaternion', 'rotmat') the Euler-order check is skipped.

Parameters:

Name Type Description Default
bvh_list list of Bvh

BVH objects to convert.

required
representation str

Rotation representation: 'euler' (default), '6d', 'quaternion', 'axisangle', or 'rotmat'.

'euler'
include_root_pos bool

If True (default), prepend root position (3 columns) to the rotation data.

True
pad bool

If True, zero-pad shorter sequences to the maximum length and return a single 3-D array (B, F_max, D). If False (default), return a list of 2-D arrays.

False
pad_value float

Value to use for padding (default 0.0).

0.0

Returns:

Type Description
ndarray or list of ndarray

If pad=True: array of shape (B, F_max, D). If pad=False: list of arrays, each (F_i, D).

Raises:

Type Description
ValueError

If skeletons are incompatible or representation is unknown.

compute_normalization_stats(bvh_list: list[Bvh], representation: str = 'euler', include_root_pos: bool = True) -> dict[str, np.ndarray]

Compute per-channel mean and std across a dataset of BVH objects.

Concatenates all frames from all clips, then computes mean and standard deviation per feature channel. Compatible with the Mean.npy / Std.npy convention used by HumanML3D and MDM.

Parameters:

Name Type Description Default
bvh_list list of Bvh

Dataset of BVH objects (must share the same skeleton topology).

required
representation str

Rotation representation: 'euler' (default), '6d', 'quaternion', 'axisangle', or 'rotmat'.

'euler'
include_root_pos bool

If True (default), include root position in the features.

True

Returns:

Type Description
dict

{"mean": ndarray (D,), "std": ndarray (D,), "constant_channels": ndarray of bool (D,)}.

constant_channels[i] is True when the raw standard deviation for channel i was below 1e-8 and the guard replaced it with 1.0. Normalized values on these channels are identically zero rather than ~N(0, 1) — use this mask to exclude them from per-channel diagnostics.

Notes

Save/load stats with np.savez("stats.npz", **stats) and dict(np.load("stats.npz")). Bool arrays round-trip cleanly through .npz.

normalize_array(data: npt.NDArray[np.float64], stats: dict[str, npt.NDArray[np.float64]]) -> npt.NDArray[np.float64]

Apply z-score normalization: (data - mean) / std.

Parameters:

Name Type Description Default
data ndarray

Data to normalize. Last dimension must match stats["mean"].

required
stats dict

{"mean": ndarray (D,), "std": ndarray (D,)} from :func:compute_normalization_stats.

required

Returns:

Type Description
ndarray

Normalized data, same shape as input.

denormalize_array(data: npt.NDArray[np.float64], stats: dict[str, npt.NDArray[np.float64]]) -> npt.NDArray[np.float64]

Reverse z-score normalization: data * std + mean.

Parameters:

Name Type Description Default
data ndarray

Normalized data to denormalize.

required
stats dict

{"mean": ndarray (D,), "std": ndarray (D,)} from :func:compute_normalization_stats.

required

Returns:

Type Description
ndarray

Denormalized data, same shape as input.