add-filter

name	add-filter
description	Guide for adding a new filter to PhysicsNeMo Curator. Covers discovery questions, implementation patterns (pass-through, stateful, in-place), generator semantics, flush/artifacts, dashboard widgets, merge for parallel workers, testing, and registration.

Adding a New Filter

This skill walks through adding a new Filter to PhysicsNeMo Curator, from initial design through implementation, testing, and registration.

Step 0: Gather Requirements

Before writing any code, answer these questions. Ask the user if anything is unclear.

Domain

Which submodule does this filter belong to?

Domain	Type parameter	Submodule	Dependency group
mesh	Filter["Mesh"]	src/physicsnemo_curator/domains/mesh/	mesh (physicsnemo, pyvista, pyarrow, torch)
da	Filter["xr.DataArray"]	src/physicsnemo_curator/domains/da/	da (xarray, earth2studio, zarr)
atm	Filter["AtomicData"]	src/physicsnemo_curator/domains/atm/	atm (nvalchemi, ase, torch)

Filter Design

1. Purpose -- What does this filter do? (compute statistics, convert precision, extract metadata, transform data, validate, etc.)
2. Pass-through or transforming? -- Does it yield the input unchanged (side-effect only), or does it modify the data before yielding?
3. Stateful? -- Does the filter accumulate state across items? (e.g. running statistics, row accumulation for output files)
4. Output -- Does it produce a side-effect output? If so, what format? (Parquet, JSON-lines, Zarr, logging only, none)
5. Flush required? -- If stateful, does it need a flush() method to write accumulated results at the end?
6. Artifacts? -- If it writes files, should it implement artifacts() to report paths to the pipeline store?
7. Dashboard widget? -- Should it provide a dashboard_panel() for interactive visualization of its output?
8. Parallel merge? -- If stateful, does it need a merge() static method to combine per-worker output files?
9. Parameters -- What user-configurable parameters does the filter need? (output path, dtype, thresholds, flags, etc.)
10. Cardinality -- Does it yield exactly 1 item per input, or can it expand (1-to-many) or contract (many-to-fewer)?
11. Helper logic -- Does it need module-level helper functions or private helper classes? (e.g. recursive TensorDict traversal, accumulator objects)
12. Dependencies -- Does it require additional imports beyond the domain group? (e.g. specific PyArrow, h5py, scipy modules)
13. Pipeline position -- Where in a typical pipeline would this filter go? (early for validation, middle for transforms, late for statistics)

Reference: Existing Filters

Filter	File	Domain	Pattern	Good example for
MeshInfoFilter	mesh/filters/mesh_info.py	Mesh	Pass-through + logging + JSON-lines	Optional output, log levels
MeanFilter	mesh/filters/mean.py	Mesh	Pass-through + accumulate rows	Parquet output, simple flush
PrecisionFilter	mesh/filters/precision.py	Mesh	In-place modification	Dtype conversion, validation
FieldSelectFilter	mesh/filters/field_select.py	Mesh	In-place modification	Include/exclude filtering
RandomPermutationFilter	mesh/filters/random_permutation.py	Mesh	In-place modification	Stateless data shuffling
EdgeComputeFilter	mesh/filters/edge_compute.py	Mesh	In-place modification	Adding computed data to mesh
WallNodeFilter	mesh/filters/wall_node.py	Mesh	In-place modification	Complex geometric filtering
MeshQualityFilter	mesh/filters/quality.py	Mesh	Stateful + flush + merge	Quality report, Parquet output
MeshStatsFilter	mesh/filters/stats.py	Mesh	Stateful + flush + merge + dashboard	Full pattern, Welford stats
DataArrayStatsFilter	da/filters/stats.py	DataArray	Stateful + flush + merge + dashboard	Zarr output, online moments
AtomicInfoFilter	atm/filters/atomic_info.py	AtomicData	Pass-through + logging

Step 1: Create the Filter Module

Create the filter file at the appropriate location:

• mesh: src/physicsnemo_curator/domains/mesh/filters/<name>.py
• da: src/physicsnemo_curator/domains/da/filters/<name>.py
• atm: src/physicsnemo_curator/domains/atm/filters/<name>.py

Required SPDX Header

Every file MUST start with:

# SPDX-FileCopyrightText: Copyright (c) 2025 - 2026 NVIDIA CORPORATION & AFFILIATES.
# SPDX-FileCopyrightText: All rights reserved.
# SPDX-License-Identifier: Apache-2.0
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

Import Block Template

"""Brief description of what this filter does."""

