runner

Base class for intermediate processing stages.

A processor takes raw acquisition files and produces derived files. The runner handles the boilerplate: hashing inputs, locating the upstream AcquisitionManifest, writing a ProcessingManifest sidecar alongside the outputs, and turning declared outputs into ProducerEntry shapes the ingest layer can consume.

Typical use:

from mesofield.processing import ProcessorRunner

class SpikeSorter(ProcessorRunner):

tool_name = “my_lab_spikesort” tool_version = “0.1.0”

def run(self, inputs, *, sigma=4.0):

in_path = inputs[0] out = in_path.parent / “spikes.csv” # … do the work; write to out … return [out]

runner = SpikeSorter() runner([recording_path], sigma=5.0) # → spikes.csv written, plus my_lab_spikesort.process.json next to it

class mesofield.processing.runner.ProcessorRunner

Bases: object

Wrap a file-to-file transformation in a ProcessingManifest contract.

Subclasses must set tool_name and tool_version, and override run(). Calling the instance executes the work and writes the sidecar.

manifest_placement: ClassVar[str] = 'output_dir'

Where the manifest sidecar lands. "output_dir" writes <tool_name>.process.json in the directory of the first output; subclasses can override manifest_path() for custom placement.

run(inputs, **params)

Do the actual work. Return the list of files written.

Parameters:

• inputs (Sequence[Path])

• params (Any)

Return type:

list[Path]

declare_outputs(outputs, params, session_root)

Turn run() outputs into ProducerEntry instances.

Default: one entry per output, with data_type=tool_name and bids_type/file_type inferred from the path. Override for richer declarations (e.g. multiple roles, sidecars).

Parameters:

• outputs (Sequence[Path])

• params (dict[str, Any])

• session_root (Path | None)

Return type:

list[ProducerEntry]