Custom Processing

rddac preprocess is the reference implementation of the processed layout, not the only allowed one. The raw dataset is immutable and readable through the public API, so anyone with a better algorithm can produce their own processed layer: the contract is the schema of the output files, not a Python interface.

Levels of adjustment

Level	When	How
1. Change parameters	the algorithm is fine, a threshold or window is not	--config my.toml, no code
2a. Replace one step	you want a different filter for one modality, the reference stages for the rest	write that modality's group yourself, let rddac preprocess do the others
2b. Add a step	the reference output is right, you need something derived on top	append new datasets to the processed files
3. Replace everything	a different pipeline altogether	produce files matching the schema in your own directory

Level 1: parameters

Every threshold of every stage is a TOML key (rddac preprocess --dump-config prints them all with their defaults). A stricter oil-film cleaning, for example:

# my.toml
[oil]
hampel_window = 10   # wider neighbourhood (mm)
hampel_k = 2.5       # flag outliers earlier

rddac preprocess oil --config my.toml --overwrite

The values used are stamped into oil_thickness attributes of every output file (hampel_window, hampel_k, ...), and the docs figure can be redrawn for the variant with the same overrides: python -m rddac._preprocess.visualize oil --id 0 --config my.toml --out ..

Level 2a: replace one step, keep the rest

Say you prefer a plain running-median filter to the Hampel filter for the oil film. Write the oil_thickness group yourself and leave the other modalities to the reference stages. The output directory is shared: rddac preprocess appends to existing files and never touches groups it did not create unless --overwrite is given.

from pathlib import Path

import h5py
import numpy as np
import pandas as pd
from scipy.signal import medfilt

import rddac
from rddac._preprocess.h5_access import available_ids

DATA_DIR, OUT_DIR = Path("./data"), Path("./data/processed")
OUT_DIR.mkdir(exist_ok=True)

def median_oil(raw: np.ndarray, kernel: int = 7) -> np.ndarray:
    """Running median instead of Hampel: truncate, drop NaN, filter, average duplicates, fill the grid."""
    df = pd.DataFrame(raw, columns=["pos", "val"]).dropna()
    df["pos"] = df["pos"].round().astype(int)
    df = df[df["pos"] < 200].sort_values("pos")
    df["val"] = medfilt(df["val"].to_numpy(), kernel)
    grid = pd.DataFrame({"pos": range(200)}).merge(df.groupby("pos", as_index=False)["val"].mean(), how="left")
    grid["val"] = grid["val"].interpolate().bfill().ffill()
    return grid.to_numpy(dtype=np.float32)

for exp_id in sorted(available_ids(DATA_DIR) or []):
    with rddac.open_h5(exp_id, data_dir=DATA_DIR) as raw:
        cleaned = median_oil(raw["oil_thickness/data"][:])
    with h5py.File(OUT_DIR / f"{exp_id:04d}.h5", "a") as out:
        if "oil_thickness" in out:
            del out["oil_thickness"]
        g = out.create_group("oil_thickness")
        g.create_dataset("data", data=cleaned)
        g.attrs["columns"], g.attrs["units"] = ["sensor_position", "oil_value"], ["mm", "g/m^2"]
        g.attrs["producer"], g.attrs["median_kernel"] = "median_oil v1", 7     # record what produced this

rddac preprocess force sheet        # the reference stages fill in the other groups

The result is one processed layer: your oil profiles next to the reference force and sheet tables, streamable through the same views (iter_view("oil-thickness", data_dir="./data/processed", source="./data/metadata.json")). To compare against the reference, run rddac preprocess oil --out ./data/reference and plot both.

Level 2b: add a step on top of a reference stage

A step that extends the reference output: let rddac preprocess pointcloud do the full reference processing, then derive from it.

Example: a fixed-size random subsample (4,096 points) of every processed cloud, the typical input size for point-cloud networks:

from pathlib import Path

import h5py
import numpy as np

OUT_DIR = Path("./data/processed")
N, SEED = 4_096, 42

