robokudo.io.storage

MongoDB storage interface for RoboKudo.

This module provides the core interface between RoboKudo and MongoDB for storing and retrieving sensor data and annotations. It supports:

• Codec-driven serialization of CAS views

• KRROOD-backed annotation serialization

• Configurable view persistence boundaries

• Thread-safe database operations

The module handles:

• RGB-D camera data

• Camera calibration and transform objects

• Custom RoboKudo types

Classes

Storage

Main interface between RoboKudo and MongoDB.

Module Contents

class robokudo.io.storage.Storage(db_name: str)

Main interface between RoboKudo and MongoDB.

This class holds the main interface code between RoboKudo and the MongoDB database. It stores sensor data and CAS views by converting specialized types (NumPy arrays, ROS messages, etc.) into a BSON‑encodable dictionary format.

BLACKLISTED_TYPES: typing_extensions.Tuple[type, Ellipsis]

VIEW_COLLECTION_NAME: str = 'cas_views'

cas_view_codecs

static is_blacklisted(obj: typing_extensions.Any) → bool

Check if an object is of a blacklisted type.

Parameters:

obj – Object to check

Returns:

True if object type is blacklisted, False otherwise

static instantiate_mongo_client() → pymongo.MongoClient

Create a MongoDB client instance.

Uses environment variables RK_MONGO_HOST and RK_MONGO_PORT if set, otherwise defaults to localhost:27017.

Returns:

MongoDB client instance

class Reader(db_name: str)

Deprecated cursor-based MongoDB reader.

Deprecated since version Use: ListReader instead for better pymongo compatibility

db_reader

reset_cursor() → None

Reset the cursor to the start of the collection.

collection_has_frames() → bool

Check if collection has any frames.

Returns:

True if frames exist, False otherwise

cursor_has_frames() → bool

Check if cursor has more frames.

Returns:

True if more frames exist, False otherwise

get_next_frame() → typing_extensions.Optional[dict]

Get the next frame from the cursor.

Returns:

Next frame data or None if no more frames

class ListReader(db_name: str)

List-based MongoDB reader.

This class reads all matching records into a list for iteration, providing better compatibility across pymongo versions and simpler cursor management.

db_reader

MongoDB database instance

index: typing_extensions.Optional[int] = None

Current position in data list

data: typing_extensions.List[typing_extensions.Dict[typing_extensions.Any, typing_extensions.Any]] = []

List of loaded documents

reset_cursor() → None

Reset the reader state.

Clears and reloads all documents from the database.

cursor_has_frames() → bool

Check if more frames are available.

Returns:

True if more frames exist, False otherwise

get_next_frame() → typing_extensions.Optional[typing_extensions.Dict[typing_extensions.Any, typing_extensions.Any]]

Get the next frame from the data list.

Returns:

Next frame data or None if no more frames

db_name

Name of the MongoDB database

client: pymongo.MongoClient

MongoDB client connection

db: pymongo.synchronous.database.Database

MongoDB database instance

drop_database() → None

Drop the entire database.

static encode_view_document(view_name: str, view_value: typing_extensions.Any) → typing_extensions.Dict[str, typing_extensions.Any]

Encode one CAS view value into a serializable view document.

static decode_view_document(view_document: typing_extensions.Dict[str, typing_extensions.Any]) → tuple[str, typing_extensions.Any]

Decode one serialized view document into a CAS view value.

store_views_in_mongo(cas_dict: typing_extensions.Dict[str, typing_extensions.Any]) → None

Store CAS views in MongoDB.

Persist CAS views in a dedicated view collection and store references to those view records in cas_dict["view_ids"].

Parameters:

cas_dict – CAS as a dictionary to persist

load_views_from_mongo_in_cas(cas_document: typing_extensions.Dict[str, typing_extensions.Any]) → None

Load views from MongoDB into a CAS document.

Retrieve and decode each persisted view referenced in view_ids.

Parameters:

cas_document – CAS document to update with loaded views

static load_annotations_from_mongo_in_cas(cas_document: typing_extensions.Dict[str, typing_extensions.Any], cas: robokudo.cas.CAS) → None

Load annotations from MongoDB into a CAS.

Restore the annotations from the database and insert them into the CAS. If no (pickled) annotations are available, cas will be untouched.

Parameters:

• cas_document – A dict representing a frame of a CAS in the database

• cas – The ‘robokudo.cas.CAS’ instance where the annotations shall be inserted to

static generate_dict_from_real_cas(cas: robokudo.cas.CAS) → typing_extensions.Dict[str, typing_extensions.Any]

Convert a CAS instance to a MongoDB-compatible dictionary.

Generate a dictionary representation from CAS that is ready for storage. All views are validated through the active view codec registry.

Parameters:

cas – Input CAS that should be used to create a dict-representation of it.

Returns:

A dict with references to parts of the input CAS.

store_cas_dict(cas_dict: typing_extensions.Dict[str, typing_extensions.Any]) → pymongo.results.InsertOneResult

Store a CAS dictionary in MongoDB.

Stores the views and creates a CAS document in MongoDB that references them.

Parameters:

cas_dict – Dictionary representation of a CAS

Returns:

MongoDB ObjectId of the stored CAS document