Dataset#

class nominal.core.Dataset(rid: str, name: str, description: str | None, properties: NominalProperties, labels: Sequence[str], bounds: DatasetBounds | None, is_archived: bool)#

Bases: HasRid, MarkableMixin, RefreshableConjureMixin[EnrichedDataset]

A collection of channels populated by uploaded files and streamed data.

rid: str#

name: str#

description: str | None#

properties: NominalProperties#

labels: Sequence[str]#

bounds: DatasetBounds | None#

is_archived: bool#

property nominal_url: str#: Returns a URL to the page in the nominal app containing this dataset

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

Add a Dataflash file to an existing dataset.

Parameters:

path – Path to the Dataflash file to add to this dataset.
tags – key-value pairs to apply as tags to all data uniformly in the file.

add_ardupilot_dataflash_from_io(dataflash: BinaryIO, *, file_name: str | None = None, tags: Mapping[str, str] | None = None) → DatasetFile#

Add an ArduPilot Dataflash file from a binary stream.

Parameters:

dataflash – Binary stream consumed from its current position. Seeking is not required; the caller retains ownership and must close it.
file_name – Upload filename. The .bin extension is added only when the name has no extension. Defaults to this dataset’s name.
tags – Key-value pairs applied as tags to all data uniformly in the file.

Returns:

Dataset file after upload and ingestion submission. Call DatasetFile.poll_until_ingestion_completed to wait for ingestion.

Raises:

TypeError – The stream is open in text mode.

add_ardupilot_dataflash_to_dataset(path: Path | str, tags: Mapping[str, str] | None = None) → DatasetFile#

Add a Dataflash file to an existing dataset.

Parameters:

path – Path to the Dataflash file to add to this dataset.
tags – key-value pairs to apply as tags to all data uniformly in the file.

add_avro_stream(path: Path | str, *, 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) → DatasetFile#

Upload an avro-stream file, which must match the schema shown below.

This is a “stream-like” file format to support use cases where a columnar/tabular format does not make sense. This closely matches Nominal’s streaming API, making it useful for use cases where network connection drops during streaming and a backup file needs to be created.

Parameters:

path – Path to the .avro or .avro.gz file to upload
timestamp_type – How to read the numbers in the file’s timestamps field: an absolute epoch (ts.Epoch, or its string literal) or an offset from a start (ts.Relative). Defaults to “epoch_nanoseconds”.
tags – Key-value tags applied to all data in the file. Tags in the records themselves take precedence – these only fill keys a record does not already set.

Returns:

Reference to the ingesting DatasetFile

Raises:

ValueError – path does not end in .avro or .avro.gz, or timestamp_type has no numeric representation (e.g. an ISO 8601 or custom string format).

Note

For struct columns, values should be converted to JSON strings and wrapped in the JsonStruct record type.

Note

The previous schema with only “double” and “string” value types is still fully supported.

Note

If this schema is not used, ingestion will fail.

The required schema is:

{
    "type": "record",
    "name": "AvroStream",
    "namespace": "io.nominal.ingest",
    "fields": [
        {
            "name": "channel",
            "type": "string",
            "doc": "Channel/series name (e.g., 'vehicle_id', 'col_1', 'temperature')",
        },
        {
            "name": "timestamps",
            "type": {"type": "array", "items": "long"},
            "doc": "Array of numeric timestamps; see timestamp_type for how they are read",
        },
        {
            "name": "values",
            "type": {"type": "array", "items": [
                "double",
                "string",
                "long",
                {"type": "record", "name": "DoubleArray", "fields": [{"name": "items", "type": {"type": "array", "items": "double"}}]},
                {"type": "record", "name": "StringArray", "fields": [{"name": "items", "type": {"type": "array", "items": "string"}}]},
                {"type": "record", "name": "JsonStruct", "fields": [{"name": "json", "type": "string"}]}
            ]},
            "doc": "Array of values. Can be doubles, longs, strings, arrays, or JSON structs",
        },
        {
            "name": "tags",
            "type": {"type": "map", "values": "string"},
            "default": {},
            "doc": "Key-value metadata tags",
        },
    ],
}

add_channel(name: str, data_type: ChannelDataType, *, description: str | None = None, unit: str | None = None) → Channel#

Create a new channel (series metadata) for this data source.

