`ncore.data.v4` Package#

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

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_dataset: Array, )#

Bases: object

References encoded image data without loading it

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

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

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: Dict[str, types.JsonLike], ) → Self#

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 zarr 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: Dict[str, types.JsonLike]#: Returns the generic meta data of the loaded component

abstractmethod static get_component_name() → str#: Returns the base name of the current 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

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

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, ) → FThetaCameraModelParameters | OpenCVPinholeCameraModelParameters | OpenCVFisheyeCameraModelParameters#: 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, ) → RowOffsetStructuredSpinningLidarModelParameters | 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: FThetaCameraModelParameters | OpenCVPinholeCameraModelParameters | OpenCVFisheyeCameraModelParameters, ) → Self#: Store camera-associated intrinsics

store_lidar_intrinsics( lidar_id: str, lidar_model_parameters: RowOffsetStructuredSpinningLidarModelParameters, ) → 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: Dict[str, types.JsonLike], ) → 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

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.PosesComponent#

Bases: object

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

COMPONENT_NAME: str = 'poses'#

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

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

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 Writer( component_group: Group, sequence_timestamp_interval_us: HalfClosedInterval, )#

Bases: ComponentWriter

Poses data component writer

data: Dict#

finalize()#: 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: Dict[str, types.JsonLike], ) → Self#

class ncore.data.v4.SequenceComponentGroupsReader( component_group_paths: List[UPath] | List[Path], open_consolidated: bool = True, max_threads: 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: Dict[str, types.JsonLike]#

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

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 itar 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: Dict[str, types.JsonLike], store_type: Literal['itar', 'directory'] = 'itar', )#

Bases: object

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

finalize() → List[UPath]#

Validates all writers and closes 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', ) → SequenceComponentGroupsWriter#: Creates a SequenceComponentGroupsWriter from an existing SequenceComponentGroupsReader instance to share consistent per-sequence meta-data

property generic_meta_data: Dict[str, types.JsonLike]#

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: Type[CW], component_instance_name: str, group_name: str | None = None, generic_meta_data: Dict[str, types.JsonLike] = {}, ) → CW#: Instantiates a component writer instance for the given component type, component instance name, and group name. Additionally stores associated generic meta data

property sequence_id: str#

property sequence_timestamp_interval_us: HalfClosedInterval#

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', )#

Bases: SequenceLoaderProtocol

SequenceLoader implementation for NCore V4 data.

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

class CameraSensor( reader: Reader, mask_reader: Reader | None, model_parameters: FThetaCameraModelParameters | OpenCVPinholeCameraModelParameters | OpenCVFisheyeCameraModelParameters, pose_graph: PoseGraphInterpolator, )#

Bases: Sensor, CameraSensorProtocol

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

property model_parameters: FThetaCameraModelParameters | OpenCVPinholeCameraModelParameters | OpenCVFisheyeCameraModelParameters#: Returns parameters specific to the camera’s intrinsic model

class LidarSensor( reader: Reader, pose_graph: PoseGraphInterpolator, model_parameters: RowOffsetStructuredSpinningLidarModelParameters | None, )#

Bases: RayBundleSensor, LidarSensorProtocol

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

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

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

property lidar_reader: Reader#

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

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

Bases: RayBundleSensor, RadarSensorProtocol

Radar sensor implementation for V4 data.

Parameters:

reader – Radar component reader
pose_graph – Pose graph interpolator for coordinate transformations

property radar_reader: Reader#

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

Bases: Sensor, RayBundleSensorProtocol

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 and motion compensation

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

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, )#: Returns the per-ray directions for the ray-bundle for a specific frame

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

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

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

property ray_bundle_reader: BaseRayBundleSensorComponentReader#

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

Bases: SensorProtocol

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

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, ) → Dict[str, Dict[str, Dict[str, JsonLike] | List[JsonLike] | str | int | float | bool | None | List[int]] | List[Dict[str, JsonLike] | List[JsonLike] | str | int | float | bool | None | List[int]] | str | int | float | bool | None | List[int]]#: 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 generic_meta_data: Dict[str, Dict[str, Dict[str, JsonLike] | List[JsonLike] | str | int | float | bool | None | List[int]] | List[Dict[str, JsonLike] | List[JsonLike] | str | int | float | bool | None | List[int]] | str | int | float | bool | None | List[int]]#: Generic meta-data associated with the sequence

get_camera_sensor( sensor_id: str, ) → CameraSensorProtocol#: Returns a camera sensor instance 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, ) → LidarSensorProtocol#: Returns a lidar sensor instance for a given sensor id

get_radar_sensor( sensor_id: str, ) → RadarSensorProtocol#: Returns a radar sensor instance for a given sensor id

get_sequence_meta() → Dict[str, Dict[str, Dict[str, JsonLike] | List[JsonLike] | str | int | float | bool | None | List[int]] | List[Dict[str, JsonLike] | List[JsonLike] | str | int | float | bool | None | List[int]] | str | int | float | bool | None | List[int]]#: Returns sequence-wide meta-data summary (format is instance-dependent)

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

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

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

ncore.data.v4 Package#

`ncore.data.v4` Package#