from __future__ import annotations

import pathlib
from typing import TYPE_CHECKING, Any, ClassVar

# Third-party imports as needed (torch, pyarrow, numpy, etc.)

from physicsnemo_curator.core.base import Filter, Param
from physicsnemo_curator.core.logging import get_logger

if TYPE_CHECKING:
    from collections.abc import Generator

    from physicsnemo.mesh import Mesh  # or: import xarray as xr

Note: Prefer get_logger(self) from physicsnemo_curator.core.logging over raw logging.getLogger(__name__). It provides worker-aware structured logging with timing and is captured by the TUI/database.

Filter Class Template

class <ClassName>(Filter["Mesh"]):
    """Detailed description of the filter.

    <2-3 sentence description of what this filter computes or transforms,
    when you would use it in a pipeline, and what output it produces (if any).>

    Parameters
    ----------
    param1 : str
        Description of first parameter.
    param2 : int
        Description of second parameter.

    Examples
    --------
    >>> filt = <ClassName>(param1="value")  # doctest: +SKIP
    >>> pipeline = source.filter(filt).write(sink)  # doctest: +SKIP
    """

    name: ClassVar[str] = "<Display Name>"
    description: ClassVar[str] = "<Short one-line description for CLI>"

    @classmethod
    def params(cls) -> list[Param]:
        """Return configurable parameters for this filter.

        Returns
        -------
        list[Param]
            Parameter descriptors.
        """
        return [
            Param(
                name="param1",
                description="Description of param1",
                type=str,
                default="default_value",
            ),
        ]

    def __init__(self, param1: str = "default_value") -> None:
        """Initialize the filter.

        Parameters
        ----------
        param1 : str
            First parameter.
        """
        self._log = get_logger(self)
        self._param1 = param1
        # Validate parameters in __init__

    def __call__(self, items: Generator[Mesh]) -> Generator[Mesh]:
        """Process a stream of meshes.

        Parameters
        ----------
        items : Generator[Mesh]
            Stream of incoming meshes.

        Yields
        ------
        Mesh
            Processed meshes.
        """
         mesh  items:
            
             mesh

Generator Semantics

The __call__ method receives and returns a Generator[T]. The standard pattern is:

def __call__(self, items: Generator[Mesh]) -> Generator[Mesh]:
    for mesh in items:
        # ... process ...
        yield mesh

Key rules:

• Always yield -- filters must yield items (even if unmodified) for downstream filters and sinks to receive them
• Lazy processing -- iterate the input generator lazily, do not consume the whole stream upfront unless absolutely necessary
• 1-to-1 is most common -- every existing filter yields exactly one item per input item
• 1-to-many is allowed -- yield multiple items per input if needed (e.g. splitting a multi-block mesh into individual blocks)
• Filtering (0-to-1) is allowed -- skip yielding for items that fail a condition

Implementation Patterns

Pattern A: Pass-Through (Side-Effect Only)

The filter examines but does not modify the data. Used for logging, metadata extraction, and statistics accumulation.

def __call__(self, items: Generator[Mesh]) -> Generator[Mesh]:
    for mesh in items:
        info = self._extract_info(mesh)
        self._log.info("Processed mesh: %s", info["name"])
        self._rows.append(info)
        yield mesh  # unchanged

Examples: MeshInfoFilter, MeanFilter, MeshStatsFilter, DataArrayStatsFilter, AtomicInfoFilter

Pattern B: In-Place Modification

The filter modifies the data before yielding. Used for dtype conversion, field selection, coordinate transforms, edge computation.

def __call__(self, items: Generator[Mesh]) -> Generator[Mesh]:
    for mesh in items:
        if mesh.point_data is not None:
            self._convert_fields(mesh.point_data)
        yield mesh  # modified in-place

Examples: PrecisionFilter, FieldSelectFilter, RandomPermutationFilter, EdgeComputeFilter, WallNodeFilter

Pattern C: Stateful with Flush

The filter accumulates results across items and writes output when flushed. Requires flush(), artifacts(), and _output_path.

def __init__(self, output: str) -> None:
    self._log = get_logger(self)
    self._output_path = pathlib.Path(output)
    self._rows: list[dict[str, object]] = []
    self._last_artifacts: list[str] = []

def __call__(self, items: Generator[Mesh]) -> Generator[Mesh]:
    for mesh in items:
        row = self._compute_row(mesh)
        self._rows.append(row)
        yield mesh

