`ncore.data.v4` Package#

Package exposing methods related to NCore’s V4 data interaction APIs

class ncore.data.v4.CameraLabelsComponent#

Bases: object

Data component for storing per-camera image-aligned labels (depth, segmentation, flow, etc.).

COMPONENT_NAME: str = 'camera_labels'#

class Reader( component_instance_name: str, component_group: Group, )#

Bases: ComponentReader

Camera-labels component reader.

class CameraLabelHandle( label_group: nodes.Group, descriptor: types.CameraLabelDescriptor, timestamp_us: int, generic_meta_data: JsonDocument, )#

Bases: object

References label data without eagerly loading it.

Implements the CameraLabelHandleProtocol protocol, providing access to the label data, schema, timestamp, and per-label metadata.

property descriptor: CameraLabelDescriptor#

property generic_meta_data: JsonDocument#

get_data() → npt.NDArray[Any]#

Load and return the label data as a numpy array.

For RAW encoding, applies de-quantization if specified in the schema. For IMAGE_ENCODED encoding, decodes the image bytes via PIL.

async get_data_async() → npt.NDArray[Any]#: Asynchronous get_data()

get_encoded_data() → bytes | None#: Return the raw encoded bytes for IMAGE_ENCODED labels, or None for RAW.

async get_encoded_data_async() → bytes | None#: Asynchronous get_encoded_data()

property timestamp_us: int#

static get_component_name() → str#: Returns the base name of the current component

get_label( timestamp_us: int, ) → CameraLabelHandle#: Return a lazy handle to the label data at the given timestamp.

property label_descriptor: CameraLabelDescriptor#

property labels_count: int#

static supports_component_version( version: str, ) → bool#: Returns true if the component version is supported by the reader

property timestamps_us: npt.NDArray[np.uint64]#

class Writer( component_group: Group, sequence_timestamp_interval_us: HalfClosedInterval, descriptor: CameraLabelDescriptor, )#

Bases: ComponentWriter

Camera-labels component writer.

finalize() → None#: Write sorted timestamps_us array.

static get_component_name() → str#: Returns the base name of the component writer

static get_component_version() → str#: Returns the version of the current component writer

store_label( data: npt.NDArray[Any] | bytes, timestamp_us: int, generic_meta_data: JsonDocumentLike | None = None, ) → None#

Store a single camera label.

Parameters:

data – For RAW encoding: a numpy array of shape (H, W) or (H, W, *schema.shape_suffix) in the logical dtype (schema.dtype). When quantization is configured, the data is automatically quantized to the on-disk integer representation. For IMAGE_ENCODED encoding: raw image bytes in the schema.encoded_format.
timestamp_us – Timestamp in microseconds - must fall within the sequence interval.
generic_meta_data – Optional per-label metadata.

class ncore.data.v4.CameraSensorComponent#

Bases: object

Camera sensor data component

COMPONENT_NAME: str = 'cameras'#

class Reader( component_instance_name: str, component_group: Group, )#

Bases: BaseSensorComponentReader

Camera sensor data component reader

class EncodedImageDataHandle( image_array: Array, )#

Bases: object

References encoded image data without loading it

get_data() → EncodedImageData#: Loads the referenced encoded image data to memory

async get_data_async() → EncodedImageData#: Asynchronous get_data()

static get_component_name() → str#: Returns the base name of the current component

get_frame_data( timestamp_us: int, ) → EncodedImageData#: Returns the frame’s encoded image data

async get_frame_data_async( timestamp_us: int, ) → EncodedImageData#: Asynchronous get_frame_data()

get_frame_handle( timestamp_us: int, ) → EncodedImageDataHandle#: Returns the frame’s encoded image data

get_frame_image( timestamp_us: int, ) → Image#: Returns the frame’s decoded image data

static supports_component_version( version: str, ) → bool#: Returns true if the component version is supported by the reader

class Writer( component_group: Group, sequence_timestamp_interval_us: HalfClosedInterval, )#

Bases: BaseSensorComponentWriter

Camera sensor data component writer

static get_component_name() → str#: Returns the base name of the current camera sensor component

static get_component_version() → str#: Returns the version of the current camera sensor component

store_frame( image_binary_data: bytes, image_format: str, frame_timestamps_us: npt.NDArray[np.uint64], generic_data: Dict[str, npt.NDArray[Any]], generic_meta_data: JsonDocumentLike, ) → Self#

class ncore.data.v4.ComponentGroupAssignments( camera_component_groups: Dict[str, str], lidar_component_groups: Dict[str, str], radar_component_groups: Dict[str, str], point_clouds_component_groups: Dict[str, str], camera_labels_component_groups: Dict[str, str], poses_component_group: str | None, intrinsics_component_group: str | None, masks_component_group: str | None, cuboid_track_observations_component_group: str | None, )#

Bases: object

Component group assignments for all default components in a V4 data output

camera_component_groups: Dict[str, str]#

camera_labels_component_groups: Dict[str, str]#