This creates the channel metadata entry without uploading any data points. Use this to pre-register channels before streaming data to them.

Parameters:

name – The name of the channel to create.
data_type – The data type of the channel (e.g., DOUBLE, STRING, LOG, INT).
description – Optional human-readable description of the channel.
unit – Optional unit symbol to associate with the channel.

Returns:

The created or existing Channel object. Existing channels are returned unchanged.

Raises:

NominalError – Error creating the channel or fetching its metadata via gRPC.
ValueError – Channel metadata was not found after creation.

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

Add data from proprietary data formats using a pre-registered custom extractor.

A containerized extraction runs asynchronously and may emit multiple output files, so this returns the tracking IngestionJob immediately rather than a single file, and job.cancel() cancels it.

Waiting takes two stages: the container finishes producing files, then those files finish ingesting. job.as_files_ingested() covers both — it re-reads the file list while the job runs, so it picks up outputs that do not exist yet at the time of the call. job.dataset_files(), by contrast, returns only the files that exist when it is called, which is empty until a manifest extractor’s container has exited; and job.status is a snapshot, so polling it requires job.refresh() to advance it.

Parameters:

extractor – ContainerizedExtractor instance (or rid of one) to use for extracting and ingesting data.
sources – Mapping of environment variables to source files to use with the extractor.

Note: These must match the registered inputs of the active container image exactly.
arguments – Mapping of key-value pairs of input arguments to the extractor.
tags – Key-value pairs of tags to apply to all data ingested from the containerized extractor run.
timestamp_column – the column in the extractor’s output that contains the timestamp data. Provided together with timestamp_type, overrides the active container image’s default_timestamp_metadata for this ingest; if omitted, the image’s default is used. See ContainerImage.default_timestamp_metadata for the full resolution order.

Note: This is applied uniformly to all output files.

Must be provided with a timestamp_type or a ValueError will be raised.
timestamp_type – the type of timestamp data in the extractor’s output.

Note: This is applied uniformly to all output files.

Must be provided with a timestamp_column or a ValueError will be raised.

Returns:

An IngestionJob handle for the asynchronous containerized ingest.