for path in sorted(OUT_DIR.glob("*.h5")):
    with h5py.File(path, "a") as f:
        if "pointcloud" not in f:
            continue
        for op_i, op in enumerate(("op10", "op20")):
            pts = f[f"pointcloud/{op}/z"][:]                       # (M, 3) processed cloud
            # seed per (file, op): rerunning any single file reproduces its sample,
            # independent of processing order or which other files are present
            rng = np.random.default_rng([SEED, int(path.stem), op_i])
            pick = rng.choice(len(pts), size=N, replace=False)
            name = f"pointcloud/{op}/z_{N}"
            if name in f:
                del f[name]
            d = f.create_dataset(name, data=pts[np.sort(pick)])
            d.attrs["producer"] = "random_subsample v1"             # record what produced this
            d.attrs["subsample_n"], d.attrs["subsample_seed"] = N, SEED

Two points: add a new dataset (z_4096) instead of replacing z, since downstream tooling expects the full cloud, and record the parameters (n, seed, producer) in the attributes. The same pattern applies to any derived quantity: normals, voxel grids, per-point features.

Level 3: your own pipeline

Read the raw files through the public API, run your own processing, and write the results into a separate directory in the same file schema. There is no Python interface to implement; output files only need to follow the processed schema:

import h5py
import rddac

for exp_id in range(9000):
    with rddac.open_h5(exp_id, data_dir="./data") as raw:      # immutable input
        oil_raw = raw["oil_thickness/data"][:]

    cleaned = my_better_oil(oil_raw)                            # your algorithm -> (200, 2) float32

    with h5py.File(f"./my_processed/{exp_id:04d}.h5", "a") as out:
        group = out.create_group("oil_thickness")
        group.create_dataset("data", data=cleaned)
        group.attrs["columns"] = ["sensor_position", "oil_value"]
        group.attrs["producer"] = "my_better_oil v1"            # record what produced this

Files in this schema work with the same views and downstream tooling as the reference output. The raw data is fixed and checksummed through the published Croissant manifest, so a processed layer is reproducible from its code alone; there is no server-side component involved.

Two conventions:

• Do not write into the raw directory. The published checksums define the dataset.
• Record the parameters and code version in the group attributes (the reference stages store every parameter they used), so a processed file can be interpreted without the code that produced it.

Internal reference

Internal API, may change without notice

The implementation lives in the private rddac._preprocess package. It is CLI-only by design; the signatures below are shown for orientation (e.g. for a notebook walking through the pipeline), not as a stable interface.

process(data: np.ndarray, *, max_sensor_position: int = OIL_MAX_SENSOR_POSITION, output_length: int = OIL_OUTPUT_LENGTH, hampel_window: int = OIL_HAMPEL_WINDOW, hampel_k: float = OIL_HAMPEL_K, value_quantization: float = OIL_VALUE_QUANTIZATION) -> tuple[np.ndarray, dict]

Process one experiment's raw oil table.

Parameters:

Name	Type	Description	Default
data	ndarray	(n, 2) raw array of [sensor_position, oil_value].	required
max_sensor_position	int	Positions at/after this (mm) are edge artifacts.	OIL_MAX_SENSOR_POSITION
output_length	int	Fixed output row count (one row per integer mm).	OIL_OUTPUT_LENGTH
hampel_window	int	Hampel half-window in mm.	OIL_HAMPEL_WINDOW
hampel_k	float	Hampel threshold in robust sigmas.	OIL_HAMPEL_K
value_quantization	float	Logging LSB; floors the flat-window fallback scale.	OIL_VALUE_QUANTIZATION

Returns:

Type	Description
ndarray	(processed, attrs) — processed is (output_length, 2) float32
dict	on the integer mm grid; attrs are the cleaning statistics plus the
tuple[ndarray, dict]	parameter values used (self-describing output).

Source code in rddac/_preprocess/oil.py