def flush(self) -> str | None:
    """Write accumulated results to output file.

    Returns
    -------
    str or None
        Path of written file, or ``None`` if nothing to write.
    """
    if not self._rows:
        return None
    # Write self._rows to self._output_path
    self._output_path.parent.mkdir(parents=True, exist_ok=True)
    # ... write data ...
    path = str(self._output_path)
    self._rows.clear()
    self._last_artifacts = [path]
    return path

 () -> []:
    
    paths = ._last_artifacts
    ._last_artifacts = []
     paths

Examples: MeshQualityFilter (Parquet), MeanFilter (Parquet), MeshInfoFilter (JSON-lines)

Pattern D: Stateful + Flush + Merge + Dashboard (Full Pattern)

The complete pattern for parallel-aware statistics filters with interactive visualization. Adds merge() and dashboard_panel().

class MyStatsFilter(Filter["Mesh"]):
    name: ClassVar[str] = "My Statistics"
    description: ClassVar[str] = "Compute statistics and save to Parquet"

    def __init__(self, output: str) -> None:
        self._log = get_logger(self)
        self._output_path = pathlib.Path(output)
        self._rows: list[dict[str, object]] = []
        self._last_artifacts: list[str] = []

    def __call__(self, items: Generator[Mesh]) -> Generator[Mesh]:
        for mesh in items:
            self._rows.extend(self._compute_stats(mesh))
            yield mesh

    def flush(self) -> str | None:
        """Write accumulated stats; append if file exists."""
        if not self._rows:
            return None
        # ... write/append to Parquet ...
        path = str(self._output_path)
        self._rows.clear()
        ._last_artifacts = [path]
         path

     () -> []:
        
        paths = ._last_artifacts
        ._last_artifacts = []
         paths


     () -> :
        
        
         output


     () -> :
        
         holoviews  hv
         panel  pn
        
         pn.Row(sidebar, plot_pane, sizing_mode=)


     () -> [, ]:
        
         {: , : }

Examples: MeshStatsFilter, DataArrayStatsFilter, AtomicStatsFilter

Pipeline Runner Integration (flush / artifacts)

The pipeline runner (run_pipeline) automatically manages stateful filters. Understanding this is critical for correct implementation:

1. Auto-flush: After each pipeline index is processed, the runner calls flush() on every filter that has both a flush method and an _output_path attribute.

2. Worker-specific paths: The runner rewrites _output_path to include a worker ID suffix for process-based parallel backends:

   • If the path contains {worker_id}, it's used as a template
   • Otherwise, the path is rewritten as {stem}_worker_{pid}{suffix}

3. Artifact tracking: After flushing, if the pipeline has track_metrics=True, the runner calls artifacts() and records the paths in the pipeline database.

4. Append semantics: Because flush() is called after each index, filters should append to existing files rather than overwrite. This is important — each flush adds the current index's data to the worker's output file.

Convention: A filter's _output_path attribute is the contract with the runner. If your filter has flush() but no _output_path, the runner will skip it.

# The runner does essentially this after each index:
for f in pipeline.filters:
    if hasattr(f, "flush") and hasattr(f, "_output_path"):
        # Rewrite path for this worker
        f._output_path = worker_specific_path
        f.flush()
        # Record artifacts
        artifact_paths = f.artifacts()

Flush Design Principles

1. Return value: flush() returns str | None — the path written, or None if nothing to write.

2. Append on subsequent calls: Since the runner calls flush() after each index, your flush should append new data to the existing file:

   def flush(self) -> str | None:
       if not self._rows:
           return None
   
       new_table = pa.table(...)
       self._output_path.parent.mkdir(parents=True, exist_ok=True)
   
       # Append to existing file if present
       if self._output_path.exists():
           existing = pq.read_table(str(self._output_path))
           combined = pa.concat_tables([existing, new_table])
           pq.write_table(combined, str(self._output_path))
       else:
           pq.write_table(new_table, str(self._output_path))
   
       path = str(self._output_path)
       self._rows.clear()
       self._last_artifacts = [path]
       return path
   

3. Clear state after flush: Always clear the in-memory accumulator (self._rows.clear()) after writing.

4. Update _last_artifacts: Set the list of written paths so artifacts() can report them.

The merge() Static Method

When running with parallel backends (n_jobs > 1), each worker writes its own output file. After all workers finish, run_pipeline calls merge() to combine them:

@staticmethod
def merge(parquet_paths: list[str], output: str) -> str:
    """Concatenate per-worker output files.

    Parameters
    ----------
    parquet_paths : list[str]
        Paths to per-worker files.
    output : str
        Path for the merged file.

    Returns
    -------
    str
        Path of written merged file.

    Raises
    ------
    ValueError
        If parquet_paths is empty.
    """
    if not parquet_paths:
        msg = "parquet_paths must be a non-empty list."
        raise ValueError(msg)

    tables = [pq.read_table(p) for p in parquet_paths]
    merged = pa.concat_tables(tables)

    out_path = pathlib.Path(output)
    out_path.parent.mkdir(parents=True, exist_ok=True)
    pq.write_table(merged, str(out_path))
    return str(out_path)

The runner detects merge capability by checking for all three attributes: flush, _output_path, and merge.

Dashboard Widget Pattern

Filters can provide an interactive visualization widget for the dashboard UI. This is a classmethod — it receives artifact paths and returns a Panel component.

When to implement: If your filter writes output files (Parquet, Zarr) that users would benefit from exploring visually.

Key design principles:

1. Import Panel/HoloViews inside the method — never at module level. Dashboard dependencies are optional.
2. Return pn.Row(sidebar, plot_pane) — standard layout with controls on the left and the visualization on the right.
3. Handle missing data gracefully — return pn.pane.Markdown("*No data*") when artifacts don't exist.
4. Use @pn.depends — for reactive widgets that update the plot.

@classmethod
def dashboard_panel(
    cls,
    artifact_paths: list[str],
    selected_index: int | None = None,
) -> Any:
    """Return an interactive widget for this filter's artifacts.

    Parameters
    ----------
    artifact_paths : list[str]
        Paths to artifact files produced by this filter.
    selected_index : int or None
        Currently selected pipeline index, if any.

    Returns
    -------
    pn.viewable.Viewable or None
        A Panel component, or None if no widget.
    """
    import holoviews as hv
    import pandas as pd
    import panel as pn
    from bokeh.models import HoverTool

    hv.extension("bokeh")

    if not artifact_paths:
        return pn.pane.Markdown("*No artifacts found.*")

    # Load data from artifact files
    frames = []
    for path in artifact_paths:
        try:
            frames.append(pd.read_parquet(path))
        except Exception:  # noqa: BLE001
            continue

    if not frames:
        return pn.pane.Markdown("*Could not read any artifacts.*")

    df = pd.concat(frames, ignore_index=True)

    # Create widgets
    x_select = pn.widgets.Select(name="X-Axis", options=columns, value="mean")
    y_select = pn.widgets.Select(name="Y-Axis", options=columns, value=)

    sidebar = pn.Column(, x_select, y_select, width=)


     () -> hv.Points:
        
        points = hv.Points(df, kdims=[x_col, y_col])
         points.opts(
            color=,
            size=,
            responsive=,
            height=,
            tools=[, , , ],
        )

    plot_pane = pn.pane.HoloViews(update_plot, sizing_mode=)

     pn.Row(sidebar, plot_pane, sizing_mode=, min_height=)


 () -> [, ]:
    
     {: , : }

Default hints (from the base class): {"cols": 6, "rows": 2}

Output Format Patterns

Parquet Output (MeshStatsFilter / MeshQualityFilter pattern)

import pyarrow as pa
import pyarrow.parquet as pq

def flush(self) -> str | None:
    if not self._rows:
        return None

    new_table = pa.table(
        {col: [row.get(col) for row in self._rows] for col in self._schema.names},
        schema=self._schema,
    )
    self._output_path.parent.mkdir(parents=True, exist_ok=True)

    # Append to existing file (worker-level aggregation)
    if self._output_path.exists():
        existing = pq.read_table(str(self._output_path))
        combined = pa.concat_tables([existing, new_table])
        pq.write_table(combined, str(self._output_path))
    else:
        pq.write_table(new_table, str(self._output_path))

    path = str(self._output_path)
    self._rows.clear()
    self._last_artifacts = [path]
    return path

JSON-Lines Output (MeshInfoFilter pattern)

import json

def __init__(self, output: str | None = None) -> None:
    self._log = get_logger(self)
    self._file_handle: TextIO | None = None
    if output:
        self._output_path = pathlib.Path(output)

def _write_to_file(self, info: dict) -> None:
    if self._file_handle is None and hasattr(self, "_output_path"):
        self._output_path.parent.mkdir(parents=True, exist_ok=True)
        self._file_handle = self._output_path.open("w")
    if self._file_handle is not None:
        self._file_handle.write(json.dumps(info) + "\n")

def flush(self) -> str | None:
    if self._file_handle is not :
        ._file_handle.close()
         (._output_path)
     