static create( camera_ids: List[str], lidar_ids: List[str], radar_ids: List[str], point_clouds_ids: List[str], camera_labels_ids: List[str], profile: Literal['default', 'separate-sensors', 'separate-all'], poses_component_group: str | None = None, intrinsics_component_group: str | None = None, masks_component_group: str | None = None, camera_component_groups: Dict[str, str] | None = None, lidar_component_groups: Dict[str, str] | None = None, radar_component_groups: Dict[str, str] | None = None, point_clouds_component_groups: Dict[str, str] | None = None, camera_labels_component_groups: Dict[str, str] | None = None, cuboid_track_observations_component_group: str | None = None, ) → ComponentGroupAssignments#

Factory function to create ComponentGroups based on a profile.

Parameters:

camera_ids – IDs of camera sensors
lidar_ids – IDs of lidar sensors
radar_ids – IDs of radar sensors
point_clouds_ids – IDs of native point cloud sources
camera_labels_ids – IDs of camera label instances
profile – One of: - “default”: Use provided overrides or fall back to default groups - “separate-sensors”: Each sensor gets its own group named “<sensor_id>” unless overwritten, remaining components use default store - “separate-all”: Each component type gets its own group named after the component type, e.g. “poses”, “intrinsics”, respecting overwrites if provided
poses_component_group – Override for poses group
intrinsics_component_group – Override for intrinsics group
masks_component_group – Override for masks group
camera_component_groups – Override for per-camera groups
lidar_component_groups – Override for per-lidar groups
radar_component_groups – Override for per-radar groups
point_clouds_component_groups – Override for per-point-cloud groups
camera_labels_component_groups – Override for per-camera-label groups
cuboid_track_observations_component_group – Override for cuboids group

Returns:

ComponentGroups with groups assigned according to profile

cuboid_track_observations_component_group: str | None#

intrinsics_component_group: str | None#

lidar_component_groups: Dict[str, str]#

masks_component_group: str | None#

point_clouds_component_groups: Dict[str, str]#

poses_component_group: str | None#

radar_component_groups: Dict[str, str]#

class ncore.data.v4.ComponentReader( component_instance_name: str, component_group: Group, )#

Bases: ABC

Base class for V4 component readers.

Subclasses must implement get_component_name() and supports_component_version(). The underlying component group is accessible as self._group; component metadata is available via the instance_name, component_version, and generic_meta_data properties.

property component_version: str#: Returns the component version of the loaded component

property generic_meta_data: JsonDocument#: Returns the generic meta data of the loaded component

abstractmethod static get_component_name() → str#: Returns the base name of the current component

get_generic_data(name: str) → ndarray#

Returns a named generic data array from this component.

Raises KeyError if the name does not exist.

async get_generic_data_async(name: str) → ndarray#: Asynchronous get_generic_data()

get_generic_data_names() → List[str]#: Returns the list of all generic data array names on this component

has_generic_data(name: str) → bool#: Returns True if a named generic data array exists on this component

property instance_name: str#: The user-defined name that distinguishes this component instance from others of the same type.

abstractmethod static supports_component_version(version: str) → bool#: Returns true if the component version is supported by the reader

class ncore.data.v4.ComponentWriter( component_group: Group, sequence_timestamp_interval_us: HalfClosedInterval, )#

Bases: ABC

Base class for V4 component writers.

Subclasses must implement get_component_name() and get_component_version(), and may override finalize() to flush buffered data. All timestamps stored by the writer must fall within the sequence’s sequence_timestamp_interval_us time range (available as self._sequence_timestamp_interval_us).

finalize() → None#

Overwrite to perform final operations after all user-data was written.

Called by SequenceComponentGroupsWriter.finalize() before component metadata is persisted. Override to flush buffered data (e.g., create arrays / groups).

abstractmethod static get_component_name() → str#: Returns the base name of the component writer

abstractmethod static get_component_version() → str#: Returns the version of the current component writer

set_generic_data( data: Dict[str, np.ndarray], meta_data: JsonDocumentLike | None = None, ) → None#

Attach named generic data arrays (and optional metadata) to this component.

Parameters:

data (Dict[str, np.ndarray]) – Named numpy arrays to store under the component’s generic_data/ group (overwrites existing arrays with the same name, if any).
meta_data (Optional[JsonDocumentLike]) – Keys provided here replace any init-time / previous generic_meta_data keys set during component registration or earlier method calls.

class ncore.data.v4.CuboidsComponent#

Bases: object

Data component representing cuboid track observations

COMPONENT_NAME: str = 'cuboids'#

class Reader( component_instance_name: str, component_group: Group, )#

Bases: ComponentReader

Cuboid tracks component reader

static get_component_name() → str#: Returns the base name of the current component

get_observations() → Generator[types.CuboidTrackObservation]#: Returns all stored cuboid track observations

static supports_component_version( version: str, ) → bool#: Returns true if the component version is supported by the reader

class Writer( component_group: Group, sequence_timestamp_interval_us: HalfClosedInterval, )#

Bases: ComponentWriter

Cuboid track observations component writer

static get_component_name() → str#: Returns the base name of the current lidar sensor component

static get_component_version() → str#: Returns the version of the current lidar sensor component

store_observations( cuboid_observations: List[CuboidTrackObservation], ) → Self#

class ncore.data.v4.IntrinsicsComponent#

Bases: object

Sensor intrinsic calibration data component

COMPONENT_NAME: str = 'intrinsics'#

class Reader( component_instance_name: str, component_group: Group, )#

Bases: ComponentReader

Sensor intrinsics data component reader

get_camera_model_parameters( camera_id: str, ) → CameraModelParameters#: Returns the camera model associated with the requested camera sensor

static get_component_name() → str#: Returns the base name of the current component

get_lidar_model_parameters( lidar_id: str, ) → LidarModelParameters | None#: Returns the lidar model associated with the requested lidar sensor

static supports_component_version( version: str, ) → bool#: Returns true if the component version is supported by the reader

class Writer( component_group: Group, sequence_timestamp_interval_us: HalfClosedInterval, )#

Bases: ComponentWriter

Sensor intrinsics data component writer

static get_component_name() → str#: Returns the base name of the current intrinsic calibration component

static get_component_version() → str#: Returns the version of the current intrinsic calibration component

store_camera_intrinsics( camera_id: str, camera_model_parameters: CameraModelParameters, ) → Self#: Store camera-associated intrinsics

store_lidar_intrinsics( lidar_id: str, lidar_model_parameters: LidarModelParameters, ) → Self#: Store lidar-associated intrinsics

class ncore.data.v4.LidarSensorComponent#

Bases: object

Lidar sensor data component

COMPONENT_NAME: str = 'lidars'#

class Reader( component_instance_name: str, component_group: Group, )#

Bases: BaseRayBundleSensorComponentReader

Lidar sensor data component reader

static get_component_name() → str#: Returns the base name of the current component

static supports_component_version( version: str, ) → bool#: Returns true if the component version is supported by the reader

class Writer( component_group: Group, sequence_timestamp_interval_us: HalfClosedInterval, )#

Bases: BaseRayBundleSensorComponentWriter

Lidar sensor data component writer

static get_component_name() → str#: Returns the base name of the current lidar sensor component

static get_component_version() → str#: Returns the version of the current lidar sensor component

store_frame( direction: npt.NDArray[np.float32], timestamp_us: npt.NDArray[np.uint64], model_element: npt.NDArray[np.uint16] | None, distance_m: npt.NDArray[np.float32], intensity: npt.NDArray[np.float32], frame_timestamps_us: npt.NDArray[np.uint64], generic_data: Dict[str, npt.NDArray[Any]], generic_meta_data: JsonDocumentLike, ) → Self#

class ncore.data.v4.MasksComponent#

Bases: object

Sensor masks data component

COMPONENT_NAME: str = 'masks'#

class Reader( component_instance_name: str, component_group: Group, )#

Bases: ComponentReader

Sensor masks data component reader

get_camera_mask_image( camera_id: str, mask_name: str, ) → Image#: Returns constant named camera mask image

async get_camera_mask_image_async( camera_id: str, mask_name: str, ) → Image#: Asynchronous get_camera_mask_image()

get_camera_mask_images( camera_id: str, ) → Generator[Tuple[str, PILImage.Image]]#: Returns all constant named camera mask images

get_camera_mask_names( camera_id: str, ) → List[str]#: Returns all constant camera mask names

static get_component_name() → str#: Returns the base name of the current component

static supports_component_version(version: str) → bool#: Returns true if the component version is supported by the reader

class Writer( component_group: Group, sequence_timestamp_interval_us: HalfClosedInterval, )#

Bases: ComponentWriter

Sensor masks data component writer

static get_component_name() → str#: Returns the base name of the current sensor masks component

static get_component_version() → str#: Returns the version of the current sensor masks component

store_camera_masks( camera_id: str, mask_images: Dict[str, Image], ) → Self#: Store camera-associated masks

class ncore.data.v4.PointCloudsComponent#

Bases: object

Data component representing unstructured point clouds with optional typed attributes.

class AttributeSchema( transform_type: AttributeTransformType = None, dtype: dtype = None, shape_suffix: Tuple[int, ...] = (), )#

Bases: DataClassJsonMixin

Schema for a single per-point attribute (frozen / hashable).

Serialization uses DataClassJsonMixin:

transform_type is stored as its uppercase enum name ("INVARIANT", "DIRECTION", "POINT").
dtype is stored as a numpy dtype string (e.g. "float32").
shape_suffix is stored as a JSON array of ints.

dtype: dtype#

shape_suffix: Tuple[int, ...]#

transform_type: AttributeTransformType#

COMPONENT_NAME: str = 'point_clouds'#

class Reader( component_instance_name: str, component_group: Group, )#

Bases: ComponentReader

Point-clouds component reader.

property attribute_names: List[str]#

property coordinate_unit: CoordinateUnit#

get_attribute_schema( name: str, ) → AttributeSchema#

static get_component_name() → str#: Returns the base name of the current component

get_pc_attribute( pc_index: int, name: str, ) → npt.NDArray[Any]#

async get_pc_attribute_async( pc_index: int, name: str, ) → npt.NDArray[Any]#: Asynchronous get_pc_attribute()