add_from_io(dataset: BinaryIO, 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'], file_type: tuple[str, str] | FileType = FileTypes.CSV, file_name: str | None = None, tag_columns: Mapping[str, str] | None = None, tags: Mapping[str, str] | None = None) → DatasetFile#

Append to a dataset from a file-like object.

Parameters:

dataset – a file-like object containing the data to append to the dataset.
timestamp_column – the column in the dataset that contains the timestamp data.
timestamp_type – the type of timestamp data in the dataset.
file_type – a (extension, mimetype) pair describing the type of file.
file_name – Upload filename. An existing extension is preserved; otherwise the selected type supplies the default extension. Defaults to this dataset’s name.
tag_columns – a dictionary mapping tag keys to column names.
tags – key-value pairs to apply as tags to all data uniformly in the file

add_journal_json(path: Path | str, *, channel: str | None = None) → DatasetFile#

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

Add a journald jsonl file to an existing dataset.

Supports any journald-style jsonl file, with a few expectations:

The file must be .jsonl or .jsonl.gz, and each line must be a json object.
Each line must hold a timestamp field and a MESSAGE field carrying the primary log message. The timestamp field is __REALTIME_TIMESTAMP unless timestamp_column names another one. Lines missing either field are skipped.
Every other top-level field becomes a log arg, with its value converted to a string.

Parameters:

path – Path to journal json file to ingest
channel – Optionally, a channel name to use for ingested logs (defaults to ‘logs’)
timestamp_column – Optionally, the name of the field in each json payload holding a timestamp. Defaults to __REALTIME_TIMESTAMP.
timestamp_type – Optionally, how to read the numbers in that field, given as a ts.Epoch or its string literal, e.g. "epoch_microseconds". Defaults to microseconds since the unix epoch.

Returns:

DatasetFile object which can be used to poll for ingestion completion

Raises:

ValueError – if only one of timestamp_column / timestamp_type is given, or if timestamp_type is not an absolute epoch type.

add_journal_json_from_io(journal: BinaryIO, *, file_type: tuple[str, str] | FileType | None = None, file_name: str | None = None, channel: str | None = None) → DatasetFile#

add_journal_json_from_io(journal: BinaryIO, *, file_type: tuple[str, str] | FileType | None = None, file_name: str | None = None, channel: str | None = None, timestamp_column: str, timestamp_type: Epoch | Literal['epoch_picoseconds', 'epoch_nanoseconds', 'epoch_microseconds', 'epoch_milliseconds', 'epoch_seconds', 'epoch_minutes', 'epoch_hours', 'epoch_days']) → DatasetFile

Add journald-style JSONL logs from a binary stream.

Parameters:

journal – Binary stream consumed from its current position. Seeking is not required; the caller retains ownership and must close it.
file_type – FileTypes.JOURNAL_JSONL or FileTypes.JOURNAL_JSONL_GZ, supplied as a FileType or an (extension, mimetype) pair. When omitted or None, infer from file_name. Without file_name, supply an explicit type. Compressed bytes must be supplied for the gzip format; contents are not inspected.
file_name – Upload filename. An existing extension is preserved; otherwise the selected type supplies the default extension. Defaults to this dataset’s name.
channel – Log channel name. Defaults to ‘logs’ when omitted.
timestamp_column – Timestamp field. Defaults to __REALTIME_TIMESTAMP when omitted.
timestamp_type – Absolute epoch unit for that field. Defaults to epoch microseconds when omitted. Pass this together with timestamp_column.

Returns:

Dataset file after upload and ingestion submission. Call DatasetFile.poll_until_ingestion_completed to wait for ingestion.

Raises:

TypeError – The stream is open in text mode.
ValueError – The format cannot be inferred or is not journal JSONL, only one timestamp option is supplied, or the timestamp type is not an absolute epoch type.

Note

Records must contain a timestamp and a MESSAGE field, as described by add_journal_json.

add_journal_json_to_dataset(path: Path | str, *, channel: str | None = None, timestamp_column: str | None = None, timestamp_type: Epoch | Literal['epoch_picoseconds', 'epoch_nanoseconds', 'epoch_microseconds', 'epoch_milliseconds', 'epoch_seconds', 'epoch_minutes', 'epoch_hours', 'epoch_days'] | None = None) → DatasetFile#

Add a journald jsonl file to an existing dataset.

Supports any journald-style jsonl file, with a few expectations:

The file must be .jsonl or .jsonl.gz, and each line must be a json object.
Each line must hold a timestamp field and a MESSAGE field carrying the primary log message. The timestamp field is __REALTIME_TIMESTAMP unless timestamp_column names another one. Lines missing either field are skipped.
Every other top-level field becomes a log arg, with its value converted to a string.

Parameters:

path – Path to journal json file to ingest
channel – Optionally, a channel name to use for ingested logs (defaults to ‘logs’)
timestamp_column – Optionally, the name of the field in each json payload holding a timestamp. Defaults to __REALTIME_TIMESTAMP.
timestamp_type – Optionally, how to read the numbers in that field, given as a ts.Epoch or its string literal, e.g. "epoch_microseconds". Defaults to microseconds since the unix epoch.

Returns:

DatasetFile object which can be used to poll for ingestion completion

Raises:

ValueError – if only one of timestamp_column / timestamp_type is given, or if timestamp_type is not an absolute epoch type.

Add an MCAP file to an existing dataset.

Parameters:

path – Path to the MCAP file to add to this dataset
include_topics – If present, list of topics to restrict ingestion to. If not present, defaults to all protobuf-encoded topics present in the MCAP.
exclude_topics – If present, list of topics to not ingest from the MCAP.
tags – key-value pairs to apply as tags to all data uniformly in the file.
ignore_invalid_topics – If true, ignore invalid MCAP topics and continue ingesting valid topics.

add_mcap_from_io(mcap: BinaryIO, include_topics: Iterable[str] | None = None, exclude_topics: Iterable[str] | None = None, file_name: str | None = None, *, tags: Mapping[str, str] | None = None, ignore_invalid_topics: bool | None = None) → DatasetFile#

Add data to this dataset from an MCAP file-like object.

The mcap must be a file-like object in binary mode, e.g. open(path, “rb”) or io.BytesIO. If the file is not in binary-mode, the requests library blocks indefinitely.

Parameters:

mcap – Binary file-like MCAP stream
include_topics – If present, list of topics to restrict ingestion to. If not present, defaults to all protobuf-encoded topics present in the MCAP.
exclude_topics – If present, list of topics to not ingest from the MCAP.
file_name – Upload filename. The .mcap extension is added only when the name has no extension. Defaults to this dataset’s name.
tags – key-value pairs to apply as tags to all data uniformly in the file.
ignore_invalid_topics – If true, ignore invalid MCAP topics and continue ingesting valid topics.

Add an MCAP file to an existing dataset.

Parameters:

path – Path to the MCAP file to add to this dataset
include_topics – If present, list of topics to restrict ingestion to. If not present, defaults to all protobuf-encoded topics present in the MCAP.
exclude_topics – If present, list of topics to not ingest from the MCAP.
tags – key-value pairs to apply as tags to all data uniformly in the file.
ignore_invalid_topics – If true, ignore invalid MCAP topics and continue ingesting valid topics.

add_mcap_to_dataset_from_io(mcap: BinaryIO, include_topics: Iterable[str] | None = None, exclude_topics: Iterable[str] | None = None, file_name: str | None = None, *, tags: Mapping[str, str] | None = None, ignore_invalid_topics: bool | None = None) → DatasetFile#

Add data to this dataset from an MCAP file-like object.

The mcap must be a file-like object in binary mode, e.g. open(path, “rb”) or io.BytesIO. If the file is not in binary-mode, the requests library blocks indefinitely.

Parameters:

mcap – Binary file-like MCAP stream
include_topics – If present, list of topics to restrict ingestion to. If not present, defaults to all protobuf-encoded topics present in the MCAP.
exclude_topics – If present, list of topics to not ingest from the MCAP.
file_name – Upload filename. The .mcap extension is added only when the name has no extension. Defaults to this dataset’s name.
tags – key-value pairs to apply as tags to all data uniformly in the file.
ignore_invalid_topics – If true, ignore invalid MCAP topics and continue ingesting valid topics.

add_mcap_video(path: Path | str, *, channel: str, topic: str, tags: Mapping[str, str] | None = None, overwrite_overlapping: bool = False) → VideoDatasetFile#

Upload video data from an MCAP file to this dataset as a channel.

Parameters:

path – Path to the MCAP file (must end in mcap) containing H264/H265 video data.
channel – Name of the video channel within this dataset that the file’s frames belong to.
topic – MCAP topic containing the video data; per-frame timestamps come from this topic’s messages.
tags – key-value pairs to apply as tags to all data uniformly in the file.
overwrite_overlapping – If True, segments from other files on this channel that overlap with the newly added file will be deleted before inserting the new segments.

Returns:

Reference to the created (still-ingesting) video dataset file.

Raises:

ValueError – path does not end in mcap.

add_mcap_video_from_io(mcap: BinaryIO, file_name: str, *, channel: str, topic: str, file_type: tuple[str, str] | FileType = FileTypes.MCAP, tags: Mapping[str, str] | None = None, overwrite_overlapping: bool = False) → VideoDatasetFile#

Upload video data from a binary MCAP file-like object to this dataset as a channel.

The mcap must be a file-like object in binary mode, e.g. open(path, “rb”) or io.BytesIO.

Parameters:

mcap – File-like binary object containing MCAP data with H264/H265 video.
file_name – Upload filename. An existing extension is preserved; otherwise the selected type supplies the default.
channel – Name of the video channel within this dataset that the file’s frames belong to.
topic – MCAP topic containing the video data; per-frame timestamps come from this topic’s messages.
file_type – Metadata about the type of file (e.g. MCAP).
tags – key-value pairs to apply as tags to all data uniformly in the file.
overwrite_overlapping – If True, segments from other files on this channel that overlap with the newly added file will be deleted before inserting the new segments.

Returns:

Reference to the created (still-ingesting) video dataset file.

Raises:

TypeError – mcap is open in text mode rather than binary mode.

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, tags: Mapping[str, str] | None = None) → DatasetFile#

