IngestBuilder#

class nominal.experimental.ingest.IngestBuilder(client: NominalClient, dataset: str | Dataset, *, tags: Mapping[str, str] | None = None)#

Bases: object

Accumulate files and submit them as a single (MULTI) ingest job.

EXPERIMENTAL / UNSTABLE — see the module docstring. Targets an existing dataset; the v2 endpoint does not create datasets. Build with add_* (fluent), then submit() exactly once: a builder is single-use, and a second submit() raises rather than re-uploading and re-ingesting everything it holds.

Create a builder that targets an existing dataset.

Parameters:

  • client – Client used to upload files and trigger the ingest job.

  • dataset – The dataset to ingest into, as a Dataset or its RID. It must already exist; the v2 ingest endpoint does not create datasets.

  • tags – Request-level tags applied to every item in the job. Add more later with add_tags, or set per-item tags on the individual add_* calls.

add_ardupilot_dataflash(path: Path | str, *, tags: Mapping[str, str] | None = None) → Self#

Register an ArduPilot Dataflash (.bin) file, mirroring Dataset.add_ardupilot_dataflash.

Parameters:

  • path – Path to the Dataflash file on disk.

  • tags – Key-value pairs applied as tags to all data from this file.

Returns:

This builder, for chaining.

add_avro_stream(path: Path | str, *, units: Mapping[str, str] | None = None, channel_prefix: 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, tags: Mapping[str, str] | None = None) → Self#

Register an Avro stream (.avro) file.

The file must conform to the canonical Avro stream schema (see Dataset.add_avro_stream for the schema definition). The schema fixes which field holds the timestamps, so this takes no timestamp column – only how to read the numbers in that field.

Parameters:

  • path – Path to the .avro file on disk.

  • units – Mapping of channel name to unit symbol.

  • channel_prefix – Prefix prepended to every channel name ingested from this file.

  • timestamp_type – How to read the file’s numeric timestamps – an absolute epoch (ts.Epoch, or its string literal) or an offset from a start (ts.Relative). Defaults to epoch nanoseconds, the canonical schema’s reading.

  • tags – Key-value pairs applied as tags to all data from this file. Tags in the records take precedence – these only fill keys a record does not already set.

Returns:

This builder, for chaining.