get_pc_generic_data( pc_index: int, name: str, ) → npt.NDArray[Any]#

async get_pc_generic_data_async( pc_index: int, name: str, ) → npt.NDArray[Any]#: Asynchronous get_pc_generic_data()

get_pc_generic_data_names( pc_index: int, ) → List[str]#

get_pc_generic_meta_data( pc_index: int, ) → JsonDocument#

get_pc_reference_frame_id( pc_index: int, ) → str#

get_pc_reference_frame_timestamp_us( pc_index: int, ) → int#

get_pc_xyz( pc_index: int, ) → npt.NDArray[np.float32]#

async get_pc_xyz_async( pc_index: int, ) → npt.NDArray[np.float32]#: Asynchronous get_pc_xyz()

has_pc_generic_data( pc_index: int, name: str, ) → bool#

property pc_timestamps_us: npt.NDArray[np.uint64]#

property pcs_count: int#

static supports_component_version( version: str, ) → bool#: Returns true if the component version is supported by the reader

class Writer( component_group: Group, sequence_timestamp_interval_us: HalfClosedInterval, coordinate_unit: CoordinateUnit, attribute_schemas: Dict[str, AttributeSchema] | None = None, )#

Bases: ComponentWriter

Point-clouds component writer.

finalize() → None#: Write pc_timestamps_us array (derived from per-pc reference_frame_timestamp_us values).

static get_component_name() → str#: Returns the base name of the component writer

static get_component_version() → str#: Returns the version of the current component writer

store_pc( xyz: npt.NDArray[np.float32], reference_frame_id: str, reference_frame_timestamp_us: int, attributes: Dict[str, npt.NDArray[Any]] | None = None, generic_data: Dict[str, npt.NDArray[Any]] | None = None, generic_meta_data: JsonDocumentLike | None = None, ) → None#

Store a single point cloud.

Attributes represent per-point data with a schema declared in the component-level attribute_schemas meta-data. Generic data and meta-data allow storing additional per-point-cloud information without a predefined schema.

The reference_frame_timestamp_us is also collected into the source-level pc_timestamps_us array written by finalize().

class ncore.data.v4.PosesComponent#

Bases: object

Represents a generic set of static / dynamic poses (rigid transformations) between named coordinate frames

COMPONENT_NAME: str = 'poses'#

class DynamicPosesRecord#

Bases: TypedDict

The json-encoded data of dynamic poses

dtype: str#

poses: List[List[List[float]]]#

timestamps_us: List[int]#

class Reader( component_instance_name: str, component_group: Group, )#

Bases: ComponentReader

static get_component_name() → str#: Returns the base name of the current component

get_dynamic_pose( source_frame_id: str, target_frame_id: str, ) → Tuple[npt.NDArray[np.floating], npt.NDArray[np.uint64]]#: Returns dynamic poses (time-dependent rigid transformations) between two named coordinate frames, if available (poses of the stored dtype, float32 / float64)

get_dynamic_poses() → Generator[Tuple[Tuple[str, str], Tuple[npt.NDArray[np.floating], npt.NDArray[np.uint64]]]]#: Returns all dynamic poses (time-dependent rigid transformations) between named coordinate frames, if available

get_static_pose( source_frame_id: str, target_frame_id: str, ) → npt.NDArray[np.floating]#: Returns static pose (rigid transformation) between two named coordinate frames, if available (of the stored dtype, float32 / float64)

get_static_poses() → Generator[Tuple[Tuple[str, str], npt.NDArray[np.floating]]]#: Returns all static poses (rigid transformations) between named coordinate frames, if available

static supports_component_version(version: str) → bool#: Returns true if the component version is supported by the reader

class StaticPoseRecord#

Bases: TypedDict

The json-encoded data of a static pose

dtype: str#

pose: List[List[float]]#

class Writer( component_group: Group, sequence_timestamp_interval_us: HalfClosedInterval, )#

Bases: ComponentWriter

Poses data component writer

data: Dict[str, JsonDocument]#

finalize() → None#: Actually store the json-encoded pose data

static get_component_name() → str#: Returns the base name of the current component’

static get_component_version() → str#: Returns the version of the current component writer

store_dynamic_pose( source_frame_id: str, target_frame_id: str, poses: npt.NDArray[np.floating], timestamps_us: npt.NDArray[np.uint64], require_sequence_time_coverage: bool = True, ) → Self#

Store a trajectory of dynamic poses (time-dependent rigid transformations) between two named coordinate frames.

Makes sure the inverse transformation is not already stored.

store_static_pose( source_frame_id: str, target_frame_id: str, pose: npt.NDArray[np.floating], ) → Self#

Store a static pose (rigid transformation) between two named coordinate frames.

Makes sure the inverse transformation is not already stored.

class ncore.data.v4.RadarSensorComponent#

Bases: object

Radar sensor data component

COMPONENT_NAME: str = 'radars'#

class Reader( component_instance_name: str, component_group: Group, )#

Bases: BaseRayBundleSensorComponentReader

Radar sensor data component reader

static get_component_name() → str#: Returns the base name of the current component