Append to a dataset from tabular data on-disk.

Currently, the supported filetypes are:

.csv / .csv.gz
.parquet / .parquet.gz
.parquet.tar / .parquet.tar.gz / .parquet.zip

Parameters:

path – Path to the file on disk to add to the dataset.
timestamp_column – Column within the file containing timestamp information.

Note: This is omitted as a channel from the data added to Nominal, and is instead used to set the timestamps for all other uploaded data channels.
timestamp_type – Type of timestamp data contained within the timestamp_column e.g. ‘epoch_seconds’.
tag_columns – a dictionary mapping tag keys to column names.
tags – key-value pairs to apply as tags to all data uniformly in the file

add_tabular_data_to_dataset(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, tags: Mapping[str, str] | None = None) → DatasetFile#

Append to a dataset from tabular data on-disk.

Currently, the supported filetypes are:

.csv / .csv.gz
.parquet / .parquet.gz
.parquet.tar / .parquet.tar.gz / .parquet.zip

Parameters:

path – Path to the file on disk to add to the dataset.
timestamp_column – Column within the file containing timestamp information.

Note: This is omitted as a channel from the data added to Nominal, and is instead used to set the timestamps for all other uploaded data channels.
timestamp_type – Type of timestamp data contained within the timestamp_column e.g. ‘epoch_seconds’.
tag_columns – a dictionary mapping tag keys to column names.
tags – key-value pairs to apply as tags to all data uniformly in the file

