IO source and sink extension#

MoDaCor separates data access from processing steps through IoSource, IoSources, IoSink, and IoSinks. New readers and writers should follow the same contracts used by the built-in HDF, YAML, CSV, and HDF-processing implementations.

Source contract#

Subclass modacor.io.io_source.IoSource and implement the methods your format supports:

  • get_data(data_key, load_slice=...)

  • get_data_shape(data_key)

  • get_data_dtype(data_key)

  • get_data_attributes(data_key)

  • get_static_metadata(data_key)

Built-in examples:

  • src/modacor/io/hdf/hdf_source.py

  • src/modacor/io/yaml/yaml_source.py

  • src/modacor/io/csv/csv_source.py

  • src/modacor/io/tiled/tiled_source.py

IoSources exposes all sources through a shared ref::path syntax. For example:

  • sample::entry1/instrument/detector00/data

  • defaults::detector/darkcurrent/value

HDF attributes can be addressed with @attribute, for example sample::entry1/instrument/detector00/frame_exposure_time@units.

Sink contract#

Subclass modacor.io.io_sink.IoSink and implement:

  • write(subpath, *args, **kwargs)

Sinks that support incremental assembly may additionally set supports_chunked_writes = True and implement initialize_chunked(...), write_chunk(...), inspect_chunked(...), and finalize_chunked(...). Persistent backends should also implement load_chunked_plan(...) so a new server process can reconstruct a handle, and recover_chunked(...) for backend-specific reconciliation, abandonment, and resumption. Ordinary sinks inherit clear unsupported-capability errors for all optional chunk methods.

IoSinks routes writes through sink_ref::subpath. The current built-in sink examples are:

  • src/modacor/io/csv/csv_sink.py

  • src/modacor/io/hdf/hdf_processing_sink.py

  • src/modacor/io/hdf/hdf_chunked_processing_sink.py

  • src/modacor/io/tiled/tiled_sink.py

Registration paths#

There are three supported ways to add sources or sinks:

  1. Register them directly in Python with IoSources.register_source(...) or IoSinks.register_sink(...).

  2. Add them dynamically inside a pipeline with the AppendSource and AppendSink process steps.

  3. Build them from normalized runtime specs via modacor.io.runtime_support.build_sources_from_specs(...) and build_sinks_from_specs(...).

The shared CLI/runtime builder currently supports:

  • source types: hdf, yaml, csv, buffer, tiled, and custom

  • sink types: csv, hdf, hdf_chunked, hdf_processing, buffer, plotly_json, tiled, and custom

hdf_chunked currently provides the complete programmatic BaseData lifecycle, including weights, uncertainties, and static or batch-dependent axes. Destination selections must currently be contiguous. The runtime builder can construct it, but the server endpoints that coordinate its initialize/write/finalize operations are not implemented yet; track that work in docs/development/design/chunked-sink-implementation-plan.md.

For custom sources or sinks, trusted/local builders can use kwargs.class_path with the fully qualified class import path. Runtime services running with the restricted policy reject arbitrary class_path imports; for that mode, register allowed custom classes in the service process and refer to them with kwargs.class_alias.

Runtime-service sink registrations use the same normalized shape as source registrations:

{
  "ref": "export_csv",
  "type": "csv",
  "location": "/data/out/current.csv",
  "kwargs": {"delimiter": ","}
}

HDF sink registrations may opt into available runtime metadata with kwargs.include_runtime_metadata, but this is not enabled by default. The process-level write_hdf path remains the main metadata-rich HDF artifact export.

Runtime-service profiles#

If you are adding facility-facing runtime workflows, also check src/modacor/server/source_profiles.py. That file defines the named source profiles exposed by the API and CLI, such as mouse and saxsess.

Testing expectations#

Add targeted tests under tests/io/... for format behavior and under tests/modules/... if the new class is used by AppendSource or AppendSink. Current examples include:

  • tests/io/hdf/test_hdf_source.py

  • tests/io/yaml/test_yaml_source.py

  • tests/io/csv/test_csv_source.py

  • tests/modules/base_modules/test_append_source.py

  • tests/modules/base_modules/test_append_sink.py

Practical guidance#

  • Keep source references stable and descriptive; they become part of pipeline configuration.

  • Prefer explicit internal paths and metadata keys over implicit defaults.

  • If the format only supports a subset of the full interface, raise NotImplementedError rather than silently guessing.

  • Reuse the existing registry and runtime-support helpers instead of creating a parallel configuration path.

Tiled source and sink#

Install the optional client with pip install 'modacor[tiled]'. Both modacor.io.tiled.TiledSource and modacor.io.tiled.TiledSink accept a Tiled URL, profile:profile-name, or an existing client via root_node=client. The runtime builders accept type: tiled and preserve the URL. Connection options, including authentication, go in connection_kwargs inside iosource_method_kwargs or iosink_method_kwargs (or runtime kwargs). See the Tiled Python client reference for connection and catalog methods.

from modacor.io.tiled import TiledSource, TiledSink

sink = TiledSink(
    sink_reference="corrected",
    resource_location="profile:beamline",
    iosink_method_kwargs={"base_path": "processed"},
)
sink.write("run_001", processing_data, data_paths=["/sample/signal"])

source = TiledSource(
    source_reference="result",
    resource_location="profile:beamline",
    iosource_method_kwargs={"base_path": "processed/run_001"},
)
array = source.get_data("sample/signal/signal")
units = source.get_static_metadata("sample/signal/signal@units")

A BaseData path such as /sample/signal exports its signal, weights, and named uncertainties arrays. Numeric leaf paths such as /sample/signal/variances/poisson are also supported. Units are stored in array metadata under attrs.units; nonnumeric leaves such as /sample/signal/units are stored in a metadata container and can be read with get_static_metadata("sample/signal/units@value"). Select BaseData roots or individual leaves; whole DataBundle roots are not supported by this sink.

The sink needs a writable Tiled catalog and write credentials. It creates missing containers under the configured base path and rejects existing targets by default. Set iosink_method_kwargs={"overwrite": True} to update existing arrays with the same shape and dtype. Writes occur one array at a time and are not transactional; use a distinct run subpath for separate exports.

The source caches full-array reads and metadata locally and returns caller-owned copies, so in-place processing does not modify later reads. Call source.clear_cache() after server-side updates. Explicit slices bypass the full-array cache. An empty key or @attribute resolves relative to base_path.

Install pip install 'modacor[tiled-tests]' and run python -m pytest tests/io/tiled for source and sink integration tests against an in-process Tiled server with temporary storage. This exercises real client requests and serialization without an external service or listening port.