static supports_component_version( version: str, ) → bool#: Returns true if the component version is supported by the reader

class Writer( component_group: Group, sequence_timestamp_interval_us: HalfClosedInterval, )#

Bases: BaseRayBundleSensorComponentWriter

Radar sensor data component writer

static get_component_name() → str#: Returns the base name of the current radar sensor component

static get_component_version() → str#: Returns the version of the current radar sensor component

store_frame( direction: npt.NDArray[np.float32], timestamp_us: npt.NDArray[np.uint64], distance_m: npt.NDArray[np.float32], frame_timestamps_us: npt.NDArray[np.uint64], generic_data: Dict[str, npt.NDArray[Any]], generic_meta_data: JsonDocumentLike, ) → Self#

class ncore.data.v4.SequenceComponentGroupsReader( component_group_paths: List[UPath] | List[Path], open_consolidated: bool = True, itar_index_tail_read_size: int = 1048576, itar_index_tail_retention: Literal['release_if_consolidated', 'release_after_opening', 'keep_complete_archives', 'keep'] = 'release_if_consolidated', itar_remote_range_reads: bool = False, max_threads: int | None = None, node_cache_size: int | None = None, )#

Bases: object

SequenceComponentReader manages data component groups for reading for NCore V4 / zarr data for a single NCore sequence

property component_store_paths: List[UPath]#

static expand_component_group_paths( component_group_paths: List[UPath] | List[Path], ) → List[UPath]#: Expands possible sequence meta-data files in the given list of component group paths to the actual component group paths they reference

property generic_meta_data: JsonDocument#

get_sequence_meta() → SequenceMeta#: Returns full sequence meta-data summary (json-serializable)

async classmethod open_async( component_group_paths: List[UPath] | List[Path], open_consolidated: bool = True, itar_index_tail_read_size: int = 1048576, itar_index_tail_retention: Literal['release_if_consolidated', 'release_after_opening', 'keep_complete_archives', 'keep'] = 'release_if_consolidated', itar_remote_range_reads: bool = False, max_concurrent_requests: int | Semaphore | None = None, node_cache_size: int | None = None, ) → SequenceComponentGroupsReader#

Opens a reader asynchronously, without blocking the running event loop (see __init__() for the parameters).

The component stores of remote sequences (e.g. .zarr.itar archives on S3) are opened via asynchronous requests (one per archive, see ncore.impl.data.itar.IndexedTarArchive.open_async()), so opening the stores of a sequence, and of many sequences concurrently (e.g., via asyncio.gather), overlaps their requests.

Parameters:: max_concurrent_requests – The maximum number of component stores opened concurrently: a number (for this sequence), a semaphore shared by readers opened concurrently (e.g., to bound the requests of many sequences opened via asyncio.gather), or None (all component stores at once).

open_component_readers( component_reader_type: Type[CR], ) → Dict[str, CR]#: Instantiates all component readers for the given component of all associated stores, identified by the component instance names

reload_resources() → None#: Trigger a reload of each store - useful to re-initialize file objects in multi-process settings

property sequence_id: str#

property sequence_timestamp_interval_us: HalfClosedInterval#

class ncore.data.v4.SequenceComponentGroupsWriter( output_dir_path: UPath, store_base_name: str, sequence_id: str, sequence_timestamp_interval_us: HalfClosedInterval, generic_meta_data: JsonDocumentLike, store_type: Literal['itar', 'directory'] = 'itar', zarr_format: nodes.ZarrFormat = 2, )#

Bases: object

SequenceComponentGroupsWriter manages store groups for writing for NCore V4 / zarr data components for a single NCore sequence

finalize() → List[UPath]#

Finalize all writers, store their generic data and meta data, and close all stores after consolidating their meta data.

Returns a list of the store paths

static from_reader( output_dir_path: UPath, store_base_name: str, sequence_reader: SequenceComponentGroupsReader, store_type: Literal['itar', 'directory'] = 'itar', zarr_format: Literal[2, 3] = 2, ) → SequenceComponentGroupsWriter#: Creates a SequenceComponentGroupsWriter from an existing SequenceComponentGroupsReader instance to share consistent per-sequence meta-data

property generic_meta_data: JsonDocument#

get_base_group( component_group_name: str | None, ) → Group#: Lazily initializes ncore base-groups and underlying stores on demand

register_component_writer(

component_writer_type: Callable[Concatenate[nodes.Group, HalfClosedInterval, P], CW],

component_instance_name: str,

group_name: str | None = None,

generic_meta_data: JsonDocumentLike | None = None,

*args: P.args,

**kwargs: P.kwargs,

) → CW#

Instantiates a component writer for the given type and instance name.

The extra positional and keyword arguments are forwarded to the writer constructor. Because the first parameter is typed with ParamSpec + Concatenate, type checkers can infer the correct extra-argument signature from the concrete writer class that is passed in.

property sequence_id: str#

property sequence_timestamp_interval_us: HalfClosedInterval#

property zarr_format: Literal[2, 3]#: The on-disk zarr format of the written component stores