add_containerized(extractor: str | ContainerizedExtractor, sources: Mapping[str, Path | str], *, arguments: Mapping[str, str] | None = None, tags: Mapping[str, str] | None = None) → Self#
add_containerized(extractor: str | ContainerizedExtractor, sources: Mapping[str, Path | str], *, arguments: Mapping[str, str] | None = None, timestamp_column: str, 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'], tags: Mapping[str, str] | None = None) → Self

Register a containerized-extractor run over one or more named source files.

Pass both timestamp_column and timestamp_type, or neither; the overloads make passing only one a type error, and the runtime guard below rejects it for callers without a type checker.

Parameters:

  • extractor – The containerized extractor to run, as a ContainerizedExtractor or its RID.

  • sources – Mapping of each registered extractor input name to a local file to upload. The names must match the extractor’s registered inputs exactly.

  • arguments – Key-value input arguments passed to the extractor.

  • timestamp_column – Column, applied uniformly to the extractor’s output files, holding each row’s timestamp.

  • timestamp_type – Type of the timestamp data in timestamp_column, e.g. ‘epoch_seconds’.

  • tags – Key-value pairs applied as tags to all data produced by this run.

Returns:

This builder, for chaining.

Raises:

ValueError – if sources is empty, or if only one of timestamp_column / timestamp_type is given.

add_csv(path: Path | str, timestamp_column: str, 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'], *, tag_columns: Mapping[str, str] | None = None, units: Mapping[str, str] | None = None, channel_prefix: str | None = None, channel_name_overrides: Mapping[str, str] | None = None, header_row: int | None = None, data_row: int | None = None, units_row: int | None = None, tags: Mapping[str, str] | None = None) → Self#

Register a CSV file (.csv or .csv.gz) with optional row selection.

All shared arguments follow add_tabular_data(). Explicit units override units read from units_row per channel; other channels retain their row units. The backend validates row numbers and their ordering when the job is submitted.

Parameters:

  • path – Path to a .csv or .csv.gz file.

  • timestamp_column – Column containing timestamps; not ingested as a data channel.

  • timestamp_type – Type of the timestamp data, e.g. ‘epoch_seconds’.

  • tag_columns – Mapping of tag keys to columns supplying their values.

  • units – Mapping of channel names to unit symbols, overriding the units record per channel.

  • channel_prefix – Prefix prepended to every ingested channel name.

  • channel_name_overrides – Mapping of original channel names to their ingested names.

  • tags – Key-value pairs applied as tags to all data from this file.

  • header_row – Positive, one-based header record number, defaulting to 1. Blank lines are ignored; a multiline record counts as one record.

  • data_row – Positive, one-based first data record, defaulting to header_row + 1. Must follow the header; intervening records are skipped.

  • units_row – Positive, one-based units record, distinct from the header and before the first data record. Set data_row past it. Unit cells match columns by position, including a placeholder for the timestamp column. Empty cells have no unit. Omit to read no units row.

Returns:

This builder, for chaining.

Raises:

ValueError – The path is not CSV.

add_journal_json(path: Path | str, *, channel: str | None = None, tags: Mapping[str, str] | None = None) → Self#
add_journal_json(path: Path | str, *, channel: 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'], tags: Mapping[str, str] | None = None) → Self

Register a journald-style .jsonl / .jsonl.gz log file.

Pass both timestamp_column and timestamp_type, or neither; the overloads make passing only one a type error, and the runtime guard below rejects it for callers without a type checker.

Parameters:

  • path – Path to the journal-json file on disk.

  • channel – Channel name to ingest the logs under. Defaults to ‘logs’ if omitted.

  • timestamp_column – Field holding each record’s timestamp. Omit to use the file’s default journald timestamp.

  • timestamp_type – How to read the numbers in timestamp_column – an absolute epoch (ts.Epoch, or its string literal, e.g. ‘epoch_microseconds’) or an offset from a start (ts.Relative). Log timestamps are read as numbers, so ISO 8601 and custom string formats cannot be used here.

  • tags – Key-value pairs applied as tags to all data from this file.

Returns:

This builder, for chaining.

Raises:

ValueError – if only one of timestamp_column / timestamp_type is given, or if timestamp_type is not numeric.

add_mcap(path: Path | str, *, include_topics: Sequence[str] | None = None, ignore_invalid_topics: bool | None = None, tags: Mapping[str, str] | None = None) → Self#
add_mcap(path: Path | str, *, exclude_topics: Sequence[str] | None = None, ignore_invalid_topics: bool | None = None, tags: Mapping[str, str] | None = None) → Self

Register an MCAP file.

Pass at most one of include_topics / exclude_topics; the overloads make passing both a type error, and the runtime guard below rejects it for callers without a type checker.

Parameters:

  • path – Path to the MCAP file on disk.

  • include_topics – If given, restrict ingestion to these topics. Defaults to all protobuf-encoded topics present in the MCAP.

  • exclude_topics – If given, ingest every topic except these.

  • ignore_invalid_topics – If true, skip invalid MCAP topics and continue ingesting valid ones.

  • tags – Key-value pairs applied as tags to all data from this file.

Returns:

This builder, for chaining.

Raises:

ValueError – if both include_topics and exclude_topics are given.

add_parquet(path: Path | str, timestamp_column: str, 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'], *, tag_columns: Mapping[str, str] | None = None, units: Mapping[str, str] | None = None, channel_prefix: str | None = None, channel_name_overrides: Mapping[str, str] | None = None, tags: Mapping[str, str] | None = None) → Self#

Register a Parquet file or archive, inferring compression and archive format from its extension.

Supports .parquet, .parquet.gz, .parquet.tar, .parquet.tar.gz, and .parquet.zip.

Parameters:

  • path – Path to a Parquet file or archive.

  • timestamp_column – Column containing timestamps; not ingested as a data channel.

  • timestamp_type – Type of the timestamp data, e.g. ‘epoch_seconds’.

  • tag_columns – Mapping of tag keys to columns supplying their values.

  • units – Mapping of channel names to unit symbols.

  • channel_prefix – Prefix prepended to every ingested channel name.

  • channel_name_overrides – Mapping of original channel names to their ingested names.

  • tags – Key-value pairs applied as tags to all data from this file.

Returns:

This builder, for chaining.

Raises:

ValueError – The path is not Parquet.

add_tabular_data(path: Path | str, timestamp_column: str, 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'], *, tag_columns: Mapping[str, str] | None = None, units: Mapping[str, str] | None = None, channel_prefix: str | None = None, channel_name_overrides: Mapping[str, str] | None = None, tags: Mapping[str, str] | None = None) → Self#

Register a tabular file by forwarding to add_csv() or add_parquet().

Supported extensions: .csv / .csv.gz, .parquet / .parquet.gz, and the parquet-archive formats (.parquet.tar / .parquet.tar.gz / .parquet.zip). The format is inferred from the extension. Use add_csv() for CSV-specific row options.

Parameters:

  • path – Path to the file on disk.

  • timestamp_column – Column containing the timestamp for each row. This column is not ingested as its own channel; it sets the timestamps for every other channel.

  • timestamp_type – Type of the timestamp data in timestamp_column, e.g. ‘epoch_seconds’.

  • tag_columns – Mapping of tag keys to the columns whose values supply each tag.

  • units – Mapping of channel name to unit symbol.

  • channel_prefix – Prefix prepended to every channel name ingested from this file.

  • channel_name_overrides – Mapping of original channel name to the name to ingest it under.

  • tags – Key-value pairs applied as tags to all data from this file.

Returns:

This builder, for chaining.

Raises:

ValueError – The path is not a supported tabular format.

add_tags(tags: Mapping[str, str]) → Self#

Add request-level tags applied to every item in the job.

Parameters:

tags – Key-value pairs to merge into the request-level tags.

Returns:

This builder, for chaining.

add_video(path: Path | str, channel: str, *, start: datetime | int, tags: Mapping[str, str] | None = None) → Self#
add_video(path: Path | str, channel: str, *, frame_timestamps: Sequence[int], tags: Mapping[str, str] | None = None) → Self

Register a video file.

Pass exactly one of start / frame_timestamps; the overloads make anything else a type error, and the runtime guard below rejects it for callers without a type checker. With start, frames are timestamped from that instant at the video’s own frame rate. With frame_timestamps, the per-frame timestamps are written to a manifest file that is uploaded alongside the video.

Parameters:

  • path – Path to the video file on disk.

  • channel – Channel name to ingest the video under.

  • start – Timestamp of the video’s first frame.

  • frame_timestamps – One epoch-nanosecond timestamp per video frame.

  • tags – Key-value pairs applied as tags to the video.

Returns:

This builder, for chaining.

Raises:

ValueError – if channel is empty, the path is not a supported video container, frame_timestamps is empty, or both or neither of start / frame_timestamps are given.

submit(*, allow_partial: bool = False, runs_to_expand: Sequence[Run | str] | None = None) → IngestionJob#

Upload all registered files and trigger one ingest job.

Uploads run in parallel, with transient failures (network weather, throttling) retried by the uploader’s own per-file budget. By default the call is atomic and fail-fast: the first file that fails permanently cancels the batch and raises, and no ingest is triggered. With allow_partial=True, every file runs to settlement instead — items whose files failed are logged and pruned, and one ingest job is triggered for the items that uploaded cleanly.

Single-use: one submit() consumes the builder, whether it succeeds or fails. A failed trigger request can have been committed server-side (a timeout, say), so there is no retry that cannot double-ingest — build a new builder instead.

Parameters:

  • allow_partial – If true, items whose files failed to upload are dropped from the job (each failure logged as an error) instead of failing the whole batch. Deliberately reported through logs, not the return value — the job always covers exactly the items that uploaded.

  • runs_to_expand – If provided, runs (or their rids) to expand upon successful ingest. This will only expand the bounds of the runs, not contract.

Returns:

The created ingest job. Track it by polling job.refresh().status, or block on its produced files with list(job.as_files_ingested()).

Raises: