Preprocessing

The supported interface is the CLI, rddac preprocess; what each stage does is documented in the Preprocessing section. The Python functions below live in the private rddac._preprocess package and are listed for orientation (notebooks, custom variants), not as a stable API.

Internal API, may change without notice

Anyone replacing a stage should follow Custom processing: read raw data via the public API and match the processed schema, instead of importing from here.

Stages

process(data: np.ndarray, columns: Sequence[str] = FORCE_COLUMNS, *, time_window_start: float = FORCE_TIME_WINDOW[0], time_window_end: float = FORCE_TIME_WINDOW[1], position_precision_mm: float = FORCE_POSITION_PRECISION_MM) -> tuple[np.ndarray, dict]

Process one experiment's raw force table.

Parameters:

Name	Type	Description	Default
data	ndarray	(n, len(columns)) raw array as stored under force/data.	required
columns	Sequence[str]	Column layout of data (pass the h5 columns attr when it differs from the default raw layout).	FORCE_COLUMNS
time_window_start	float	Forming window start in s (exclusive).	FORCE_TIME_WINDOW[0]
time_window_end	float	Forming window end in s (inclusive).	FORCE_TIME_WINDOW[1]
position_precision_mm	float	Punch position quantization step.	FORCE_POSITION_PRECISION_MM

Returns:

Type	Description
ndarray	(processed, attrs) — processed is (m, len(columns)) float32
dict	with the input column order preserved; attrs hold row counts plus
tuple[ndarray, dict]	the parameter values used (self-describing output).

Source code in rddac/_preprocess/force.py

def process(
    data: np.ndarray,
    columns: Sequence[str] = FORCE_COLUMNS,
    *,
    time_window_start: float = FORCE_TIME_WINDOW[0],
    time_window_end: float = FORCE_TIME_WINDOW[1],
    position_precision_mm: float = FORCE_POSITION_PRECISION_MM,
) -> tuple[np.ndarray, dict]:
    """Process one experiment's raw force table.

    Args:
        data: ``(n, len(columns))`` raw array as stored under ``force/data``.
        columns: Column layout of ``data`` (pass the h5 ``columns`` attr when
            it differs from the default raw layout).
        time_window_start: Forming window start in s (exclusive).
        time_window_end: Forming window end in s (inclusive).
        position_precision_mm: Punch position quantization step.

    Returns:
        ``(processed, attrs)`` — ``processed`` is ``(m, len(columns)) float32``
        with the input column order preserved; ``attrs`` hold row counts plus
        the parameter values used (self-describing output).
    """
    df = pd.DataFrame(np.asarray(data, dtype=float), columns=list(columns))
    if "time" not in df.columns:
        raise ValueError(f"force table has no 'time' column: {list(columns)}")
    n_raw = len(df)

    df = df[(df["time"] > time_window_start) & (df["time"] <= time_window_end)].reset_index(drop=True)
    df["time"] = (df["time"] - df["time"].min()).round(FORCE_TIME_DECIMALS)

    for name in FORCE_SIGNALS:
        if name in df.columns:
            # Load cells rest at the blankholder preload, not at zero.
            df[name] -= np.abs(df[name].min())
            df[name] = df[name].round(FORCE_KN_DECIMALS)
    if "punch_temp" in df.columns:
        df["punch_temp"] = df["punch_temp"].round(FORCE_TEMP_DECIMALS)
    if "punch_pos" in df.columns:
        df["punch_pos"] = np.round(df["punch_pos"] / position_precision_mm) * position_precision_mm

    processed = df.to_numpy(dtype=np.float32)
    attrs = {
        "n_raw_rows": n_raw,
        "n_rows": len(df),
        "time_window_start": time_window_start,
        "time_window_end": time_window_end,
        "position_precision_mm": position_precision_mm,
    }
    return processed, attrs

process(data: np.ndarray, *, last_n: int = SHEET_LAST_N) -> tuple[np.ndarray, dict]

Process one experiment's raw sheet-thickness table.

Parameters:

Name	Type	Description	Default
data	ndarray	(n, 2) raw array of [sensor_position, sheet_thickness].	required
last_n	int	How many trailing traverse positions to keep.	SHEET_LAST_N

Returns:

Type	Description
ndarray	(processed, attrs) — processed is (<=last_n, 2) float32
dict	with NaN where the sensor reported error values; attrs hold the
tuple[ndarray, dict]	counts plus the parameter values used (self-describing output).