class ncore.data.v4.SequenceLoaderV4( reader: SequenceComponentGroupsReader, poses_component_group_name: str = 'default', intrinsics_component_group_name: str = 'default', masks_component_group_name: str | None = 'default', cuboids_component_group_name: str | None = 'default', motion_compensator_parameters: Parameters | None = None, )#

Bases: AsyncSequenceLoaderProtocol

SequenceLoader implementation for NCore V4 data (with asynchronous reads, see AsyncSequenceLoaderProtocol).

The constructor takes an opened (synchronously or asynchronously) SequenceComponentGroupsReader, see open_async() for opening a sequence’s component stores and the loader asynchronously in one step.

Provides a unified interface to access V4 format sequence data including sensors, poses, intrinsics, masks, and cuboid annotations.

Parameters:

reader – Component store reader for V4 data
poses_component_group_name – Name of the poses component group to load
intrinsics_component_group_name – Name of the intrinsics component group to load
masks_component_group_name – Name of the masks component group to load
cuboids_component_group_name – Name of the cuboids component group to load
motion_compensator_parameters – Parameters of the motion compensation of the ray bundle sensors’ point clouds (defaults to the exact method)

class CameraLabels( reader: Reader, )#

Bases: AsyncCameraLabelsProtocol

Wraps a CameraLabelsComponent.Reader to implement AsyncCameraLabelsProtocol.

get_label( timestamp_us: int, ) → CameraLabelHandle#: Return a lazy handle to the label data at the given timestamp (loadable synchronously or asynchronously)

property label_descriptor: CameraLabelDescriptor#: Descriptor of this label instance.

property label_timestamps_us: npt.NDArray[np.uint64]#: Timestamps of all stored labels, sorted ascending.

property labels_count: int#: Number of stored labels.

property labels_generic_meta_data: JsonDocument#: Generic metadata associated with all labels.

class CameraSensor( reader: Reader, mask_reader: Reader | None, model_parameters: CameraModelParameters, pose_graph: PoseGraphInterpolator, )#

Bases: Sensor, AsyncCameraSensorProtocol

Camera sensor implementation for V4 data.

Parameters:

reader – Camera component reader
mask_reader – Masks component reader for camera masks
model_parameters – Camera intrinsic model parameters
pose_graph – Pose graph interpolator for coordinate transformations

property camera_reader: Reader#

get_frame_handle( frame_index: int, ) → EncodedImageDataHandleProtocol#: Returns the frame’s encoded image data

get_mask_images() → Dict[str, Image]#: Returns all available named camera mask images

async get_mask_images_async() → Dict[str, Image]#: Returns all available named camera mask images, read asynchronously (concurrently)

property model_parameters: CameraModelParameters#: Returns parameters specific to the camera’s intrinsic model

class LidarSensor( reader: Reader, pose_graph: PoseGraphInterpolator, model_parameters: LidarModelParameters | None, motion_compensator: MotionCompensator, )#

Bases: RayBundleSensor, AsyncLidarSensorProtocol

Lidar sensor implementation for V4 data.

Parameters:

reader – Lidar component reader
pose_graph – Pose graph interpolator for coordinate transformations
model_parameters – Lidar intrinsic model parameters, if available
motion_compensator – Motion compensator of the point clouds

get_frame_ray_bundle_model_element( frame_index: int, ) → npt.NDArray[np.uint16] | None#: Returns the per-ray model elements for a ray bundle for a specific frame, if available

async get_frame_ray_bundle_model_element_async( frame_index: int, ) → npt.NDArray[np.uint16] | None#: Returns the per-ray model elements for a ray bundle for a specific frame (None if not available), read asynchronously

get_frame_ray_bundle_return_intensity( frame_index: int, return_index: int = 0, ) → npt.NDArray[np.float32]#: Returns the per-ray measured intensities for a ray bundle return for a specific frame

async get_frame_ray_bundle_return_intensity_async( frame_index: int, return_index: int = 0, ) → npt.NDArray[np.float32]#: Returns the per-ray measured intensities for a ray bundle return for a specific frame, read asynchronously

property lidar_reader: Reader#

property model_parameters: LidarModelParameters | None#: Returns parameters specific to the lidar’s intrinsic model, if available

class PointCloudsSource( reader: Reader, )#

Bases: AsyncPointCloudsSourceProtocol

Native point-clouds source wrapping a PointCloudsComponent.Reader.

Parameters:: reader – Component reader for the point-clouds data.

get_pc( pc_index: int, ) → PointCloud#

Return the point cloud at pc_index as an immutable PointCloud instance.

Schema-declared attributes are available via PointCloud.get_attribute().

async get_pc_async( pc_index: int, ) → PointCloud#: Returns the point cloud at pc_index like PointCloudsSourceProtocol.get_pc(), with its points read asynchronously (attributes are loaded lazily, as for PointCloudsSourceProtocol.get_pc())

get_pc_generic_data( pc_index: int, name: str, ) → npt.NDArray#

Return a generic data array by name for the given point cloud.

Unlike PointCloud attributes, generic data arrays are not necessarily per-point and do not participate in coordinate-frame transforms.

