ManifestExtractorContext#

class nominal.experimental.extractor.context.ManifestExtractorContext(output_dir: Path)#

Bases: ExtractorContext

Context for manifest_extractor() functions.

One declaration method per output format: add_tabular(), add_avro_stream(), add_journal_json(), add_video(). Each exposes only the options its format actually uses.

output_dir: Path#

property additional_tags: dict[str, str]#: Tags the ingest request applies to all data from this run; empty when Nominal didn’t inject them.

property dataset_rid: str | None#

RID of the dataset this run ingests into.

None when Nominal injected no value, empty included; see ingest_job_rid.

property ingest_job_rid: str | None#

RID of the ingest job running this extractor.

None when Nominal injected no value, empty included – an empty environment variable means the same thing as an absent one here, as it does everywhere else in this contract.

property inputs: list[Path]#

All input files Nominal mounted for this run.

Legacy authoring API: prefer a separate @input declaration per registered input. Access logs one migration warning per run; it contributes no registration metadata. Taken from the registered _NOMINAL_INPUTS metadata when present, in the order Nominal serializes them; otherwise discovered by listing the input mount, sorted by name.

property job_timestamp_metadata: TimestampMetadata | None#

The job-level timestamp metadata this run’s outputs default to.

This is the metadata the pipeline resolved for the whole job (the ingest request’s override when given, else the image’s registered default) – the value a manifest output falls back to when it declares no per-output timestamp metadata of its own. None when Nominal didn’t inject it.

add_avro_stream(path: str | PathLike[str], *, channel_prefix: str | None = None, units: Mapping[str, str] | None = None, timestamp_type: Epoch | Relative | Literal['epoch_picoseconds', 'epoch_nanoseconds', 'epoch_microseconds', 'epoch_milliseconds', 'epoch_seconds', 'epoch_minutes', 'epoch_hours', 'epoch_days'] | None = None) → Path#

Declare an avro-stream file you wrote (.avro or .avro.gz).

Avro records carry their own channel, values, and tags, so this takes no tag columns and no timestamp column. The schema fixes which field holds the timestamps; timestamp_type says how to read the numbers in it, overriding the job-level timestamp metadata for this output. Only numeric types work here: absolute epochs (ts.Epoch) or offsets from a starting time (ts.Relative).

Omit timestamp_type to inherit the job-level metadata. That is only correct when the job-level type is numeric too, since avro timestamps are integers and a string format cannot read them.

channel_prefix is prepended to every channel from this file. units maps channel names to unit symbols for this output. The mapping is copied when declared, and symbols are passed through unchanged.

add_journal_json(path: str | PathLike[str]) → Path#

add_journal_json(path: str | PathLike[str], *, timestamp_column: str, timestamp_type: Epoch | Relative | Literal['epoch_picoseconds', 'epoch_nanoseconds', 'epoch_microseconds', 'epoch_milliseconds', 'epoch_seconds', 'epoch_minutes', 'epoch_hours', 'epoch_days']) → Path

Declare a journal JSONL file you wrote (.jsonl or .jsonl.gz); it is ingested as logs.

Each line must carry a MESSAGE field and a timestamp field; lines missing either are skipped. Every other top-level field becomes a log arg, with its value converted to a string. timestamp_column/timestamp_type (provided together) name the top-level JSON field holding each line’s timestamp, overriding the job-level metadata; the same numeric-only restriction as add_tabular() applies.

Log samples carry no tags and every log point lands on one channel, so this takes neither tag columns nor a channel prefix – the ingest pipeline ignores both for log outputs. The overloads make passing only one of the pair a type error.

add_tabular(path: str | PathLike[str], *, tag_columns: Mapping[str, str] | None = None, channel_prefix: str | None = None, units: Mapping[str, str] | None = None) → Path#

add_tabular(path: str | PathLike[str], *, tag_columns: Mapping[str, str] | None = None, channel_prefix: str | None = None, units: Mapping[str, str] | None = None, timestamp_column: str, timestamp_type: Epoch | Relative | Literal['epoch_picoseconds', 'epoch_nanoseconds', 'epoch_microseconds', 'epoch_milliseconds', 'epoch_seconds', 'epoch_minutes', 'epoch_hours', 'epoch_days']) → Path

Declare a CSV or Parquet file you wrote; its columns become channels.

tag_columns maps tag names to the columns carrying their values. channel_prefix is prepended to every channel from this file. timestamp_column/timestamp_type (provided together) override the job-level timestamp metadata for this output, so each file can carry its own timestamp column; only numeric types work here – absolute epochs (ts.Epoch) or offsets from a starting time (ts.Relative). Outputs needing ISO 8601 or custom formats omit the pair and inherit the job-level metadata, which supports the full range. The overloads make passing only one of the pair a type error.

units maps channel names to unit symbols for this output. The mapping is copied when declared, and symbols are passed through unchanged.

add_video(path: str | PathLike[str], *, channel: str, frame_timestamps: Sequence[int]) → Path

Declare a video you wrote to the output directory; it becomes one video manifest entry.

The video is ingested as channel on the dataset this run writes to, alongside any telemetry outputs. Exactly one of start or frame_timestamps is required, and they fix the video’s absolute time two different ways:

start – the video’s absolute starting timestamp; each frame’s time comes from the video’s own encoded presentation timestamps, offset from that start. Pass one of ending_timestamp, true_frame_rate, or scale_factor (at most one) when the media plays at a different rate than the camera recorded at.
frame_timestamps – one absolute nanosecond timestamp per frame, when precise per-frame metadata is available. Unlike every other declaration method, this one writes: the runtime serializes the timestamps to a sidecar beside the video (cam.mp4 gets cam.mp4.timestamps.json) and declares it for you, so authors never have to reproduce that file’s format.

The video file itself must already exist under output_dir and carry a supported video extension.

Rejections come back as ExtractorError when they are about the extractor’s own contract – a reserved file name, a file outside the output directory – and as the argument errors the rest of the client raises (all ValueError subclasses) when the arguments themselves are malformed, which is what Dataset.add_video() does too.

Note

Video outputs require a recent version of the Nominal platform. An older ingest pipeline ignores them, and rejects a manifest whose only outputs are videos.

build_manifest() → dict[str, Any]#: Build the manifest document from the declared outputs, exactly as it is written to disk.

get_param(name: str, default: str | None = None) → str | None#

Read an optional parameter from the environment, or default when unset.

Legacy authoring API: prefer @parameter("parts", type=int, default=2) and a parts callback argument. This method logs one migration warning per run and contributes no registration metadata. Its returned values remain strings (or None). Name resolution matches param(). With registered contract metadata present, an unregistered name raises ExtractorError.

input(name: str | None = None) → Path#

Resolve an input file.

Legacy authoring API: prefer @input("source", envvar="SOURCE") and a source callback argument. This method logs one migration warning per run and contributes no registration metadata.

With name – the input’s registered display name or its environment variable – returns that input’s path. Without it, returns the sole mounted input file, raising if there is not exactly one.

param(name: str) → str#

Read a required parameter from the environment.

Legacy authoring API: prefer @parameter("parts", envvar="PARTS", type=int) and a parts callback argument. This method logs one migration warning per run and contributes no registration metadata.

name – the parameter’s registered display name or its environment variable – is resolved against _NOMINAL_PARAMETERS when Nominal injected it; otherwise it is treated directly as the environment variable. Raises ExtractorError when the parameter is not set. Values returned by this legacy method remain strings. With registered contract metadata present, an unregistered name raises ExtractorError.