Source code in rddac/_preprocess/sheet.py

def process(data: np.ndarray, *, last_n: int = SHEET_LAST_N) -> tuple[np.ndarray, dict]:
    """Process one experiment's raw sheet-thickness table.

    Args:
        data: ``(n, 2)`` raw array of ``[sensor_position, sheet_thickness]``.
        last_n: How many trailing traverse positions to keep.

    Returns:
        ``(processed, attrs)`` — ``processed`` is ``(<=last_n, 2) float32``
        with NaN where the sensor reported error values; ``attrs`` hold the
        counts plus the parameter values used (self-describing output).
    """
    df = pd.DataFrame(np.asarray(data, dtype=float), columns=list(COLUMNS))
    n_raw = len(df)
    df = df.tail(last_n).reset_index(drop=True)

    df["sensor_position"] -= df["sensor_position"].min()
    n_negative = int((df["sheet_thickness"] < 0).sum())
    df.loc[df["sheet_thickness"] < 0, "sheet_thickness"] = np.nan

    df["sheet_thickness"] = (
        np.round(df["sheet_thickness"] / SHEET_THICKNESS_PRECISION_UM) * SHEET_THICKNESS_PRECISION_UM
    ).round(2)
    df["sensor_position"] = (
        np.round(df["sensor_position"] / SHEET_POSITION_PRECISION_MM) * SHEET_POSITION_PRECISION_MM
    ).round(2)

    processed = df.to_numpy(dtype=np.float32)
    attrs = {"n_raw_rows": n_raw, "n_rows": len(df), "n_negative_masked": n_negative, "last_n": last_n}
    return processed, attrs

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

hampel_filter(positions: np.ndarray, values: np.ndarray, window_size: int = OIL_HAMPEL_WINDOW, k: float = OIL_HAMPEL_K, quantization: float = OIL_VALUE_QUANTIZATION) -> np.ndarray

Outlier mask for a 1D traverse via a position-windowed Hampel filter.

Windows are selected by position distance (not index), so gaps from removed dropouts do not misalign them. In flat windows where the MAD is zero the scale falls back to the mean absolute deviation, floored at one logging quantization step.

Parameters:

Name	Type	Description	Default
positions	ndarray	Integer sensor positions, one per measurement.	required
values	ndarray	Measured oil values (must not contain NaN).	required
window_size	int	Half-window size in position units (mm).	OIL_HAMPEL_WINDOW
k	float	Threshold in robust sigmas.	OIL_HAMPEL_K

Returns:

Type	Description
ndarray	Boolean array; True marks an outlier.

Source code in rddac/_preprocess/oil.py

def hampel_filter(
    positions: np.ndarray,
    values: np.ndarray,
    window_size: int = OIL_HAMPEL_WINDOW,
    k: float = OIL_HAMPEL_K,
    quantization: float = OIL_VALUE_QUANTIZATION,
) -> np.ndarray:
    """Outlier mask for a 1D traverse via a position-windowed Hampel filter.

    Windows are selected by position distance (not index), so gaps from removed
    dropouts do not misalign them. In flat windows where the MAD is zero the
    scale falls back to the mean absolute deviation, floored at one logging
    quantization step.

    Args:
        positions: Integer sensor positions, one per measurement.
        values: Measured oil values (must not contain NaN).
        window_size: Half-window size in position units (mm).
        k: Threshold in robust sigmas.

    Returns:
        Boolean array; True marks an outlier.
    """
    outlier = np.zeros(len(values), dtype=bool)
    for i in range(len(values)):
        window = values[np.abs(positions - positions[i]) <= window_size]
        median = np.median(window)
        abs_dev = np.abs(window - median)
        scale = 1.4826 * np.median(abs_dev)
        if scale == 0:
            # 1.2533 = sqrt(pi/2) makes the mean abs dev a consistent sigma estimate.
            scale = max(1.2533 * float(np.mean(abs_dev)), quantization)
        if np.abs(values[i] - median) > k * scale:
            outlier[i] = True
    return outlier

The pointcloud stage owns its whole file section (rddac._preprocess.pointcloud.stage.process_experiment) and needs the [preprocessing] extra plus the DDACS simulations; see Point Clouds.

Orchestration and configuration

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