async get_pc_generic_data_async( pc_index: int, name: str, ) → npt.NDArray#: Returns a generic data array by name for the given point cloud, read asynchronously

get_pc_generic_data_names( pc_index: int, ) → List[str]#: Return the names of all generic data arrays stored for the given point cloud.

get_pc_generic_meta_data( pc_index: int, ) → JsonDocument#: Returns generic point cloud meta-data for a specific point-cloud.

has_pc_generic_data( pc_index: int, name: str, ) → bool#: Check whether a named generic data array exists for the given point cloud.

property pc_timestamps_us: npt.NDArray[np.uint64]#: Snapshot timestamps in microseconds, shape (pcs_count,).

property pcs_count: int#: Total number of point-cloud snapshots in this source.

property point_clouds_source_id: str#: Unique identifier of this point-clouds source.

class RadarSensor( reader: Reader, pose_graph: PoseGraphInterpolator, motion_compensator: MotionCompensator, )#

Bases: RayBundleSensor, AsyncRadarSensorProtocol

Radar sensor implementation for V4 data.

Parameters:

reader – Radar component reader
pose_graph – Pose graph interpolator for coordinate transformations
motion_compensator – Motion compensator of the point clouds

property radar_reader: Reader#

class RayBundleSensor( reader: Reader | Reader, pose_graph: PoseGraphInterpolator, motion_compensator: MotionCompensator, )#

Bases: Sensor, AsyncRayBundleSensorProtocol

Base ray bundle sensor implementation for V4 data (lidar/radar).

Parameters:

reader – Ray bundle sensor component reader (lidar or radar)
pose_graph – Pose graph interpolator for coordinate transformations
motion_compensator – Motion compensator of the point clouds

get_frame_point_cloud( frame_index: int, motion_compensation: bool, with_start_points: bool, return_index: int = 0, ) → FramePointCloud#: Returns motion-compensated or non-motion-compensated point-cloud for a specific frame

async get_frame_point_cloud_async( frame_index: int, motion_compensation: bool, with_start_points: bool, return_index: int = 0, ) → FramePointCloud#: Returns a frame point cloud like get_frame_point_cloud(), with its data read asynchronously (concurrently)

get_frame_ray_bundle_count( frame_index: int, ) → int#: Returns the number of rays for a specific frame without decoding it

get_frame_ray_bundle_direction( frame_index: int, ) → npt.NDArray[np.float32]#: Returns the per-ray directions for the ray-bundle for a specific frame

async get_frame_ray_bundle_direction_async( frame_index: int, ) → npt.NDArray[np.float32]#: Returns the per-ray directions for the ray-bundle for a specific frame, read asynchronously

get_frame_ray_bundle_return_count( frame_index: int, ) → int#: Returns the number of different ray returns for a specific frame without decoding it

get_frame_ray_bundle_return_distance_m( frame_index: int, return_index: int = 0, ) → npt.NDArray[np.float32]#: Returns the per-ray measured metric distances for the ray bundle return for a specific frame

async get_frame_ray_bundle_return_distance_m_async( frame_index: int, return_index: int = 0, ) → npt.NDArray[np.float32]#: Returns the per-ray measured metric distances for the ray bundle return for a specific frame, read asynchronously

get_frame_ray_bundle_return_valid_mask( frame_index: int, return_index: int = 0, ) → npt.NDArray[np.bool_]#: Returns the per-ray valid mask of the ray bundle returns for a specific frame

async get_frame_ray_bundle_return_valid_mask_async( frame_index: int, return_index: int = 0, ) → npt.NDArray[np.bool_]#: Returns the per-ray valid mask of the ray bundle returns for a specific frame, read asynchronously

get_frame_ray_bundle_timestamp_us( frame_index: int, ) → npt.NDArray[np.uint64]#: Returns the per-ray timestamps for the ray-bundle for a specific frame

async get_frame_ray_bundle_timestamp_us_async( frame_index: int, ) → npt.NDArray[np.uint64]#: Returns the per-ray timestamps for the ray-bundle for a specific frame, read asynchronously

property ray_bundle_reader: BaseRayBundleSensorComponentReader#

class Sensor( sensor_reader: Reader | Reader | Reader, pose_graph: PoseGraphInterpolator, )#

Bases: AsyncSensorProtocol

Base sensor implementation for V4 data providing common sensor functionality.

Parameters:

sensor_reader – Component reader for the sensor data
pose_graph – Pose graph interpolator for coordinate transformations

property frames_count: int#: The number of frames associated with the sensor

property frames_timestamps_us: npt.NDArray[np.uint64]#: The start/end timestamps of the frames associated with the sensor [(N,2) array]

get_frame_generic_data( frame_index: int, name: str, ) → npt.NDArray[Any]#: Returns generic frame-data for a specific frame and name

async get_frame_generic_data_async( frame_index: int, name: str, ) → npt.NDArray[Any]#: Returns generic frame-data for a specific frame and name, read asynchronously

get_frame_generic_data_names( frame_index: int, ) → List[str]#: List of all generic frame-data names

get_frame_generic_meta_data( frame_index: int, ) → JsonDocument#: Returns generic frame meta-data for a specific frame