Zarr Output (DataArrayStatsFilter pattern)

def flush(self) -> str | None:
    if not self._accumulators:
        return None

    output_path = self._output_path
    output_path.mkdir(parents=True, exist_ok=True)

    for var_name, acc in self._accumulators.items():
        new_stats = acc.finalize()
        group_path = output_path / var_name

        # Merge with existing (worker-level aggregation)
        if group_path.exists():
            existing = xr.open_zarr(str(group_path))
            merged = _merge_moment_datasets([existing, new_stats])
            merged.to_zarr(str(group_path), mode="w", zarr_format=3)
        else:
            new_stats.to_zarr(str(group_path), mode="w", zarr_format=3)

    path = str(output_path)
    self._accumulators.clear()
    self._last_artifacts = [path]
    return path

Helper Function Patterns

Module-level private helpers for reusable logic:

def _extract_leaf_tensors(
    td: object, prefix: str = ""
) -> list[tuple[str, torch.Tensor]]:
    """Recursively extract leaf tensors from a TensorDict.

    Parameters
    ----------
    td : object
        TensorDict-like object.
    prefix : str
        Key prefix for nested access.

    Returns
    -------
    list[tuple[str, torch.Tensor]]
        Pairs of (key_path, tensor).
    """
    from tensordict import TensorDictBase

    if not isinstance(td, TensorDictBase):
        return []

    results: list[tuple[str, torch.Tensor]] = []
    for key in td.keys():  # noqa: SIM118
        child = td[key]
        full_key = f"{prefix}{key}" if prefix else key
        if isinstance(child, TensorDictBase):
            results.extend(_extract_leaf_tensors(child, prefix=f"{full_key}/"))
        elif isinstance(child, torch.Tensor):
            results.append((full_key, child))
    return results

Helper classes for complex state:

class _MyAccumulator:
    """Internal accumulator for running computations.

    Parameters
    ----------
    dims : list[str]
        Dimensions to retain (not reduced).
    """

    def __init__(self, dims: list[str]) -> None:
        self._dims = dims
        self._count: int = 0
        self._mean: np.ndarray | None = None
        self._m2: np.ndarray | None = None

    def update(self, data: np.ndarray) -> None:
        """Incorporate a new observation (Welford's online algorithm)."""
        data = np.asarray(data, dtype=np.float64)
        if self._mean is None:
            self._mean = np.zeros_like(data)
            self._m2 = np.zeros_like(data)
        self._count += 1
        delta = data - self._mean
        self._mean += delta / self._count
        delta2 = data - self._mean
        self._m2 += delta * delta2

    def finalize(self) -> dict[str, np.ndarray]:
        """Return final computed values."""
        variance = ._m2 / ._count  ._count >   np.zeros_like(._mean)
         {: ._mean, : variance, : ._count}

Param Declaration Patterns

Required parameter (no default -- user must provide):

Param(name="output", description="Output file path", type=str)

Optional with default:

Param(name="log_level", description="Logging level", type=str,
      default="info", choices=["info", "debug"])

Boolean flag:

Param(name="include_fields", description="Include field details",
      type=bool, default=True)

With constrained choices:

Param(name="target_dtype", description="Target floating-point dtype",
      type=str, default="float32",
      choices=["float32", "float16", "bfloat16"])

None-able (optional output):

Param(name="output", description="Optional output path", type=str,
      default=None)

Step 2: Write Tests

Create the test file at test/domains/<domain>/test_<name>.py.

Test File Structure

"""Tests for <ClassName>."""

# SPDX header (same as source files)

from __future__ import annotations

import pathlib

import numpy as np
import pyvista as pv
import pytest

pytestmark = pytest.mark.requires("<domain>")


# ---------------------------------------------------------------------------
# Test helpers
# ---------------------------------------------------------------------------

def _create_test_mesh(
    directory: pathlib.Path, name: str = "test.vtu"
) -> pathlib.Path:
    """Create a simple VTK unstructured grid for testing.

    Returns the path to the written file.
    """
    points = np.array(
        [[0, 0, 0], [1, 0, 0], [1, 1, 0], [0, 1, 0]], dtype=np.float64
    )
    cells = np.array([[3, 0, 1, 2], [3, 0, 2, 3]])
    cell_types = np.array([5, 5])  # VTK_TRIANGLE

    grid = pv.UnstructuredGrid(cells, cell_types, points)
    grid.point_data["temperature"] = np.array([100.0, 200.0, 300.0, ])
    grid.point_data[] = np.array([, , , ])
    grid.cell_data[] = np.array([, ])

    directory.mkdir(parents=, exist_ok=)
    path = directory / name
    grid.save((path))
     path