load(path: str) -> dict[str, dict]

Merge a user TOML file over the defaults.

Parameters:

Name	Type	Description	Default
path	str	Path to the TOML file.	required

Returns:

Type	Description
dict[str, dict]	The complete configuration (defaults with the file's overrides).

Raises:

Type	Description
ValueError	On unknown tables or keys — a silently ignored typo would mean silently NOT applying the user's intended override.
OSError	If the file cannot be read.

Source code in rddac/_preprocess/config.py

def load(path: str) -> dict[str, dict]:
    """Merge a user TOML file over the defaults.

    Args:
        path: Path to the TOML file.

    Returns:
        The complete configuration (defaults with the file's overrides).

    Raises:
        ValueError: On unknown tables or keys — a silently ignored typo would
            mean silently NOT applying the user's intended override.
        OSError: If the file cannot be read.
    """
    with open(path, "rb") as f:
        user = tomllib.load(f)

    cfg = defaults_config()
    for table, values in user.items():
        if table not in cfg:
            raise ValueError(f"unknown config table [{table}] — valid: {', '.join(cfg)}")
        if not isinstance(values, dict):
            raise ValueError(f"[{table}] must be a table of key = value pairs")
        for key, value in values.items():
            if key not in cfg[table]:
                raise ValueError(f"unknown key '{key}' in [{table}] — valid: {', '.join(cfg[table])}")
            cfg[table][key] = value
    return cfg

dump()

The complete default configuration as a ready-to-edit TOML string.

Source code in rddac/_preprocess/config.py

def dump() -> str:
    """The complete default configuration as a ready-to-edit TOML string."""
    lines = [
        "# rddac preprocess configuration: these are the defaults (the",
        "# reference recipe). Edit values and pass the file via",
        "#   rddac preprocess --config my.toml",
        "# Publish it next to your code to make the variant reproducible.",
        "",
    ]
    for table, values in DEFAULTS.items():
        lines.append(f"[{table}]")
        for key, value in values.items():
            if isinstance(value, bool):
                lines.append(f"{key} = {'true' if value else 'false'}")
            elif isinstance(value, str):
                lines.append(f'{key} = "{value}"')
            else:
                lines.append(f"{key} = {value}")
        lines.append("")
    return "\n".join(lines)

Processing figures

The figures on the preprocessing pages are drawn from raw arrays by these functions; python -m rddac._preprocess.visualize --help renders them to files.

plot_oil_processing(data, *, figsize=FIGSIZE, **params)

Three panels for ONE experiment's raw oil_thickness/data.

(a) raw readings with the truncation line, (b) truncated readings with the Hampel-flagged outliers, (c) the processed fixed-grid profile.

Parameters:

Name	Type	Description	Default
data	ndarray	(n, 2) raw array of [sensor_position, oil_value].	required
figsize	tuple[float, float]	Figure size in inches.	FIGSIZE
**params		Keyword overrides for :func:rddac._preprocess.oil.process.	{}

Source code in rddac/_preprocess/visualize.py

def plot_oil_processing(data: np.ndarray, *, figsize: tuple[float, float] = FIGSIZE, **params) -> Figure:
    """Three panels for ONE experiment's raw ``oil_thickness/data``.

    (a) raw readings with the truncation line, (b) truncated readings with the
    Hampel-flagged outliers, (c) the processed fixed-grid profile.

    Args:
        data: ``(n, 2)`` raw array of ``[sensor_position, oil_value]``.
        figsize: Figure size in inches.
        **params: Keyword overrides for :func:`rddac._preprocess.oil.process`.
    """
    processed, attrs = oil.process(data, **params)
    max_pos = attrs["max_sensor_position"]

    df = pd.DataFrame(np.asarray(data, dtype=float), columns=list(oil.COLUMNS)).dropna(subset=["sensor_position"])
    df["pos_int"] = df["sensor_position"].round(0).astype(int)
    # Same pre-filter path as oil.process: truncate, drop dropouts, sort.
    kept = df[df["pos_int"] < max_pos].dropna(subset=["oil_value"]).sort_values("pos_int")
    outlier = oil.hampel_filter(
        kept["pos_int"].to_numpy(),
        kept["oil_value"].to_numpy(),
        window_size=attrs["hampel_window"],
        k=attrs["hampel_k"],
        quantization=attrs["value_quantization"],
    )

    fig, axes = plt.subplots(1, 3, figsize=figsize, sharey=True)
    ax = axes[0]
    ax.scatter(df["pos_int"], df["oil_value"], c=RAW_COLOR, s=10, alpha=0.5)
    ax.axvline(max_pos, color=CUT_COLOR, linestyle="--", linewidth=1.5)
    ax.set_ylabel("Oil in g/m²")
    ax.set_title("(a) Raw")

    ax = axes[1]
    ax.scatter(kept["pos_int"][~outlier], kept["oil_value"][~outlier], c=RAW_COLOR, s=10, alpha=0.5)
    ax.scatter(kept["pos_int"][outlier], kept["oil_value"][outlier], c=CUT_COLOR, s=30, marker="x", linewidths=1.5)
    ax.set_title(f"(b) Hampel filter ({int(outlier.sum())} outliers)")

    ax = axes[2]
    ax.plot(processed[:, 0], processed[:, 1], color=PROCESSED_COLOR, linewidth=1.0)
    ax.scatter(processed[:, 0], processed[:, 1], c=PROCESSED_COLOR, s=8, alpha=0.7)
    ax.set_title("(c) Processed")

    for ax in axes:
        ax.set_xlabel("Position in mm")
        ax.grid(True, alpha=0.3)
    fig.tight_layout()
    return fig

plot_force_processing(tables, columns=FORCE_COLUMNS, *, figsize=FIGSIZE, **params)

Two panels with several experiments' force/data overlaid.

(a) raw total force over time with the forming window, (b) the processed curves: windowed, time re-zeroed, rest offset removed.

Parameters:

Name	Type	Description	Default
tables	Sequence[ndarray]	Raw (n, len(columns)) arrays, one per experiment.	required
columns	Sequence[str]	Column layout of the tables (the h5 columns attr).	FORCE_COLUMNS
figsize	tuple[float, float]	Figure size in inches.	FIGSIZE
**params		Keyword overrides for :func:rddac._preprocess.force.process.	{}

Source code in rddac/_preprocess/visualize.py

def plot_force_processing(
    tables: Sequence[np.ndarray],
    columns: Sequence[str] = FORCE_COLUMNS,
    *,
    figsize: tuple[float, float] = FIGSIZE,
    **params,
) -> Figure:
    """Two panels with several experiments' ``force/data`` overlaid.

    (a) raw total force over time with the forming window, (b) the processed
    curves: windowed, time re-zeroed, rest offset removed.

    Args:
        tables: Raw ``(n, len(columns))`` arrays, one per experiment.
        columns: Column layout of the tables (the h5 ``columns`` attr).
        figsize: Figure size in inches.
        **params: Keyword overrides for :func:`rddac._preprocess.force.process`.
    """
    columns = list(columns)
    i_t, i_f = columns.index("time"), columns.index("total_force")
    alpha = _overlay_alpha(len(tables))

    fig, axes = plt.subplots(1, 2, figsize=figsize, sharey=True)
    t_start = t_end = None
    t_max = 0.0
    for table in tables:
        table = np.asarray(table, dtype=float)
        processed, attrs = force.process(table, columns, **params)
        t_start, t_end = attrs["time_window_start"], attrs["time_window_end"]
        t_max = max(t_max, float(np.nanmax(table[:, i_t])))
        axes[0].plot(table[:, i_t], table[:, i_f], color=RAW_COLOR, alpha=alpha, linewidth=1.0)
        axes[1].plot(processed[:, i_t], processed[:, i_f], color=PROCESSED_COLOR, alpha=alpha, linewidth=1.0)

    ax = axes[0]
    if t_start is not None:
        ax.axvline(t_start, color=CUT_COLOR, linestyle="--", linewidth=1.5)
        ax.axvline(t_end, color=CUT_COLOR, linestyle="--", linewidth=1.5)
        ax.axvspan(0, t_start, alpha=0.15, color=REMOVED_COLOR)
        ax.axvspan(t_end, t_max, alpha=0.15, color=REMOVED_COLOR)
        axes[1].set_xlim(0, t_end - t_start)
    ax.set_xlim(0, t_max)
    ax.set_ylabel("Force in kN")
    ax.set_title("(a) Raw with forming window")
    axes[1].set_title("(b) Processed")

    for ax in axes:
        ax.set_xlabel("Time in s")
        ax.grid(True, alpha=0.3)
    fig.tight_layout()
    return fig

plot_sheet_processing(tables, *, figsize=FIGSIZE, **params)

Two panels with several experiments' sheet_thickness/data overlaid.

(a) raw traverses (error codes hidden) with the tail-selection cutoff, (b) the processed, position-normalized profiles.

Parameters:

Name	Type	Description	Default
tables	Sequence[ndarray]	Raw (n, 2) arrays of [sensor_position, sheet_thickness].	required
figsize	tuple[float, float]	Figure size in inches.	FIGSIZE
**params		Keyword overrides for :func:rddac._preprocess.sheet.process.	{}

Source code in rddac/_preprocess/visualize.py

def plot_sheet_processing(tables: Sequence[np.ndarray], *, figsize: tuple[float, float] = FIGSIZE, **params) -> Figure:
    """Two panels with several experiments' ``sheet_thickness/data`` overlaid.

    (a) raw traverses (error codes hidden) with the tail-selection cutoff,
    (b) the processed, position-normalized profiles.

    Args:
        tables: Raw ``(n, 2)`` arrays of ``[sensor_position, sheet_thickness]``.
        figsize: Figure size in inches.
        **params: Keyword overrides for :func:`rddac._preprocess.sheet.process`.
    """
    alpha = _overlay_alpha(len(tables))
    fig, axes = plt.subplots(1, 2, figsize=figsize, sharey=True)

    cutoffs: list[float] = []
    starts: list[float] = []
    for table in tables:
        table = np.asarray(table, dtype=float)
        processed, attrs = sheet.process(table, **params)
        pos, thick = table[:, 0], table[:, 1]
        valid = thick > 0  # negative readings are sensor error codes
        axes[0].plot(pos[valid], thick[valid], color=RAW_COLOR, alpha=alpha, linewidth=1.0)
        axes[1].plot(processed[:, 0], processed[:, 1], color=PROCESSED_COLOR, alpha=alpha, linewidth=1.0)
        if len(pos) > attrs["last_n"]:
            cutoffs.append(float(pos[len(pos) - attrs["last_n"]]))
            starts.append(float(pos[0]))

    ax = axes[0]
    if cutoffs:
        cutoff = float(np.median(cutoffs))
        ax.axvline(cutoff, color=CUT_COLOR, linestyle="--", linewidth=1.5)
        ax.axvspan(min(starts), cutoff, alpha=0.15, color=REMOVED_COLOR)
    ax.set_ylabel("Sheet thickness in µm")
    ax.set_title("(a) Raw with cutoff")
    axes[1].set_title("(b) Processed")

    for ax in axes:
        ax.set_xlabel("Position in mm")
        ax.grid(True, alpha=0.3)
    fig.tight_layout()
    return fig

plot_luminescence_processing(lumi_2d, *, figsize=(14.0, 3.2), **params)

Four panels for one scan's raw luminescence grid.

(a) raw grid, (b) connected foreground patches, (c) the validity mask after the patch-size filter, (d) the packed uint8 grayscale image.

Parameters:

Name	Type	Description	Default
lumi_2d	ndarray	(H, W) raw luminescence grid (reshaped luminescence).	required
figsize	tuple[float, float]	Figure size in inches.	(14.0, 3.2)
**params		[pointcloud] parameters; only lumi_min_patch_size is used.	{}

Source code in rddac/_preprocess/visualize.py

def plot_luminescence_processing(
    lumi_2d: np.ndarray, *, figsize: tuple[float, float] = (14.0, 3.2), **params
) -> Figure:
    """Four panels for one scan's raw luminescence grid.

    (a) raw grid, (b) connected foreground patches, (c) the validity mask
    after the patch-size filter, (d) the packed uint8 grayscale image.

    Args:
        lumi_2d: ``(H, W)`` raw luminescence grid (reshaped ``luminescence``).
        figsize: Figure size in inches.
        **params: ``[pointcloud]`` parameters; only ``lumi_min_patch_size`` is used.
    """
    from matplotlib.colors import ListedColormap
    from scipy import ndimage

    from .defaults import PC_LUMI_MIN_PATCH_SIZE
    from .pointcloud import geometry

    min_patch = params.get("lumi_min_patch_size", PC_LUMI_MIN_PATCH_SIZE)
    lumi_2d = np.asarray(lumi_2d)
    labeled, n_patches = ndimage.label(lumi_2d > 0)
    valid = geometry.lumi_valid_mask(lumi_2d, min_patch)
    packed = geometry.pack_luminescence(lumi_2d, valid)
    n_kept = len(np.unique(labeled[valid])) if valid.any() else 0

    raw_masked = np.where(lumi_2d > 0, lumi_2d.astype(float), np.nan)
    fig, axes = plt.subplots(1, 4, figsize=figsize)

    im = axes[0].imshow(raw_masked, cmap="viridis", aspect="auto")
    fig.colorbar(im, ax=axes[0], shrink=0.8)
    axes[0].set_title("(a) Raw")

    n_colors = min(n_patches + 1, len(_PATCH_COLORS))
    axes[1].imshow(
        np.clip(labeled, 0, n_colors - 1),
        cmap=ListedColormap(_PATCH_COLORS[:n_colors]),
        aspect="auto",
        vmin=0,
        vmax=n_colors - 1,
    )
    axes[1].set_title(f"(b) Connected patches ({n_patches})")

    axes[2].imshow(np.where(valid, raw_masked, np.nan), cmap="viridis", aspect="auto")
    axes[2].set_title(f"(c) Size filter ({n_kept} kept, {n_patches - n_kept} removed)")

    im = axes[3].imshow(packed, cmap="gray", aspect="auto", vmin=0, vmax=255)
    fig.colorbar(im, ax=axes[3], shrink=0.8)
    axes[3].set_title("(d) Packed uint8")

    for ax in axes:
        ax.set_xlabel("X in px")
    axes[0].set_ylabel("Y in px")
    fig.tight_layout()
    return fig

plot_pointcloud_processing(z_2d, lumi_2d, processed_points, *, sim_points=None, stats=None, max_points=300000, figsize=None, **params)

Top-view panels: raw calibrated scan, processed cloud, and optionally the matched simulation and the distance to it.

The raw panel applies only the validity mask and the calibration (steps 1–2 of the stage); the processed panel is the stage's output (cleaned, ICP-aligned to the matched simulation), so gaps between the two are the removed artifacts. With sim_points two more panels follow: the simulation in the same frame, and the processed points coloured by their nearest-neighbour distance to it (the kd sim-distance feature).

Parameters:

Name	Type	Description	Default
z_2d	ndarray	(H, W) raw z grid (sensor units).	required
lumi_2d	ndarray	(H, W) raw luminescence grid (for the validity mask).	required
processed_points	ndarray	(N, 3) processed pointcloud/<op>/z.	required
sim_points	ndarray | None	(M, 3) matched simulation points (mirrored full part), e.g. SimContext(data_dir).points(simulation_id, op).	None
stats	dict | None	The processed group's attrs; when given, removal counts are shown in the panel title.	None
max_points	int	Subsampling cap per panel (keeps the figure light).	300000
figsize	tuple[float, float] | None	Figure size in inches.	None
**params		[pointcloud] parameters; only lumi_min_patch_size is used.	{}

Source code in rddac/_preprocess/visualize.py

def plot_pointcloud_processing(
    z_2d: np.ndarray,
    lumi_2d: np.ndarray,
    processed_points: np.ndarray,
    *,
    sim_points: np.ndarray | None = None,
    stats: dict | None = None,
    max_points: int = 300_000,
    figsize: tuple[float, float] | None = None,
    **params,
) -> Figure:
    """Top-view panels: raw calibrated scan, processed cloud, and optionally
    the matched simulation and the distance to it.

    The raw panel applies only the validity mask and the calibration (steps
    1–2 of the stage); the processed panel is the stage's output (cleaned,
    ICP-aligned to the matched simulation), so gaps between the two are the
    removed artifacts. With ``sim_points`` two more panels follow: the
    simulation in the same frame, and the processed points coloured by their
    nearest-neighbour distance to it (the ``kd`` sim-distance feature).

    Args:
        z_2d: ``(H, W)`` raw z grid (sensor units).
        lumi_2d: ``(H, W)`` raw luminescence grid (for the validity mask).
        processed_points: ``(N, 3)`` processed ``pointcloud/<op>/z``.
        sim_points: ``(M, 3)`` matched simulation points (mirrored full part),
            e.g. ``SimContext(data_dir).points(simulation_id, op)``.
        stats: The processed group's attrs; when given, removal counts are
            shown in the panel title.
        max_points: Subsampling cap per panel (keeps the figure light).
        figsize: Figure size in inches.
        **params: ``[pointcloud]`` parameters; only ``lumi_min_patch_size`` is used.
    """
    from .defaults import PC_LUMI_MIN_PATCH_SIZE
    from .pointcloud import geometry

    calib = geometry.load_calibration()
    min_patch = params.get("lumi_min_patch_size", PC_LUMI_MIN_PATCH_SIZE)
    z_2d, lumi_2d = np.asarray(z_2d), np.asarray(lumi_2d)
    valid = geometry.lumi_valid_mask(lumi_2d, min_patch) & (z_2d > 0)
    if not valid.any():
        raise ValueError("no valid pixels in the raw scan (check lumi_min_patch_size)")
    y_mm_per_px = calib["y_mm_per_pixel"]  # constant line spacing of the scanner
    raw_points = geometry.extract_points(z_2d, valid, calib["x_mm_per_pixel"], y_mm_per_px, calib["z_mm_per_unit"])
    processed_points = np.asarray(processed_points, dtype=float)

    rng = np.random.default_rng(0)

    def _subsample(points: np.ndarray) -> np.ndarray:
        if len(points) <= max_points:
            return points
        return points[rng.choice(len(points), size=max_points, replace=False)]

    raw_s, proc_s = _subsample(raw_points), _subsample(processed_points)
    sim_s = _subsample(np.asarray(sim_points, dtype=float)) if sim_points is not None else None
    n_panels = 2 if sim_s is None else 4
    if figsize is None:
        figsize = (9.0, 3.2) if n_panels == 2 else (17.0, 3.4)

    def _scatter(ax, pts, values, cmap, label, vmin, vmax):
        sc = ax.scatter(pts[:, 0], pts[:, 1], c=values, s=0.3, cmap=cmap, vmin=vmin, vmax=vmax, rasterized=True)
        fig.colorbar(sc, ax=ax, shrink=0.8, label=label)

    fig, axes = plt.subplots(1, n_panels, figsize=figsize, sharex=True, sharey=True)
    # Raw z is centred on its mean, processed and simulation share the
    # simulation frame: the raw panel gets its own colour scale, the others
    # share one.
    if len(raw_s):
        _scatter(axes[0], raw_s, raw_s[:, 2], "viridis", "Z in mm", *np.percentile(raw_s[:, 2], [1, 99]))
    z_ref = proc_s[:, 2] if len(proc_s) else raw_s[:, 2]
    vmin, vmax = np.percentile(z_ref, [1, 99]) if len(z_ref) else (0, 1)
    if len(proc_s):
        _scatter(axes[1], proc_s, proc_s[:, 2], "viridis", "Z in mm", vmin, vmax)
    axes[0].set_title(f"(a) Raw, calibrated\n{len(raw_points):,} points")
    title = f"(b) Processed\n{len(processed_points):,} points"
    if stats and "outlier_pct" in stats:
        title += f", {float(stats['outlier_pct']):.1f} % removed"
    axes[1].set_title(title)

    if sim_s is not None:
        from scipy.spatial import cKDTree

        _scatter(axes[2], sim_s, sim_s[:, 2], "viridis", "Z in mm", vmin, vmax)
        sim_title = "(c) Matched simulation"
        if stats and "simulation_id" in stats:
            sim_title += f" {int(stats['simulation_id'])}"
        axes[2].set_title(f"{sim_title}\n{len(sim_points):,} points")
        distances, _ = (
            cKDTree(np.asarray(sim_points, dtype=float)).query(proc_s) if len(proc_s) else (np.zeros(0), None)
        )
        if len(distances):
            _scatter(axes[3], proc_s, distances, "magma", "Distance in mm", 0.0, np.percentile(distances, 99))
        axes[3].set_title(
            f"(d) Distance to simulation\nmedian {np.median(distances):.2f} mm"
            if len(distances)
            else "(d) Distance to simulation"
        )

    for ax in axes:
        ax.set_aspect("equal")
        ax.set_xlabel("X in mm")
        ax.grid(True, alpha=0.3)
    axes[0].set_ylabel("Y in mm")
    fig.tight_layout()
    return fig