has_frame_generic_data( frame_index: int, name: str, ) → bool#: Signals if named generic frame-data exists

property sensor_id: str#: The ID of the sensor

property camera_ids: List[str]#: All camera sensor IDs in the sequence

property camera_labels_ids: List[str]#: List of all camera label instance IDs.

property generic_meta_data: JsonDocument#: Generic meta-data associated with the sequence

get_camera_labels( camera_label_id: str, ) → AsyncCameraLabelsProtocol#: Get a camera label instance (with asynchronous reads) by instance ID.

get_camera_sensor( sensor_id: str, ) → AsyncCameraSensorProtocol#: Returns a camera sensor instance (with asynchronous reads) for a given sensor id

get_cuboid_track_observations( timestamp_interval_us: HalfClosedInterval | None = None, ) → Generator[CuboidTrackObservation]#

Returns all available cuboid track observations in the sequence.

Parameters:: timestamp_interval_us – If provided, only observations whose timestamp_us falls within this half-closed interval [start, stop) are returned. When None (default), all observations are returned.

get_lidar_sensor( sensor_id: str, ) → AsyncLidarSensorProtocol#: Returns a lidar sensor instance (with asynchronous reads) for a given sensor id

get_point_clouds_source( source_id: str, *, return_index: int = 0, ) → AsyncPointCloudsSourceProtocol#: Returns a point-clouds source (with asynchronous reads) for a given source id (see SequenceLoaderProtocol.get_point_clouds_source())

get_radar_sensor( sensor_id: str, ) → AsyncRadarSensorProtocol#: Returns a radar sensor instance (with asynchronous reads) for a given sensor id

get_sequence_meta() → JsonDocument#: Returns sequence-wide meta-data summary (format is instance-dependent)

property lidar_ids: List[str]#: All lidar sensor IDs in the sequence

async classmethod open_async( component_group_paths: List[UPath] | List[Path], poses_component_group_name: str = 'default', intrinsics_component_group_name: str = 'default', masks_component_group_name: str | None = 'default', cuboids_component_group_name: str | None = 'default', open_consolidated: bool = True, itar_index_tail_read_size: int = 1048576, itar_index_tail_retention: Literal['release_if_consolidated', 'release_after_opening', 'keep_complete_archives', 'keep'] = 'release_if_consolidated', itar_remote_range_reads: bool = False, max_concurrent_requests: int | Semaphore | None = None, node_cache_size: int | None = None, motion_compensator_parameters: Parameters | None = None, ) → SequenceLoaderV4#

Opens a sequence asynchronously, without blocking the running event loop (see SequenceComponentGroupsReader.open_async() and __init__() for the parameters).

Opening many sequences concurrently overlaps their requests, e.g. with a shared bound of the requests in flight:

limit = asyncio.Semaphore(64)
loaders = await asyncio.gather(
    *(SequenceLoaderV4.open_async([path], max_concurrent_requests=limit) for path in sequence_paths)
)

property point_clouds_ids: List[str]#: All native point-clouds source IDs in the sequence

property pose_graph: PoseGraphInterpolator#: The pose graph representing all static and dynamic transformations in the sequence

query_camera_labels( camera_id: str, label_type: LabelType | None = None, label_category: LabelCategory | None = None, ) → Sequence[AsyncCameraLabelsProtocol]#: Query camera label instances (with asynchronous reads) matching filters (see SequenceLoaderProtocol.query_camera_labels())

property radar_ids: List[str]#: All radar sensor IDs in the sequence

reload_resources() → None#: Reloads any resources used by the internal sequence loader (potentially required for multi-process data loading)

property sequence_id: str#: The unique identifier of the sequence

property sequence_paths: List[UPath]#: List of all dataset paths comprising this sequence (shards / components)

property sequence_timestamp_interval_us: HalfClosedInterval#: The time range of the sequence in microseconds

class ncore.data.v4.SequenceMeta( sequence_id: str, sequence_timestamp_interval_us: TimestampInterval, generic_meta_data: Dict[str, Any], version: str, component_stores: List[ComponentStoreMeta], )#

Bases: DataClassJsonMixin

Strongly typed representation for V4 sequence metadata

class ComponentInstanceMeta( version: str, generic_meta_data: Dict[str, Any], )#

Bases: DataClassJsonMixin

Component instance metadata

generic_meta_data: Dict[str, Any]#

version: str#

class ComponentStoreMeta( path: str, md5: str, components: Dict[str, Dict[str, ComponentInstanceMeta]], )#

Bases: DataClassJsonMixin

Component store metadata

components: Dict[str, Dict[str, ComponentInstanceMeta]]#

md5: str#

path: str#

class TimestampInterval(start: int, stop: int)#

Bases: DataClassJsonMixin

Timestamp interval with start and stop times in microseconds

start: int#

stop: int#

component_stores: List[ComponentStoreMeta]#

generic_meta_data: Dict[str, Any]#

sequence_id: str#

sequence_timestamp_interval_us: TimestampInterval#

version: str#

ncore.data.v4 Package#

`ncore.data.v4` Package#