Alternatively, for filters that do not need VTK files, you can construct Mesh objects directly using mock data (see the NavierStokesCylinderSource test pattern in test/domains/mesh/test_ns_cylinder.py).

Test Classes

Unit Tests (Metadata)

class Test<ClassName>Unit:
    """Metadata tests."""

    def test_params_list(self) -> None:
        from <module> import <ClassName>

        params = <ClassName>.params()
        assert len(params) >= 0
        names = [p.name for p in params]
        # Assert expected parameter names

    def test_name_and_description(self) -> None:
        from <module> import <ClassName>

        assert isinstance(<ClassName>.name, str)
        assert len(<ClassName>.name) > 0
        assert len(<ClassName>.description) > 0

Pass-Through Tests

@pytest.mark.integration
class Test<ClassName>Integration:
    """Tests against local test data."""

    def test_yields_mesh_unchanged(self, tmp_path: pathlib.Path) -> None:
        """Filter should yield the mesh without modification."""
        from physicsnemo_curator.domains.mesh.sources.vtk import VTKSource

        _create_test_mesh(tmp_path / "vtk")
        source = VTKSource.from_path(str(tmp_path / "vtk"))
        mesh_before = next(source[0])

        filt = <ClassName>(...)

        def gen():
            yield mesh_before

        meshes_out = list(filt(gen()))
        assert len(meshes_out) == 1
        assert meshes_out[0] is mesh_before  # identity check

Flush and Artifacts Tests

    def test_flush_writes_output(self, tmp_path: pathlib.Path) -> None:
        """Stateful filter writes output on flush."""
        _create_test_mesh(tmp_path / "vtk")
        source = VTKSource.from_path(str(tmp_path / "vtk"))

        output_path = tmp_path / "output.parquet"
        filt = <ClassName>(output=str(output_path))
        list(filt(source[0]))
        result = filt.flush()

        assert result == str(output_path)
        assert output_path.exists()

    def test_artifacts_returns_paths(self, tmp_path: pathlib.Path) -> None:
        """artifacts() returns paths written by flush."""
        output_path = tmp_path / "output.parquet"
        filt = <ClassName>(output=str(output_path))
        # ... process some data ...
        list(filt(gen()))
        filt.flush()

        paths = filt.artifacts()
        assert paths == [str(output_path)]

        # Second call returns empty (consumed)
        assert filt.artifacts() == []

    def test_flush_appends_on_second_call(self, tmp_path: pathlib.Path) -> None:
        """Subsequent flush appends to existing file."""
        output_path = tmp_path / "output.parquet"
        filt = <ClassName>(output=str(output_path))

        # Process first batch
        list(filt(gen_batch_1()))
        filt.flush()

        # Process second batch
        (filt(gen_batch_2()))
        filt.flush()

        
        table = pq.read_table((output_path))
         table.num_rows == expected_total_rows

Merge Tests

    def test_merge_combines_worker_files(self, tmp_path: pathlib.Path) -> None:
        """merge() concatenates per-worker output files."""
        # Create two separate worker files
        worker_0 = tmp_path / "stats_worker_0.parquet"
        worker_1 = tmp_path / "stats_worker_1.parquet"
        # ... write data to each ...

        merged_path = tmp_path / "merged.parquet"
        result = <ClassName>.merge(
            [str(worker_0), str(worker_1)],
            output=str(merged_path),
        )

        assert result == str(merged_path)
        merged = pq.read_table(str(merged_path))
        assert merged.num_rows == expected_combined_rows

Dashboard Tests

    @pytest.mark.requires("dashboard")
    def test_dashboard_panel_returns_widget(self, tmp_path: pathlib.Path) -> None:
        """dashboard_panel returns a Panel component."""
        # Create an artifact file
        output_path = tmp_path / "stats.parquet"
        # ... write test data ...

        widget = <ClassName>.dashboard_panel([str(output_path)])
        assert widget is not None

    def test_dashboard_panel_no_data(self) -> None:
        """dashboard_panel handles empty artifact list."""
        widget = <ClassName>.dashboard_panel([])
        assert widget is not None  # Returns Markdown message

    def test_dashboard_layout_hints(self) -> None:
        """Layout hints return valid grid dimensions."""
        hints = <ClassName>.dashboard_layout_hints()
        assert "cols" in hints
        assert "rows" in hints
        assert 1 <= hints["cols"] <= 12
        assert hints["rows"] >= 1