def process(
    data: np.ndarray,
    *,
    max_sensor_position: int = OIL_MAX_SENSOR_POSITION,
    output_length: int = OIL_OUTPUT_LENGTH,
    hampel_window: int = OIL_HAMPEL_WINDOW,
    hampel_k: float = OIL_HAMPEL_K,
    value_quantization: float = OIL_VALUE_QUANTIZATION,
) -> tuple[np.ndarray, dict]:
    """Process one experiment's raw oil table.

    Args:
        data: ``(n, 2)`` raw array of ``[sensor_position, oil_value]``.
        max_sensor_position: Positions at/after this (mm) are edge artifacts.
        output_length: Fixed output row count (one row per integer mm).
        hampel_window: Hampel half-window in mm.
        hampel_k: Hampel threshold in robust sigmas.
        value_quantization: Logging LSB; floors the flat-window fallback scale.

    Returns:
        ``(processed, attrs)`` — ``processed`` is ``(output_length, 2) float32``
        on the integer mm grid; ``attrs`` are the cleaning statistics plus the
        parameter values used (self-describing output).
    """
    df = pd.DataFrame(np.asarray(data, dtype=float), columns=list(COLUMNS))
    df = df.dropna(subset=["sensor_position"])
    df["pos_int"] = df["sensor_position"].round(0).astype(int)
    df = df[df["pos_int"] < max_sensor_position]
    n_in_range = len(df)

    n_nan = int(df["oil_value"].isna().sum())
    df = df.dropna(subset=["oil_value"]).sort_values("pos_int").reset_index(drop=True)

    outlier = hampel_filter(
        df["pos_int"].to_numpy(),
        df["oil_value"].to_numpy(),
        window_size=hampel_window,
        k=hampel_k,
        quantization=value_quantization,
    )
    n_outliers = int(outlier.sum())
    df = df[~outlier]

    df = df.groupby("pos_int", as_index=False)["oil_value"].mean()
    template = pd.DataFrame({"pos_int": range(max_sensor_position)})
    df = template.merge(df, on="pos_int", how="left")
    n_interpolated = int(df["oil_value"].isna().sum())
    df["oil_value"] = df["oil_value"].interpolate(method="linear").bfill().ffill()

    df = df.head(output_length)
    processed = df[["pos_int", "oil_value"]].to_numpy(dtype=np.float32)
    attrs = {
        "n_raw_in_range": n_in_range,
        "n_nan_removed": n_nan,
        "n_hampel_outliers": n_outliers,
        "n_positions_interpolated": n_interpolated,
        "max_sensor_position": max_sensor_position,
        "output_length": output_length,
        "hampel_window": hampel_window,
        "hampel_k": hampel_k,
        "value_quantization": value_quantization,
    }
    return processed, attrs

run(names: list[str], data_dir: str, out_dir: str, ids: str | None = None, split: str | None = None, workers: int = 1, overwrite: bool = False, quiet: bool = False, console=None, config: dict | None = None, skip_unavailable: bool = False, rebuild_models: bool = False) -> dict

Process the selected modalities for all selected experiments.

Parameters:

Name	Type	Description	Default
names	list[str]	Modalities to run, in order.	required
data_dir	str	Directory holding the raw dataset (zips or loose .h5).	required
out_dir	str	Output directory for processed files (created if missing).	required
ids	str | None	Optional id selection, e.g. '0-999' or '42,1035'.	None
split	str | None	Optional predefined split name to restrict to.	None
workers	int	Number of parallel worker processes.	1
overwrite	bool	Recompute modality groups that already exist.	False
quiet	bool	Suppress progress bar and summary output.	False
console		rich console to print to; a new one is created if None.	None
config	dict | None	Modality name → parameter overrides (see :mod:.config); missing entries use the defaults.	None
skip_unavailable	bool	Skip (instead of fail on) modalities whose optional dependencies are missing — used when the user did not name modalities explicitly.	False
rebuild_models	bool	Force retraining of cached models in preflight.	False

Returns:

Type	Description
dict	Stats dict {modality: {status: count}} plus "elapsed_s".

Source code in rddac/_preprocess/runner.py