add_to_dataset_from_io(dataset: BinaryIO, 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'], file_type: tuple[str, str] | FileType = FileTypes.CSV, file_name: str | None = None, tag_columns: Mapping[str, str] | None = None, tags: Mapping[str, str] | None = None) → DatasetFile#

Append to a dataset from a file-like object.

Parameters:

dataset – a file-like object containing the data to append to the dataset.
timestamp_column – the column in the dataset that contains the timestamp data.
timestamp_type – the type of timestamp data in the dataset.
file_type – a (extension, mimetype) pair describing the type of file.
file_name – Upload filename. An existing extension is preserved; otherwise the selected type supplies the default extension. Defaults to this dataset’s name.
tag_columns – a dictionary mapping tag keys to column names.
tags – key-value pairs to apply as tags to all data uniformly in the file

add_video(path: Path | str, *, channel: str, start: datetime | int, tags: Mapping[str, str] | None = None, overwrite_overlapping: bool = False) → VideoDatasetFile#

add_video(path: Path | str, *, channel: str, frame_timestamps: Sequence[int], tags: Mapping[str, str] | None = None, overwrite_overlapping: bool = False) → VideoDatasetFile

Upload a video file to this dataset as a channel.

Parameters:

path – Path to the H264/H265-encoded video file to add to this dataset.
channel – Name of the video channel within this dataset that the file’s frames belong to.
start – Starting timestamp of the video file in absolute UTC time (datetime or epoch nanoseconds). Exactly one of start or frame_timestamps must be provided.
frame_timestamps – Per-frame absolute nanosecond timestamps. Most usecases should instead use the ‘start’ parameter, unless precise per-frame metadata is available and desired.
tags – key-value pairs to apply as tags to all data uniformly in the file.
overwrite_overlapping – If True, segments from other files on this channel that overlap with the newly added file will be deleted before inserting the new segments.

Returns:

Reference to the created (still-ingesting) video dataset file.

Raises:

ValueError – neither or both of start and frame_timestamps were provided.

add_video_from_io(video: BinaryIO, file_name: str, *, channel: str, start: datetime | int, file_type: tuple[str, str] | FileType = FileTypes.MP4, tags: Mapping[str, str] | None = None, overwrite_overlapping: bool = False) → VideoDatasetFile#

add_video_from_io(video: BinaryIO, file_name: str, *, channel: str, frame_timestamps: Sequence[int], file_type: tuple[str, str] | FileType = FileTypes.MP4, tags: Mapping[str, str] | None = None, overwrite_overlapping: bool = False) → VideoDatasetFile

Upload video data from a binary file-like object to this dataset as a channel.

The video must be a file-like object in binary mode, e.g. open(path, “rb”) or io.BytesIO, containing H264 or H265-encoded video data.

Parameters:

video – File-like object containing video data encoded in H264 or H265.
file_name – Upload filename. An existing extension is preserved; otherwise the selected type supplies the default.
channel – Name of the video channel within this dataset that the file’s frames belong to.
start – Starting timestamp of the video file in absolute UTC time (datetime or epoch nanoseconds). Exactly one of start or frame_timestamps must be provided.
frame_timestamps – Per-frame absolute nanosecond timestamps. Most usecases should instead use the ‘start’ parameter, unless precise per-frame metadata is available and desired.
file_type – Metadata about the type of video file, e.g., MP4 vs. MKV.
tags – key-value pairs to apply as tags to all data uniformly in the file.
overwrite_overlapping – If True, segments from other files on this channel that overlap with the newly added file will be deleted before inserting the new segments.

Returns:

Reference to the created (still-ingesting) video dataset file.

Raises:

TypeError – video is open in text mode rather than binary mode.
ValueError – neither or both of start and frame_timestamps were provided.

apply_markings(markings: Iterable[Marking | str]) → None#

Apply markings to this resource, leaving any already-applied markings in place.

Applying a marking that is already applied is a no-op rather than an error.

Parameters:: markings – Markings, or marking RIDs, to apply.
Raises:: NominalError – If the request fails, including when a marking is archived or the user lacks permission to change markings on this resource.

archive() → None#: Archive this dataset. Archived datasets are not deleted, but are hidden from the UI.

Note

This does not update the instance in place; call refresh() to see the change reflected.

batch_add_channels(channels: Sequence[CreateChannelRequest]) → BatchAddChannelsResult#

Create multiple channels (series metadata) for this data source in batches.

Parameters:

channels – Sequence of CreateChannelRequest objects.

Returns:

A BatchAddChannelsResult with the created channels and any requests that were not found after creation (i.e. silently dropped by the server).

Raises:

ValueError – A channel name appears more than once in channels.
NominalError – Error creating channels or fetching their metadata via gRPC.

Note

Existing channels are returned unchanged. Repeated names anywhere in channels raise ValueError before any requests are sent, even if their metadata matches.

Batches are written sequentially; earlier batches remain applied if a later batch fails.

batch_update_or_create_channels(channels: Sequence[CreateChannelRequest]) → BatchAddChannelsResult#

Create or update multiple channels (series metadata) for this data source in batches.

For each request, creates the channel if it does not already exist. If it does exist, updates the locator and any provided unit and description fields while preserving existing values for fields left as None.

Parameters:: channels – Sequence of CreateChannelRequest objects.
Returns:: A BatchAddChannelsResult with the upserted channels and any requests that were not found afterwards (i.e. silently dropped by the server).

Note

If the same channel name appears more than once within a single batch, the entire batch will fail. Callers are responsible for deduplicating channel names before passing them to this method. Earlier batches remain applied if a later batch fails.

get_channel(name: str) → Channel#

get_channels(*, names: Iterable[str] | None = None) → Iterable[Channel]#

Look up the metadata for all matching channels associated with this datasource

Parameters:: names – List of channel names to look up metadata for.
Yields:: Yields a sequence of channel metadata objects which match the provided query parameters

get_dataset_file(dataset_file_id: str) → DatasetFile#

Retrieve the given dataset file by ID

Parameters:: dataset_file_id – ID of the file to retrieve from the dataset
Returns:: Metadata for the requested dataset file
Raises:: FileNotFoundError – Details about the requested file could not be found

get_log_stream(batch_size: int = 50_000, max_wait: timedelta = timedelta(seconds=1)) → WriteStreamBase[str]#

Stream to asynchronously write log data to a dataset.

Parameters:

batch_size – Number of records to upload at a time to Nominal.

Note: Raising this may improve performance in high latency scenarios.
max_wait – Maximum number of seconds to allow data to be locally buffered before streaming to Nominal.

Returns:

Write stream object configured to send logs to nominal. This may be used as a context manager (so that resources are automatically released upon exiting the context), or if not used as a context manager, should be explicitly close()-ed once no longer needed.

get_video_file(dataset_file_id: str) → VideoDatasetFile#

Retrieve a video dataset file by ID.

Raises:

FileNotFoundError – the file does not exist in this dataset.
TypeError – the ID identifies a non-video dataset file.

get_write_stream(batch_size: int = 250000, max_wait: timedelta = timedelta(seconds=0.25), implementation: Literal['rust', 'rust_experimental'] | None = None, file_fallback: PathLike | None = None, log_level: str | None = None, num_workers: int | None = None, track_metrics: bool = False) → DataStream#

get_write_stream(batch_size: int = 250000, max_wait: timedelta = timedelta(seconds=0.25), implementation: Literal['python', 'json', 'protobuf', 'experimental'] = None) → DataStream

get_write_stream(batch_size: int = 250000, max_wait: timedelta = timedelta(seconds=0.25), *, data_format: Literal['rust', 'rust_experimental'], file_fallback: PathLike | None = None, log_level: str | None = None, num_workers: int | None = None, track_metrics: bool = False) → DataStream

get_write_stream(batch_size: int = 250000, max_wait: timedelta = timedelta(seconds=0.25), *, data_format: Literal['python', 'json', 'protobuf', 'experimental']) → DataStream

Stream to write timeseries data to a datasource.

Data is written asynchronously.

Parameters:

batch_size – How big the batch can get before writing to Nominal.
max_wait – How long a batch can exist before being flushed to Nominal.
implementation – Streaming implementation to use: ‘rust’ or ‘python’. When omitted, uses ‘rust’ if nominal-streaming is available, otherwise ‘python’. Automatic selection cannot fall back to ‘python’ when file_fallback, log_level, or num_workers is supplied, or when track_metrics=True.

Note: ‘json’, ‘protobuf’, and ‘rust_experimental’ are deprecated spellings of ‘python’, ‘python’, and ‘rust’ respectively.

Note: ‘experimental’ is deprecated and retains its Python metrics implementation. Use implementation='rust' with track_metrics=True to migrate.
file_fallback – Filepath to write failed batches to during streaming.

Note: Expects a .avro filename.

Note: Supported only by the Rust backend.
log_level – Log level to use in underlying rust streaming code.

Note: Should be a rust log level e.g. ‘debug’, ‘trace’, ‘info’, etc.

Note: Supported only by the Rust backend.
num_workers – Number of Rust upload workers and runtime worker threads.

Note: Use with care– this may have large impacts on streaming performance.

Note: Supported only by the Rust backend.
track_metrics – Emit request round-trip time and point staleness in seconds as __nominal.metric.* channels on the same dataset. Staleness is the SDK machine’s wall-clock UTC time minus the point timestamp; future timestamps produce negative values. Request metrics measure the oldest and newest points immediately before and after a successful upload. enqueue_from_dict also emits enque_dict_start_staleness and enque_dict_end_staleness, measured at the start and end of the enqueue. Disabled by default.

Note: Request metrics are best-effort and travel with later data uploads. Pending request metrics are not sent separately on close and may be dropped if their buffer fills.

Note: Supported only by the Rust backend.
data_format – Deprecated name for implementation. Passing both is an error.

Returns:

Write stream object configured to send data to nominal. This may be used as a context manager (so that resources are automatically released upon exiting the context), or if not used as a context manager, should be explicitly close()-ed once no longer needed.

Raises:

ValueError – both implementation and data_format were given, or the implementation is unknown.
ImportError – nominal-streaming is not installed and rust streaming was asked for, either explicitly via implementation or implicitly by passing a rust-only argument.

list_files(*, successful_only: bool = True) → Iterable[DatasetFile]#

List files ingested to this dataset.

If successful_only, yields files with a ‘success’ ingest status only.

list_markings() → Sequence[Marking]#

The markings currently applied to this resource.

Markings the user does not have permission to read are omitted.

Note

The service filters unreadable markings out of the response, so this may be a strict subset of what is actually applied. Treat it as “what I can see”, not “what is there”: a resource can carry markings that never appear here. Use apply_markings and remove_markings, which name their targets outright, rather than computing a desired set from this list.

Raises:: NominalError – If the request fails.

list_video_files(*, successful_only: bool = True) → Iterable[VideoDatasetFile]#

List video files ingested to this dataset.

If successful_only, yields successfully-ingested video files only; otherwise also yields queued, ingesting, failed, and deletion-state video files.

refresh() → Self#

remove_markings(markings: Iterable[Marking | str]) → None#

Remove markings from this resource.

Removing a marking that is not applied is a no-op rather than an error.

Parameters:: markings – Markings, or marking RIDs, to remove.
Raises:: NominalError – If the request fails.

search_channels(exact_match: Sequence[str] = (), fuzzy_search_text: str = '', *, data_types: Sequence[ChannelDataType] | None = None) → Iterable[Channel]#

Look up channels associated with a datasource.

Parameters:

exact_match – Filter the returned channels to those whose names match all provided strings (case insensitive).
fuzzy_search_text – Filters the returned channels to those whose names fuzzily match the provided string.
data_types – Filter the returned channels to those that match any of the provided types

Yields:

Channel objects for each matching channel

Search for files within this dataset. Filters are ANDed together. Time range uses overlap semantics: a file is included if its time range overlaps with [start, end], i.e. file.end >= start AND file.start <= end.

Parameters:

start – Inclusive lower bound of the search window. Files whose time range ends at or after this timestamp are returned — including files that started before start but still overlap the window. Files ending entirely before start are excluded.

Note: Truncated to whole seconds — sub-second precision is dropped.
end – Inclusive upper bound of the search window. Files whose time range starts at or before this timestamp are returned — including files that end after end but still overlap the window. Files starting entirely after end are excluded.

Note: Truncated to whole seconds — sub-second precision is dropped.
file_tags – A mapping of key-value tag pairs that must ALL be present on a dataset file to be included.

Returns:

All dataset files within this dataset which match all of the provided conditions

set_channel_prefix_tree(delimiter: str = '.') → None#

Index channels hierarchically by a given delimiter.

Primarily, the result of this operation is to prompt the frontend to represent channels in a tree-like manner that allows folding channels by common roots.

set_channel_units(channels_to_units: Mapping[str, Unit | str | None], validate_schema: bool = False, allow_display_only_units: bool = False) → None#

Set units for channels based on a provided mapping of channel names to units.

Updates are sent sequentially in batches of at most 10,000 matching channels. Each batch is atomic. If a later batch fails, earlier batches remain applied and subsequent batches are not sent.

Parameters:

channels_to_units – A mapping of channel names to unit symbols.

Note: Any existing units may be cleared from a channel by providing None as a symbol.
validate_schema – If true, raises a ValueError if non-existent channel names are provided in channels_to_units. Default is False.
allow_display_only_units – If true, allow units that would be treated as display-only by Nominal.

Raises:

ValueError – Unsupported unit symbol provided
NominalError – Error validating or updating units via gRPC (e.g. NominalNotFoundError).
conjure_python_client.ConjureHTTPError – Error completing requests.

unarchive() → None#: Unarchives this dataset, allowing it to show up in the ‘All Datasets’ pane in the UI.

Note

This does not update the instance in place; call refresh() to see the change reflected.

Replace dataset metadata. Updates the current instance, and returns it.

Only the metadata passed in will be replaced, the rest will remain untouched.

Parameters:

name – Replacement name.
description – Replacement description.
properties – String or numeric properties. Integers are stored as floats. None leaves properties unchanged; an empty mapping clears them.
labels – Replacement labels.

Note

This replaces the metadata rather than appending it. To append to labels or properties, merge them before calling this method. E.g.:

new_labels = ["new-label-a", "new-label-b"]
for old_label in dataset.labels:
    new_labels.append(old_label)
dataset = dataset.update(labels=new_labels)

Raises:: TypeError – If a property value is a boolean or is not a string, integer, or float.

update_bounds(*, start: str | datetime | int, end: str | datetime | int) → Self#

Update the bounds (start and end timestamps) of the dataset. Updates the current instance, and returns it.

Parameters:

start – The start timestamp of the dataset bounds. Can be a datetime, ISO 8601 string, or integer nanoseconds since epoch.
end – The end timestamp of the dataset bounds. Can be a datetime, ISO 8601 string, or integer nanoseconds since epoch.

Returns:

The updated Dataset instance with new bounds.

write_logs(logs: Iterable[LogPoint], channel_name: str = 'logs', batch_size: int = 1000) → None#

Stream logs to the datasource.

This method executes synchronously, i.e. it blocks until all logs are sent to the API. Logs are sent in batches. The logs can be any iterable of LogPoints, including a generator.

Parameters:

logs – LogPoints to stream to Nominal.
channel_name – Name of the channel to stream logs to.
batch_size – Number of logs to send to the API at a time.

Example

from typing import Iterable

from nominal.core import LogPoint

def parse_logs_from_file(file_path: str) -> Iterable[LogPoint]:
    # 2025-04-08T14:26:28.679052Z [INFO] Sent ACTUATE_MOTOR command
    with open(file_path, "r") as f:
        for line in f:
            timestamp, level, message = line.removesuffix("\n").split(maxsplit=2)
            yield LogPoint.create(timestamp, message, {"level": level.removeprefix("[").removesuffix("]")})

dataset = client.get_dataset("dataset_rid")
logs = parse_logs_from_file("logs.txt")
dataset.write_logs(logs)