Logging Tests

    def test_logs_output(
        self, tmp_path: pathlib.Path, caplog: pytest.LogCaptureFixture
    ) -> None:
        """Filter logs expected information."""
        import logging

        _create_test_mesh(tmp_path / "vtk")
        source = VTKSource.from_path(str(tmp_path / "vtk"))

        filt = <ClassName>(...)

        with caplog.at_level(logging.INFO):
            list(filt(source[0]))

        assert "expected text" in caplog.text

Pipeline Integration Tests

@pytest.mark.e2e
class Test<ClassName>Pipeline:
    """End-to-end pipeline tests."""

    def test_chained_in_pipeline(self, tmp_path: pathlib.Path) -> None:
        """Filter works correctly in a multi-filter pipeline."""
        from physicsnemo_curator.domains.mesh.sinks.mesh_writer import MeshSink
        from physicsnemo_curator.domains.mesh.sources.vtk import VTKSource

        vtk_dir = tmp_path / "vtk"
        _create_test_mesh(vtk_dir, "mesh_0.vtu")
        _create_test_mesh(vtk_dir, "mesh_1.vtu")

        output_dir = tmp_path / "output"

        filt = <ClassName>(...)

        pipeline = (
            VTKSource.from_path(str(vtk_dir))
            .filter(filt)
            .write(MeshSink(output_dir=str(output_dir)))
        )

        assert len(pipeline) == 2

        for i in range(len(pipeline)):
            paths = pipeline[i]
            assert len(paths) == 1
            assert pathlib.Path(paths[0]).exists()

        # Flush if stateful
        # filt.flush()

Registry Tests

class Test<ClassName>Registry:
    """Test that the filter is registered."""

    def test_filter_registered(self) -> None:
        from physicsnemo_curator.core.registry import registry

        names = [f.name for f in registry.list_filters("<domain>")]
        assert "<Display Name>" in names

Test Markers Reference

Marker	Purpose
pytestmark = pytest.mark.requires("mesh")	Module-level: skip all if mesh deps missing
@pytest.mark.requires("da")	Skip if da dependencies not installed
@pytest.mark.requires("atm")	Skip if atm dependencies not installed
@pytest.mark.requires("dashboard")	Skip if dashboard deps not installed
@pytest.mark.integration	Tests that touch filesystem
@pytest.mark.e2e	End-to-end pipeline tests
@pytest.mark.slow	Slow tests, excluded from quick CI

Running Tests

# Unit and integration tests (fast, no network)
uv run pytest test/domains/<domain>/test_<name>.py -v

# Pipeline integration test
uv run pytest test/domains/<domain>/test_<name>.py -v -k "Pipeline"

# Full test suite to check for regressions
uv run pytest test/domains/<domain>/ -v -k "not slow"

Step 3: Register the Filter

Edit the domain __init__.py

For mesh filters, edit src/physicsnemo_curator/domains/mesh/__init__.py:

# Add import (alphabetical order among filters)
from physicsnemo_curator.domains.mesh.filters.<module> import <ClassName>

# Add registration (after existing register_filter calls)
registry.register_filter("mesh", <ClassName>)

# Add to __all__ (alphabetical order)
__all__ = [
    ...,
    "<ClassName>",
    ...,
]

For da filters, edit src/physicsnemo_curator/domains/da/__init__.py. For atm filters, edit src/physicsnemo_curator/domains/atm/__init__.py.

Export any public helpers

If the filter provides a public helper function (like merge_welford_stats in MeshStatsFilter), add it to the import and __all__ as well:

from physicsnemo_curator.domains.mesh.filters.<module> import <ClassName>, <helper_func>

__all__ = [
    ...,
    "<ClassName>",
    "<helper_func>",
    ...,
]

Step 4: Quality Checks

Run all checks before committing:

# Format
uv run ruff format \
  src/physicsnemo_curator/domains/<domain>/filters/<name>.py \
  test/domains/<domain>/test_<name>.py

# Lint
uv run ruff check --fix \
  src/physicsnemo_curator/domains/<domain>/filters/<name>.py \
  test/domains/<domain>/test_<name>.py

# Docstring coverage (must be >= 99%)
uv run interrogate

# Type checking
uv run ty check

# Run filter tests
uv run pytest test/domains/<domain>/test_<name>.py -v

# Run full domain test suite for regressions
uv run pytest test/domains/<domain>/ -v -k "not slow"

Step 5: Commit