def run(
    names: list[str],
    data_dir: str,
    out_dir: str,
    ids: str | None = None,
    split: str | None = None,
    workers: int = 1,
    overwrite: bool = False,
    quiet: bool = False,
    console=None,
    config: dict | None = None,
    skip_unavailable: bool = False,
    rebuild_models: bool = False,
) -> dict:
    """Process the selected modalities for all selected experiments.

    Args:
        names: Modalities to run, in order.
        data_dir: Directory holding the raw dataset (zips or loose ``.h5``).
        out_dir: Output directory for processed files (created if missing).
        ids: Optional id selection, e.g. ``'0-999'`` or ``'42,1035'``.
        split: Optional predefined split name to restrict to.
        workers: Number of parallel worker processes.
        overwrite: Recompute modality groups that already exist.
        quiet: Suppress progress bar and summary output.
        console: rich console to print to; a new one is created if None.
        config: Modality name → parameter overrides (see :mod:`.config`);
            missing entries use the defaults.
        skip_unavailable: Skip (instead of fail on) modalities whose optional
            dependencies are missing — used when the user did not name
            modalities explicitly.
        rebuild_models: Force retraining of cached models in preflight.

    Returns:
        Stats dict ``{modality: {status: count}}`` plus ``"elapsed_s"``.
    """
    if console is None:
        console = Console(quiet=quiet)
    config = config or {}

    selected: list[str] = []
    for name in names:
        try:
            module = modality_module(name)
        except RuntimeError as exc:
            if skip_unavailable:
                console.print(f"[yellow]{exc} — skipped[/yellow]")
                continue
            raise
        selected.append(name)
    names = selected
    if not names:
        return {"elapsed_s": 0.0}

    # One-time per-run hooks (main process, before forking workers). A stage
    # whose prerequisites are missing (e.g. pointcloud without simulations)
    # is skipped with a notice unless the user named it explicitly.
    ready: list[str] = []
    for name in names:
        module = modality_module(name)
        if hasattr(module, "preflight"):
            try:
                module.preflight(
                    data_dir, config.get(name, {}), console=console, rebuild_models=rebuild_models, out_dir=out_dir
                )
            except FileNotFoundError as exc:
                if not skip_unavailable:
                    raise
                console.print(f"[yellow]{name}: {exc} — skipped[/yellow]")
                continue
        ready.append(name)
    names = ready
    if not names:
        return {"elapsed_s": 0.0}

    experiment_ids = _experiment_ids(data_dir, ids, split)
    os.makedirs(out_dir, exist_ok=True)
    Path(out_dir, PROCESSED_MARKER).touch()

    jobs = [(i, data_dir, out_dir, names, overwrite, config) for i in experiment_ids]
    t0 = time.time()
    results = []

    # Chunking amortises the pool overhead for the millisecond-scale modalities;
    # the pointcloud stage takes ~100 s per experiment, where a chunk of 8 would
    # keep the progress bar at 0 for a quarter of an hour.
    chunksize = 1 if "pointcloud" in names else 8

    def _iter_results():
        if workers > 1:
            with Pool(workers) as pool:
                yield from pool.imap_unordered(_process_one, jobs, chunksize=chunksize)
        else:
            yield from map(_process_one, jobs)

    if quiet:
        results = list(_iter_results())
    else:
        with Progress(
            "[progress.description]{task.description}",
            BarColumn(),
            MofNCompleteColumn(),
            TimeRemainingColumn(),
            console=console,
        ) as progress:
            task = progress.add_task(f"preprocess {'+'.join(names)}", total=len(jobs))
            for result in _iter_results():
                results.append(result)
                progress.advance(task)

    stats: dict = {name: {} for name in names}
    for result in results:
        for name, state in result["status"].items():
            state = state.split(":")[0]  # collapse per-file error messages
            stats[name][state] = stats[name].get(state, 0) + 1
    stats["elapsed_s"] = round(time.time() - t0, 1)

    for name, counts in stats.items():
        if name == "elapsed_s":
            continue
        parts = [
            f"{count} {_STATUS_TEXT.get(state, state)}" for state, count in sorted(counts.items(), key=_status_order)
        ]
        console.print(f"[bold]{name}[/bold]: " + ", ".join(parts))
    console.print(f"[dim]finished in {stats['elapsed_s']} s[/dim]")
    return stats