manifest_extractor

nominal.experimental.extractor.decorators.manifest_extractor(fn: Callable[[...], None], *, default_timestamp_column: str | None = None, default_timestamp_type: Iso8601 | Epoch | Relative | Custom | Literal['iso_8601', 'epoch_picoseconds', 'epoch_nanoseconds', 'epoch_microseconds', 'epoch_milliseconds', 'epoch_seconds', 'epoch_minutes', 'epoch_hours', 'epoch_days'] | None = None) → Extractor[ManifestExtractorContext]

nominal.experimental.extractor.decorators.manifest_extractor(*, default_timestamp_column: str | None = None, default_timestamp_type: Iso8601 | Epoch | Relative | Custom | Literal['iso_8601', 'epoch_picoseconds', 'epoch_nanoseconds', 'epoch_microseconds', 'epoch_milliseconds', 'epoch_seconds', 'epoch_minutes', 'epoch_hours', 'epoch_days'] | None = None) → Callable[[Callable[[...], None]], Extractor[ManifestExtractorContext]]

Turn a parsing function into a manifest extractor, recommended for new images.

Declare the callback’s files and settings with @input and @parameter below this decorator. The framework resolves them before invoking the callback; ctx supplies output paths and job metadata. Add @error for expected failure mappings.

Write files under ctx.output_dir and declare each with ManifestExtractorContext.add_tabular(), add_avro_stream(), add_journal_json(), or add_video(). One run can produce several files with different timestamps, tags, and channel prefixes. Extractor.run() writes manifest.json after the callback returns.

Supply both timestamp defaults here to reuse the declarations through Extractor.registration_kwargs() or Extractor.catalog_manifest(). They describe image defaults; per-output timestamp settings belong on the context’s output methods. Register the image with MANIFEST; the runner rejects a different injected output format at startup.

Example

from pathlib import Path

from nominal.experimental.extractor import ManifestExtractorContext, input, manifest_extractor, parameter

@manifest_extractor(
    default_timestamp_column="time_us",
    default_timestamp_type="epoch_microseconds",
)
@input("source")
@parameter("parts", type=int, default=2)
def split(ctx: ManifestExtractorContext, *, source: Path, parts: int) -> None:
    table = read_parquet(source)
    for i, chunk in enumerate(chunks_of(table, parts)):
        out = ctx.output_dir / f"part_{i}.parquet"
        write_parquet(chunk, out)
        ctx.add_tabular(out, timestamp_column="time_us", timestamp_type="epoch_microseconds")

    footage = ctx.output_dir / "camera.mp4"
    write_video(footage)
    ctx.add_video(footage, channel="camera/front", start=recording_started_at)

if __name__ == "__main__":
    split.run()