Use the commit tool (or follow Conventional Commits format):

feat(<domain>): add <ClassName> for <purpose>

<Optional body describing what the filter computes and what output it produces.>

Checklist

Before considering the filter complete, verify:

Core Implementation

• SPDX license headers on all new files
• Filter inherits from Filter["Mesh"], Filter["xr.DataArray"], or Filter["AtomicData"]
• name and description ClassVars are set
• params() returns all configurable parameters
• __init__() validates parameter constraints
• __init__() uses self._log = get_logger(self) for logging
• __call__() receives Generator[T], yields Generator[T]
• Generator is consumed lazily (no upfront list() conversion)
• Pass-through: yields items for downstream consumption

Stateful Filters (if applicable)

• flush() method returns str | None
• flush() appends to existing file (not overwrites) for per-index semantics
• flush() clears accumulated state after writing
• _output_path attribute set as pathlib.Path (required for runner integration)
• artifacts() method returns paths written by last flush, then clears
• _last_artifacts: list[str] instance variable for tracking
• merge() static method for combining per-worker output files
• merge() raises ValueError on empty input list

Dashboard (if applicable)

• dashboard_panel() classmethod returns Panel widget or None
• Panel/HoloViews imports inside method body (not at module level)
• Handles empty/missing artifacts gracefully (returns Markdown message)
• dashboard_layout_hints() classmethod returns {"cols": N, "rows": M}
• Widget uses @pn.depends for reactive updates
• Layout uses pn.Row(sidebar, plot_pane, sizing_mode="stretch_width")

Documentation & Style

• NumPy-style docstrings on class, __init__, __call__, flush, artifacts, merge, dashboard_panel
• Helper functions/classes have docstrings and type annotations
• from __future__ import annotations at top
• TYPE_CHECKING block for Generator and domain type imports

Registration

• Registered in domain __init__.py with registry.register_filter()
• Added to __all__ in domain __init__.py
• Public helpers also exported in __all__

Testing

• Unit tests: params, name, description
• Integration tests: pass-through, output files, computed values
• Flush/artifacts tests (if stateful)
• Merge tests (if parallel-aware)
• Dashboard tests (if widget provided)
• Pipeline test: filter works in a chained pipeline (marked @pytest.mark.e2e)
• Registry test: filter is discoverable

Quality Gates

• ruff format clean
• ruff check clean
• interrogate >= 99%
• ty check clean
• All tests pass
• No regressions in existing tests

Reference: Filter[T] ABC

From src/physicsnemo_curator/core/base.py:

class Filter[T](ABC):
    """Abstract filter/transform that processes a stream of T items."""

    name: ClassVar[str]
    description: ClassVar[str]

    @classmethod
    @abstractmethod
    def params(cls) -> list[Param]: ...

    @abstractmethod
    def __call__(self, items: Generator[T]) -> Generator[T]: ...

    def artifacts(self) -> list[str]:
        """Return paths of files produced since the last call.

        Override in stateful filters that write side-effect files.
        The framework calls this after each index to record artifacts.
        Default returns [].
        """
        return []

    @classmethod
    def dashboard_panel(
        cls,
        artifact_paths: list[str],
        selected_index: int | None = None,
    ) -> Any:
        """Return a Panel component visualizing this filter's artifacts.

        Override to provide a custom dashboard widget.
        Default returns None (no widget).
        """
        return None

    @classmethod
    def dashboard_layout_hints(cls) -> dict[str, int]:
        """Declare grid space for dashboard GridStack placement.

        Default: {"cols": 6, "rows": 2}
        """
         {: , : }

Key differences from Source[T]:

• Sources have __len__ and __getitem__ -- filters do not
• Filters receive a generator and return a generator (__call__)
• Sources are indexed by integer -- filters process streams
• Filters optionally provide flush(), artifacts(), merge() for stateful behavior
• Filters optionally provide dashboard_panel() for visualization

Reference: Param Dataclass

From src/physicsnemo_curator/core/base.py:

@dataclass(frozen=True)
class Param:
    """Descriptor for a configurable parameter on a pipeline component."""

    name: str           # Must match __init__ keyword argument
    description: str    # Human-readable help text for CLI
    type: type = str    # Expected Python type (str, int, float, bool, pathlib.Path, ...)
    default: Any = REQUIRED  # Use REQUIRED sentinel for mandatory params
    choices: list[str] | None = None  # CLI shows selection prompt if set

    @property
    def required(self) -> bool:
        """Return True if this parameter has no default value."""
        return self.default is REQUIRED