cosim_toolbox.dbms package

CoSim Toolbox Data Management Module.

This module provides a unified API for reading and writing co-simulation time-series data and metadata to various storage backends.

The primary entry points are the factory functions and the manager classes.

Example Usage:

from cosim_toolbox.dbms import create_timeseries_manager, TSRecord

# Create a CSV-based manager
with create_timeseries_manager("csv", "/path/to/data", analysis_name="my_sim") as ts_manager:
    record = TSRecord(...)
    ts_manager.add_record(record)

class cosim_toolbox.dbms.CSVTimeSeriesManager(*, location: str, analysis_name: str = 'default', **kwargs)

Bases: TSDataManager

Joint CSV time-series manager using composition. Manages a shared Helper for a single reader and writer instance.

property analysis_name: str

Get the current analysis name.

Returns:: analysis name
Return type:: str

backup_analysis(backup_path: str) → bool

Create a backup of the entire analysis.

Parameters:: backup_path (str) – path to back up data
Returns:: flag indicating whether backup completed
Return type:: bool

connect() → bool

Establish connection for both reader and writer.

Returns:

flag indicating whether the connection to the data backend: exists

Return type:

bool

delete_federate_data(federate_name: str) → bool

Delete all data for a specific federate.

Parameters:: federate_name (str) – federate for which to delete data
Returns:: flag indicating whether data was deleted
Return type:: bool

delete_scenario_data(scenario_name: str) → bool

NOT IMPLEMENTED

Delete data from the data backend for the specified scenario

Parameters:: scenario_name (str) – Name of scenario being deleted
Raises:: NotImplementedError –
Returns:: flag indicating whether data was deleted
Return type:: bool

disconnect() → None

Close connection for both reader and writer.

Returns:: None

get_time_range(**kwargs) → Dict[str, float]

Get time range of data in data backend

Parameters:

Dict[str (**kwargs)
any]

Returns:

Dictionary with the time range for a given set of data.

Returned dictionary is structured as:

{
    "min_time": float value,
    "max_time": float value
}

Return type:

Dict[str, float]

list_data_types(federate_names: list[str] | None = None) → List[str]

Lists data types for a given federate in the data backend

Parameters:: federate_names (str) – Name of federate being queried
Returns:: list of data types in data backend
Return type:: List[str]

list_federates() → List[str]

Lists federates with data in data backend

Returns:: list of federates with data in data backend
Return type:: List[str]

list_scenarios() → List[str]

List of scenarios in the data backend

Returns:: list of scenarios in the data backend
Return type:: List[str]

property location: Path

Location property

Returns:: location as a file path
Return type:: Path

class cosim_toolbox.dbms.JSONMetadataManager(location: str)

Bases: MDDataManager

Joint JSON metadata manager using composition. Manages a shared Path Helper for a single reader and writer instance.

backup_collection(collection_type: str, backup_path: str) → bool

Backs up the specified collection to the specified location

Parameters:

collection_type (str) – Name of collection type
backup_path (str) – Backup location for collection

Returns:

Flag indicating the collection has been backed up

Return type:

bool

connect() → bool

Establish connection for both reader and writer.

Returns:: Flag indicating that writer and reader are connected
Return type:: bool

delete(collection_type: str, name: str) → bool

Deletes the specified metadata

Parameters:

collection_type (str) – Name of the collection type
name (str) – Name of metadata to delete

Returns:

Flag indicating the data has been deleted.

Return type:

bool

delete_federation(name: str) → bool

Deletes the indicated federation metadata from the metadata store

Parameters:: name (str) – Name of federation to delete
Returns:: Indicates whether the federation metadata has been deleted
Return type:: bool

delete_scenario(name: str) → bool

Deletes the indicated scenario metadata from the metadata store

Parameters:: name (str) – Name of scenario to delete
Returns:: Indicates whether the scenario metadata has been deleted
Return type:: bool

disconnect() → None

Close connection for both reader and writer.

Returns:: None

exists(collection_type: str, name: str) → bool

Checks to see if metadata exists

Parameters:

collection_type (str) – Collection type name where metadata is being checked for existence.
name (str) – Metadata name whose existence is being checked.

Returns:

Flag indicating the existence of the named metadata

Return type:

bool

exists_federation(name: str) → bool

Checks to see if federation metadata exists in metadata store

Parameters:

name (str) – Name of federation metadata to check for existence

Returns:

Flag indicating whether federation metadata exists in: metadata store

Return type:

bool

exists_scenario(name: str) → bool

Checks to see if scenario metadata exists in metadata store

Parameters:

name (str) – Name of scenario metadata to check for existence

Returns:

Flag indicating whether scenario metadata exists in: metadata store

Return type:

bool

property federations_path: Path

property location: Path

property scenarios_path: Path

cosim_toolbox.dbms.MetadataManager: alias of MDDataManager

class cosim_toolbox.dbms.MongoMetadataManager(*, location: str, port: int | None = None, user: str | None = None, password: str | None = None, database: str = 'cst')

Bases: MDDataManager

Joint MongoDB metadata manager using composition.

connect() → bool

Establish connection for both reader and writer to MongoDB.

Returns:: Flag indicating that writer and reader are connected
Return type:: bool

property database: str

delete(collection_type: str, name: str) → bool

Deletes the specified metadata

Parameters:

collection_type (str) – Name of the collection type
name (str) – Name of metadata to delete

Returns:

Flag indicating the data has been deleted.

Return type:

bool

delete_federation(name: str) → bool

Deletes the indicated federation metadata from MongoDB

Parameters:: name (str) – Name of federation to delete
Returns:: Indicates whether the federation metadata has been deleted
Return type:: bool

delete_scenario(name: str) → bool

Deletes the indicated scenario metadata from MongoDB

Parameters:: name (str) – Name of scenario to delete
Returns:: Indicates whether the scenario metadata has been deleted
Return type:: bool

disconnect() → None

Close connection for both reader and writer to MongoDB.

Returns:: None

exists(collection_type: str, name: str) → bool

Checks to see if metadata exists

Parameters:

collection_type (str) – Collection type name where metadata is being checked for existence.
name (str) – Metadata name whose existence is being checked.

Returns:

Flag indicating the existence of the named metadata

Return type:

bool

exists_federation(name: str) → bool

Checks to see if federation metadata exists in metadata store

Parameters:

name (str) – Name of federation metadata to check for existence

Returns:

Flag indicating whether federation metadata exists in: metadata store

Return type:

bool

exists_scenario(name: str) → bool

Checks to see if scenario metadata exists in metadata store

Parameters:

name (str) – Name of scenario metadata to check for existence

Returns:

Flag indicating whether scenario metadata exists in: metadata store

Return type:

bool

get_database_stats() → Dict[str, Any]

Gets MongoDB stats

Returns:: MongoDB database statistics
Return type:: dict

property location: str

class cosim_toolbox.dbms.PostgreSQLTimeSeriesManager(*, location: str, port: int, database: str, user: str, password: str, analysis_name: str, use_timescale: bool = False, **kwargs)

Bases: TSDataManager

Joint PostgreSQL time-series manager using composition.

connect() → bool

Establishes connection to the PostgreSQL server.

Returns:

Flag indicating successful connection to the PostgreSQL: database

Return type:

bool

delete_federate_data(federate_name: str) → bool

Delete data from PostgreSQL database for specified data

Parameters:: federate_name
Returns:: Flag indicating data was deleted
Return type:: bool

delete_scenario_data(scenario_name: str) → bool

Delete data from PostgreSQL database from specified scenario

Parameters:: scenario_name (str) – name of scenario to delete from database
Returns:: Flag indicating data was deleted
Return type:: bool

disconnect() → None

Closes the connection to the PostgreSQL server.

Returns:: None

get_time_range(**kwargs) → Dict[str, float]

Produces time range of data in PostgreSQL database

Returns:

Time range information as:

{
    "min_time": float
    "max_time": float
}

Time values are in ordinal time

Return type:

Dict[str, float]

list_data_types() → List[str]

Produces a list of data types in the PostgreSQL database

Returns:: ist of data types in the PostgreSQL database
Return type:: List[str]

list_federates() → List[str]

Produces a list of federations stored in the metadata store

Returns:: list of federations
Return type:: List[str]

list_scenarios() → List[str]

Produces a list of scenarios stored in the metadata store

Returns:: list of scenarios
Return type:: List[str]

class cosim_toolbox.dbms.TSRecord(real_time: datetime.datetime, sim_time: float, scenario: str, federate: str, data_name: str, data_value: Any, receiving_federate: str | None = None, receiving_endpoint: str | None = None, data_type: str | None = None)

Bases: object

data_name: str

data_type: str | None = None

data_value: Any

federate: str

real_time: datetime

receiving_endpoint: str | None = None

receiving_federate: str | None = None

scenario: str

sim_time: float

cosim_toolbox.dbms.TimeSeriesManager: alias of TSDataManager

cosim_toolbox.dbms.create_metadata_manager(backend: str = 'mongo', **kwargs) → MDDataManager

Factory function to create appropriate metadata manager. :param backend: Backend type (“json”, “mongo”). :type backend: str :param **kwargs: Backend-specific options.

Returns:: Manager for writing data with the specified backend
Return type:: MetadataManger

cosim_toolbox.dbms.create_timeseries_manager(backend: str = 'postgres', analysis_name: str = 'default', **kwargs) → TSDataManager

Factory function to create appropriate time-series data manager. :param backend: Backend type (“csv”, “postgres”). :type backend: str :param analysis_name: Analysis name. :type analysis_name: str :param **kwargs: Backend-specific options.

Returns:: Manager for writing data with the specified backend
Return type:: TimeSeriesManager

Submodules

Core data management abstractions for CoSim Toolbox. Provides TSRecord dataclass and abstract base classes for data I/O.

@author Nathan Gray

class cosim_toolbox.dbms.abstractions.MDDataManager(**kwargs)

Bases: ABC

Abstract base class for combined metadata management.

abstractmethod connect() → bool

Establish connection for both reader and writer.

Returns:: flag indicating if connected to the data store
Return type:: bool

abstractmethod disconnect() → None

Close connection for both reader and writer.

Returns:: None

property is_connected: bool

Check if the data store is connected

Returns:: True if writer is connected to data store
Return type:: bool

list_custom_collections() → list[str]

Provides the list of custom collection names in data store

Returns:: List of custom collection names
Return type:: list[str]

list_federations() → list[str]

Provides list of federations with dictionaries in the data store

Returns:: List of federations dictionary names
Return type:: list[str]

list_items(collection_type: str) → list[str]

Lists the names of the dictionaries in the specified collection

Parameters:: collection_type (str) – Name of collection whose dictionaries are to be listed
Returns:: List dictionary names in specified collection
Return type:: list[str]

list_scenarios() → list[str]

Provides list of scenarios with dictionaries in the data store

Returns:: List of scenario dictionary names
Return type:: list[str]

read(collection_type: str, name: str) → Dict[str, Any] | None

Generic read method for data backend

Allows users to access any dictionary saved in the metadata store as long as they know the name of the dictionary and the collection in which it resides.

Parameters:

collection_type (str) – Name of metadata collection from which to read
name (str) – Name of dictionary to be read

Returns:

Requested data

Return type:

Optional[Dict[str, Any]]

read_federation(name: str) → Dict[str, Any] | None

Reads the federation dictionary from the data store

Parameters:: name (str) – Name of federation dictionary to read
Returns:: requested federation dictionary
Return type:: Optional[Dict[str, Any]]

read_scenario(name: str) → Dict[str, Any] | None

Reads the scenario dictionary from the data store

Parameters:: name (str) – Name of scenario to read
Returns:: requested scenario dictionary
Return type:: Optional[Dict[str, Any]]

write(collection_type: str, name: str, data: Dict[str, Any], overwrite: bool = False) → bool

Generic write method for data backend

This method allows users to write arbitrary metadata in a hierarchy that is helpful for them.

Parameters:

collection_type (str) – Name of collection (folder or group) into which this dictionary is added. Default values for CST are “federation” and “scenario” but users can define custom collections with this method.
name (str) – Name of dictionary being added; serves as a unique identifier.
data (Dict[str, Any]) – Data to be written to the data store
overwrite (bool, optional) – Flag indicating if any existing dictionary should be overwritten. Defaults to False.

Returns:

_description_

Return type:

bool

write_federation(name: str, federation_data: Dict[str, Any], overwrite: bool = False) → bool

Writes passed in federation data to the data store

Parameters:

name (str) – Name of federation; unique identifier for this dictionary
federation_data (Dict[str, Any]) – federation data dictionary to be written
overwrite (bool, optional) – Flag indicating if any existing federation dictionary should be overwritten. Defaults to False.

Returns:

flag indicating whether the federation dictionary was: successfully written to the data store

Return type:

bool

write_scenario(name: str, scenario_data: Dict[str, Any], overwrite: bool = False) → bool

Writes passed in scenario data to the data store

Parameters:

name (str) – Name of scenario; unique identifier for this dictionary
scenario_data (Dict[str, Any]) – scenario data dictionary to be written
overwrite (bool, optional) – Flag indicating if any existing scenario dictionary should be overwritten. Defaults to False.

Returns:

flag indicating whether the scenario dictionary was: successfully written to the data store

Return type:

bool

class cosim_toolbox.dbms.abstractions.MDDataReader

Bases: ABC

Abstract base class for metadata readers. All metadata readers must implement these methods to provide a consistent API across different storage backends.

abstractmethod connect() → bool

Establish connection to the data store.

Returns:: flag indicated if the connection to the data store exists
Return type:: bool

abstractmethod disconnect() → None

Close connection to the data store.

Returns:: flag indicated if the connection to the data store exists
Return type:: bool

property is_connected: bool

Check if the reader is connected.

Returns:: flag indicating if data store is connected
Return type:: bool

abstractmethod list_custom_collections() → list[str]

List available custom collection names.

Returns:: List of custom collection names
Return type:: list[str]

abstractmethod list_federations() → list[str]

List available federation names.

Returns:: list of names of federations as strings
Return type:: list[str]

abstractmethod list_items(collection_type: str) → list[str]

List available items in a collection (generic method).

Parameters:: collection_type (str) – Collection/category name
Returns:: List of item names
Return type:: list[str]

abstractmethod list_scenarios() → list[str]

List available scenario names.

Returns:: list of names of scenarios as strings
Return type:: list[str]

abstractmethod read(collection_type: str, name: str) → Dict[str, Any] | None

Read metadata from the data store (generic method).

Parameters:

collection_type (str) – Collection/category name type
name (str) – Data identifier name

Returns:

Data or None if not found

Return type:

Optional[Dict[str, Any]]

abstractmethod read_federation(name: str) → Dict[str, Any] | None

Read federation metadata from the data store.

Parameters:: name (str) – Federation name
Returns:: Federation data or None if not found
Return type:: Optional[Dict[str, Any]]

abstractmethod read_scenario(name: str) → Dict[str, Any] | None

Read scenario metadata from the data store.

Parameters:: name (str) – Scenario name
Returns:: Scenario data or None if not found
Return type:: Optional[Dict[str, Any]]

class cosim_toolbox.dbms.abstractions.MDDataWriter

Bases: ABC

Abstract base class for metadata writers. All metadata writers must implement these methods to provide a consistent API across different storage backends.

abstractmethod connect() → bool

Establish connection to the data store.

Returns:

flag indicated if the connection to the data store: was established

Return type:

bool

abstractmethod disconnect() → None

Close connection to the data store.

Returns:: None

property is_connected: bool

Check if the writer is connected.

Returns:: flag indicated if the connection to the data store exists
Return type:: bool

abstractmethod write(collection_type: str, name: str, data: Dict[str, Any], overwrite: bool = False) → bool

Write metadata to the data store (generic method).

Parameters:

collection_type (str) – Collection/category name type
name (str) – Data identifier name
data (Dict[str, Any]) – Data to write
overwrite (bool) – Whether to overwrite existing data

Returns:

True if write successful, False otherwise

Return type:

bool

abstractmethod write_federation(name: str, federation_data: Dict[str, Any], overwrite: bool = False) → bool

Write federation metadata to the data store.

Parameters:

name (str) – Federation name
federation_data (Dict[str, Any]) – Federation configuration data
overwrite (bool) – Whether to overwrite existing data

Returns:

True if write successful, False otherwise

Return type:

bool

abstractmethod write_scenario(name: str, scenario_data: Dict[str, Any], overwrite: bool = False) → bool

Write scenario metadata to the data store.

Parameters:

name (str) – Scenario name
scenario_data (Dict[str, Any]) – Scenario configuration data
overwrite (bool) – Whether to overwrite existing data

Returns:

True if write successful, False otherwise

Return type:

bool

cosim_toolbox.dbms.abstractions.MetadataManager: alias of MDDataManager

class cosim_toolbox.dbms.abstractions.TSDataManager(**kwargs)

Bases: ABC

Abstract base class for combined time-series data management.

add_record(record: TSRecord) → None

Adds a single record (data point) to the data store

Parameters:: record (TSRecord) – Data record to be written
Returns:: None

property buffer_size: int

Gets the number of elements in the data buffer

Returns:: number of elements in the data buffer
Return type:: int

abstractmethod connect() → bool

Establish connection for both reader and writer.

Returns:: flag indicating whether the connection to the store exist
Return type:: bool

abstractmethod disconnect() → None

Close connection for both reader and writer.

Returns:: None

flush() → bool

_summary_

Returns:: flag indicating the success of the write to disk
Return type:: bool

property is_connected: bool

Flag indicating if connected to the data store

Returns:: flag indicating if connected to the data store
Return type:: bool

read_data(**kwargs) → DataFrame

Reads data from data backend

Returns:: Requested data from data store
Return type:: pd.DataFrame

write_records(records: list[TSRecord]) → bool

Writes records to the data store

Parameters:

records (list[TSRecord]) – Data to be written to the data store

Returns:

flag indicating whether the data was successfully written: to the data store

Return type:

bool

class cosim_toolbox.dbms.abstractions.TSDataReader

Bases: ABC

Abstract base class for time-series data readers. All time-series data readers must implement these methods to provide a consistent API across different storage backends.

abstractmethod connect() → bool

Establish connection to the data store.

Returns:: flag indicated if the connection to the data store exists
Return type:: bool

abstractmethod disconnect() → None

Close connection to the data store.

Returns:: flag indicated if the connection to the data store exists
Return type:: bool

property is_connected: bool

Check if the reader is connected.

Returns:: flag indicated if the connection to the data store exists
Return type:: bool

Read time-series data from the data store.

Parameters:

start_time (Optional[float]) – Starting time for data query
duration (Optional[float]) – Duration in seconds for data query
scenario_name (Optional[str]) – Filter by scenario name
federate_name (Optional[str]) – Filter by federate name
data_name (Optional[str]) – Filter by data name
data_type (Optional[str]) – Filter by data type

Returns:

Time-series data as a Pandas DataFrame

Return type:

pd.DataFrame

class cosim_toolbox.dbms.abstractions.TSDataWriter

Bases: ABC

Abstract base class for time-series data writers. All time-series data writers must implement these methods to provide a consistent API across different storage backends.

add_record(record: TSRecord) → None

Add a single TSRecord to the internal buffer.

Parameters:: record (TSRecord) – Time-series record to add
Returns:: None

property buffer_size: int

Get current buffer size.

Returns:: number of elements in buffer list
Return type:: int

abstractmethod connect() → bool

Establish connection to the data store.

Returns:: None

abstractmethod disconnect() → None

Close connection to the data store.

Returns:: None

flush() → bool

Write all buffered records to the data store and clear buffer.

Returns:: True if flush successful, False otherwise
Return type:: bool

property is_connected: bool

Check if the writer is connected.

Returns:: True if writer is connected to data store
Return type:: bool

abstractmethod write_records(records: list[TSRecord]) → bool

Write TSRecord objects to the data store.

Parameters:: records (list[TSRecord]) – List of time-series records to write
Returns:: True if write successful, False otherwise
Return type:: bool

class cosim_toolbox.dbms.abstractions.TSRecord(real_time: datetime.datetime, sim_time: float, scenario: str, federate: str, data_name: str, data_value: Any, receiving_federate: str | None = None, receiving_endpoint: str | None = None, data_type: str | None = None)

Bases: object

data_name: str

data_type: str | None = None

data_value: Any

federate: str

real_time: datetime

receiving_endpoint: str | None = None

receiving_federate: str | None = None

scenario: str

sim_time: float

cosim_toolbox.dbms.abstractions.TimeSeriesManager: alias of TSDataManager

CSV file-based time-series data management for CoSim Toolbox. Refactored to use a composition-based architecture for clarity, testability, and maintainability.

@author Nathan Gray

class cosim_toolbox.dbms.csv_timeseries.CSVTimeSeriesManager(*, location: str, analysis_name: str = 'default', **kwargs)

Bases: TSDataManager

Joint CSV time-series manager using composition. Manages a shared Helper for a single reader and writer instance.

property analysis_name: str

Get the current analysis name.

Returns:: analysis name
Return type:: str

backup_analysis(backup_path: str) → bool

Create a backup of the entire analysis.

Parameters:: backup_path (str) – path to back up data
Returns:: flag indicating whether backup completed
Return type:: bool

connect() → bool

Establish connection for both reader and writer.

Returns:

flag indicating whether the connection to the data backend: exists

Return type:

bool

delete_federate_data(federate_name: str) → bool

Delete all data for a specific federate.

Parameters:: federate_name (str) – federate for which to delete data
Returns:: flag indicating whether data was deleted
Return type:: bool

delete_scenario_data(scenario_name: str) → bool

NOT IMPLEMENTED

Delete data from the data backend for the specified scenario

Parameters:: scenario_name (str) – Name of scenario being deleted
Raises:: NotImplementedError –
Returns:: flag indicating whether data was deleted
Return type:: bool

disconnect() → None

Close connection for both reader and writer.

Returns:: None

get_time_range(**kwargs) → Dict[str, float]

Get time range of data in data backend

Parameters:

Dict[str (**kwargs)
any]

Returns:

Dictionary with the time range for a given set of data.

Returned dictionary is structured as:

{
    "min_time": float value,
    "max_time": float value
}

Return type:

Dict[str, float]

list_data_types(federate_names: list[str] | None = None) → List[str]

Lists data types for a given federate in the data backend

Parameters:: federate_names (str) – Name of federate being queried
Returns:: list of data types in data backend
Return type:: List[str]

list_federates() → List[str]

Lists federates with data in data backend

Returns:: list of federates with data in data backend
Return type:: List[str]

list_scenarios() → List[str]

List of scenarios in the data backend

Returns:: list of scenarios in the data backend
Return type:: List[str]

property location: Path

Location property

Returns:: location as a file path
Return type:: Path

class cosim_toolbox.dbms.csv_timeseries.CSVTimeSeriesReader(*, location: str | None = None, analysis_name: str = 'default', helper: _CSVHelper | None = None)

Bases: TSDataReader

CSV file-based time-series data reader.

connect() → bool

Verify that the directory structure exists.

Returns:

flag indicated whether the connection to the data backend: exists

Return type:

bool

disconnect() → None

Close connection (no-op for files).

Returns:: None

get_time_range(**kwargs) → Dict[str, float]

Get the time range (min and max simulation times) for the data.

Parameters:

**kwargs (Dict[str, float])

Returns:

Dictionary with the time range for a given set of data.

Returned dictionary is structured as:

{
    "min_time": float value,
    "max_time": float value
}

Return type:

Dict[str, float]

list_data_types(federate_names: list[str] | None = None) → List[str]

Provides a list of data types in data backend for a given federate

Parameters:: federate_names (str) – name of federate whose available data types are being determined
Returns:: list of data types in data backend
Return type:: List[str]

list_federates() → List[str]

Provides a list of federates with data in the data backend

Returns:: list of federates with data in the data backend
Return type:: List[str]

list_scenarios() → List[str]

Get list of unique scenarios in the data.

Returns:: list of unique scenarios in the data
Return type:: List[str]

Read time-series data from CSV files.

Parameters:

start_time (Optional[float], optional) – Start time (ordinal time in seconds) for requested data. Defaults to None.
duration (Optional[float], optional) – Length of time (in seconds) from the data to read . Defaults to None.
scenario_name (Optional[str | list], optional) – Name(s) of scenario to read. Defaults to None.
federate_name (Optional[str | list], optional) – Name(s) of federate to read. Defaults to None.
data_name (Optional[str | list], optional) – Name(s) of data to read. Defaults to None.
data_type (Optional[str | list], optional) – Data type(s) to read. Defaults to None.

Returns:

requested data

Return type:

pd.DataFrame

class cosim_toolbox.dbms.csv_timeseries.CSVTimeSeriesWriter(*, location: str | None = None, analysis_name: str = 'default', helper: _CSVHelper | None = None)

Bases: TSDataWriter

CSV file-based time-series data writer.

connect() → bool

Create directory structure if it doesn’t exist.

Returns:

Flag indicating whether directory structure has been: created.

Return type:

bool

disconnect() → None

Close connection (no-op for files, but maintains consistency).

Returns:: None

write_records(records: List[TSRecord]) → bool

Write TSRecord objects to CSV files.

Parameters:

records (List[TSRecord]) – Records (data) to be written to the data backend

Returns:

flag indicating whether the data was written to the data: backend

Return type:

bool

Factory functions for creating data managers.

@author Nathan Gray

cosim_toolbox.dbms.factories.create_metadata_manager(backend: str = 'mongo', **kwargs) → MDDataManager

Factory function to create appropriate metadata manager. :param backend: Backend type (“json”, “mongo”). :type backend: str :param **kwargs: Backend-specific options.

Returns:: Manager for writing data with the specified backend
Return type:: MetadataManger

cosim_toolbox.dbms.factories.create_timeseries_manager(backend: str = 'postgres', analysis_name: str = 'default', **kwargs) → TSDataManager

Factory function to create appropriate time-series data manager. :param backend: Backend type (“csv”, “postgres”). :type backend: str :param analysis_name: Analysis name. :type analysis_name: str :param **kwargs: Backend-specific options.

Returns:: Manager for writing data with the specified backend
Return type:: TimeSeriesManager

JSON file-based metadata management for CoSim Toolbox. Refactored to use Composition over Inheritance for clarity and maintainability.

@author Nathan Gray

class cosim_toolbox.dbms.json_metadata.JSONMetadataManager(location: str)

Bases: MDDataManager

Joint JSON metadata manager using composition. Manages a shared Path Helper for a single reader and writer instance.

backup_collection(collection_type: str, backup_path: str) → bool

Backs up the specified collection to the specified location

Parameters:

collection_type (str) – Name of collection type
backup_path (str) – Backup location for collection

Returns:

Flag indicating the collection has been backed up

Return type:

bool

connect() → bool

Establish connection for both reader and writer.

Returns:: Flag indicating that writer and reader are connected
Return type:: bool

delete(collection_type: str, name: str) → bool

Deletes the specified metadata

Parameters:

collection_type (str) – Name of the collection type
name (str) – Name of metadata to delete

Returns:

Flag indicating the data has been deleted.

Return type:

bool

delete_federation(name: str) → bool

Deletes the indicated federation metadata from the metadata store

Parameters:: name (str) – Name of federation to delete
Returns:: Indicates whether the federation metadata has been deleted
Return type:: bool

delete_scenario(name: str) → bool

Deletes the indicated scenario metadata from the metadata store

Parameters:: name (str) – Name of scenario to delete
Returns:: Indicates whether the scenario metadata has been deleted
Return type:: bool

disconnect() → None

Close connection for both reader and writer.

Returns:: None

exists(collection_type: str, name: str) → bool

Checks to see if metadata exists

Parameters:

collection_type (str) – Collection type name where metadata is being checked for existence.
name (str) – Metadata name whose existence is being checked.

Returns:

Flag indicating the existence of the named metadata

Return type:

bool

exists_federation(name: str) → bool

Checks to see if federation metadata exists in metadata store

Parameters:

name (str) – Name of federation metadata to check for existence

Returns:

Flag indicating whether federation metadata exists in: metadata store

Return type:

bool

exists_scenario(name: str) → bool

Checks to see if scenario metadata exists in metadata store

Parameters:

name (str) – Name of scenario metadata to check for existence

Returns:

Flag indicating whether scenario metadata exists in: metadata store

Return type:

bool

property federations_path: Path

property location: Path

property scenarios_path: Path

class cosim_toolbox.dbms.json_metadata.JSONMetadataReader(*, location: str | Path | None = None, helper: _JSONPathHelper | None = None)

Bases: MDDataReader

JSON file-based metadata reader.

For standalone use:: reader = JSONMetadataReader(location=”/path/to/data”)
For managed use (by JSONMetadataManager):: helper = _JSONPathHelper(…) reader = JSONMetadataReader(helper=helper)

connect() → bool

Verify that the directory structure exists.

Returns:: Flag indicating whether the JSON files able to be accessed or not.
Return type:: bool

disconnect() → None

Close connection (no-op for files, but maintains state).

Returns:: None

list_custom_collections() → List[str]

Lists custom collection names in metadata store

Returns:: List of collection names in metadata store
Return type:: List[str]

list_federations() → List[str]

Produces a list of federations stored in the metadata store

Returns:: list of federations
Return type:: List[str]

list_items(collection_type: str) → List[str]

Produces a list of the named data in the collection in the metadata store.

Parameters:: collection_type (str) – Collection whose contents are being listed
Returns:: List of the named data in the specified collection
Return type:: List[str]

list_scenarios() → List[str]

Produces a list of scenarios stored in the metadata store

Returns:: list of scenarios
Return type:: List[str]

read(collection_type: str, name: str) → Dict[str, Any] | None

Reads data from metadata store, generally data in custom collections.

Parameters:

collection_type (str) – name of collection type
name (str) – name of data to be read

Returns:

read data

Return type:

Optional[Dict[str, Any]]

read_federation(name: str) → Dict[str, Any] | None

Reads federation configuration data

Parameters:: name (str) – name of federation being read
Returns:: Federation configuration data
Return type:: Optional[Dict[str, Any]]

read_scenario(name: str) → Dict[str, Any] | None

Reads scenario metadata

Parameters:: name (str) – name of scenario being read
Returns:: Scenario metadata
Return type:: Optional[Dict[str, Any]]

class cosim_toolbox.dbms.json_metadata.JSONMetadataWriter(*, location: str | Path | None = None, helper: _JSONPathHelper | None = None)

Bases: MDDataWriter

JSON file-based metadata writer.

For standalone use:: writer = JSONMetadataWriter(location=”/path/to/data”)
For managed use (by JSONMetadataManager):: helper = _JSONPathHelper(…) writer = JSONMetadataWriter(helper=helper)

connect() → bool

Create directory structure if it doesn’t exist.

Returns:: Flag indicating success in creating the necessary folders for storing the JSON files.
Return type:: bool

disconnect() → None

Close connection (no-op for files, but maintains state).

Returns:: None

write(collection_type: str, name: str, data: Dict[str, Any], overwrite: bool = False) → bool

_summary_

Parameters:

collection_type (str) – Type of data to store. Standard values are “federations” and “scenarios” but could also be custom values of an arbitrary types (e.g. “configurations”)
name (str) – Name of data to store
data (Dict[str, Any]) – Data to be written
overwrite (bool, optional) – Flag indicating if existing data should be overwritten. Defaults to False.

Returns:

Flag indicating whether the data was written successfully.

Return type:

bool

write_federation(name: str, federation_data: Dict[str, Any], overwrite: bool = False) → bool

Write federation metadata to the data store.

Parameters:

name (str) – Federation name
federation_data (Dict[str, Any]) – Federation configuration data
overwrite (bool) – Whether to overwrite existing data

Returns:

True if write successful, False otherwise

Return type:

bool

write_scenario(name: str, scenario_data: Dict[str, Any], overwrite: bool = False) → bool

Write scenario metadata to the data store.

Parameters:

name (str) – Scenario name
scenario_data (Dict[str, Any]) – Scenario configuration data
overwrite (bool) – Whether to overwrite existing data

Returns:

True if write successful, False otherwise

Return type:

bool

MongoDB-based metadata management for CoSim Toolbox. Refactored to use a composition-based architecture for clarity, testability, and maintainability.

class cosim_toolbox.dbms.mongo_metadata.MongoMetadataManager(*, location: str, port: int | None = None, user: str | None = None, password: str | None = None, database: str = 'cst')

Bases: MDDataManager

Joint MongoDB metadata manager using composition.

connect() → bool

Establish connection for both reader and writer to MongoDB.

Returns:: Flag indicating that writer and reader are connected
Return type:: bool

property database: str

delete(collection_type: str, name: str) → bool

Deletes the specified metadata

Parameters:

collection_type (str) – Name of the collection type
name (str) – Name of metadata to delete

Returns:

Flag indicating the data has been deleted.

Return type:

bool

delete_federation(name: str) → bool

Deletes the indicated federation metadata from MongoDB

Parameters:: name (str) – Name of federation to delete
Returns:: Indicates whether the federation metadata has been deleted
Return type:: bool

delete_scenario(name: str) → bool

Deletes the indicated scenario metadata from MongoDB

Parameters:: name (str) – Name of scenario to delete
Returns:: Indicates whether the scenario metadata has been deleted
Return type:: bool

disconnect() → None

Close connection for both reader and writer to MongoDB.

Returns:: None

exists(collection_type: str, name: str) → bool

Checks to see if metadata exists

Parameters:

collection_type (str) – Collection type name where metadata is being checked for existence.
name (str) – Metadata name whose existence is being checked.

Returns:

Flag indicating the existence of the named metadata

Return type:

bool

exists_federation(name: str) → bool

Checks to see if federation metadata exists in metadata store

Parameters:

name (str) – Name of federation metadata to check for existence

Returns:

Flag indicating whether federation metadata exists in: metadata store

Return type:

bool

exists_scenario(name: str) → bool

Checks to see if scenario metadata exists in metadata store

Parameters:

name (str) – Name of scenario metadata to check for existence

Returns:

Flag indicating whether scenario metadata exists in: metadata store

Return type:

bool

get_database_stats() → Dict[str, Any]

Gets MongoDB stats

Returns:: MongoDB database statistics
Return type:: dict

property location: str

Bases: MDDataReader

MongoDB-based metadata reader.

connect() → bool

Connect to MongoDB

Returns:: Flag indicating database connection has been established
Return type:: bool

disconnect() → None

Disconnects from MongoDB

Returns:: None

list_custom_collections() → List[str]

Lists custom collection names in metadata store

Returns:: List of collection names in metadata store
Return type:: List[str]

list_federations() → List[str]

Produces a list of federations stored in the metadata store

Returns:: list of federations
Return type:: List[str]

list_items(collection_type: str) → List[str]

Produces a list of the named data in the collection in the metadata store.

Parameters:: collection_type (str) – Collection whose contents are being listed
Returns:: List of the named data in the specified collection
Return type:: List[str]

list_scenarios() → List[str]

Produces a list of scenarios stored in the metadata store

Returns:: list of scenarios
Return type:: List[str]

read(collection_type: str, name: str) → Dict[str, Any] | None

General read from MongoDB

Parameters:

collection_type (str) – Name of collection metadata
name (str) – Metadata name

Returns:

metadata from MongoDB

Return type:

Optional[Dict[str, Any]]

read_federation(name: str) → Dict[str, Any] | None

Reads the specified federation metadata from MongoDB

Parameters:: name (str) – Name of federation to read from MongoDB
Returns:: Federation metadata
Return type:: Optional[Dict[str, Any]]

read_scenario(name: str) → Dict[str, Any] | None

Reads the specified scenario metadata from MongoDB

Parameters:: name (str) – Name of scenario to read from MongoDB
Returns:: Scenario metadata
Return type:: Optional[Dict[str, Any]]

Bases: MDDataWriter

MongoDB-based metadata writer.

connect() → bool

Connects to the database, creating the connection if owned.

Returns:: Flag indicating successful connection to the Mongo database
Return type:: bool

disconnect() → None

Disconnects from the database if the connection is owned.

Returns:: None

write(collection_type: str, name: str, data: Dict[str, Any], overwrite: bool = False) → bool

Writes arbitrary data to MongoDB database

Parameters:

collection_type (str) – Name of collection type where data is to be written.
name (str) – Name of metadata to write
data (Dict[str, Any]) – Metadata to write
overwrite (bool, optional) – Set to overwrite any exiting metadata of the same name in the MongoDB. Defaults to False.

Raises:

ValidationError – Data to write is not a dictionary

Returns:

Flag indicating data was successfully written to MongoDB.

Return type:

bool

write_federation(name: str, federation_data: Dict[str, Any], overwrite: bool = False) → bool

Write federation metadata to MongoDB

Parameters:

name (str) – Name of federation
federation_data (Dict[str, Any]) – Federation metadata
overwrite (bool, optional) – Set flag to overwrite any existing metadata. Defaults to False.

Returns:

Flag indicating data was successfully added to MongoDB database

Return type:

bool

write_scenario(name: str, scenario_data: Dict[str, Any], overwrite: bool = False) → bool

Write scenario metadata to MongoDB

Parameters:

scenario_data
name (str) – Name of scenario
overwrite (bool, optional) – Set flag to overwrite any existing metadata. Defaults to False.

Returns:

Flag indicating data was successfully added to MongoDB: database

Return type:

bool

PostgreSQL-based time-series data management for CoSim Toolbox. Refactored to use a composition-based architecture for clarity, testability, and maintainability.

@author Nathan Gray

class cosim_toolbox.dbms.postgresql_timeseries.PostgreSQLTimeSeriesManager(*, location: str, port: int, database: str, user: str, password: str, analysis_name: str, use_timescale: bool = False, **kwargs)

Bases: TSDataManager

Joint PostgreSQL time-series manager using composition.

connect() → bool

Establishes connection to the PostgreSQL server.

Returns:

Flag indicating successful connection to the PostgreSQL: database

Return type:

bool

delete_federate_data(federate_name: str) → bool

Delete data from PostgreSQL database for specified data

Parameters:: federate_name
Returns:: Flag indicating data was deleted
Return type:: bool

delete_scenario_data(scenario_name: str) → bool

Delete data from PostgreSQL database from specified scenario

Parameters:: scenario_name (str) – name of scenario to delete from database
Returns:: Flag indicating data was deleted
Return type:: bool

disconnect() → None

Closes the connection to the PostgreSQL server.

Returns:: None

get_time_range(**kwargs) → Dict[str, float]

Produces time range of data in PostgreSQL database

Returns:

Time range information as:

{
    "min_time": float
    "max_time": float
}

Time values are in ordinal time

Return type:

Dict[str, float]

list_data_types() → List[str]

Produces a list of data types in the PostgreSQL database

Returns:: ist of data types in the PostgreSQL database
Return type:: List[str]

list_federates() → List[str]

Produces a list of federations stored in the metadata store

Returns:: list of federations
Return type:: List[str]

list_scenarios() → List[str]

Produces a list of scenarios stored in the metadata store

Returns:: list of scenarios
Return type:: List[str]

Bases: TSDataReader

PostgreSQL-based time-series data reader.

connect() → bool

Establishes connection to the PostgreSQL server.

Returns:

Flag indicating successful connection to the PostgreSQL: database

Return type:

bool

disconnect() → None

Closes the connection to the PostgreSQL server.

Returns:: None

list_data_types() → List[str]

Produces the list of data types in PostgreSQL database

Returns:: Data types in database
Return type:: List[str]

list_federates() → List[str]

Produces a list of federations stored in the metadata store

Returns:: list of federations
Return type:: List[str]

list_scenarios() → List[str]

Produces a list of scenarios stored in the metadata store

Returns:: list of scenarios
Return type:: List[str]

Generic PostgreSQL data read

Parameters:

start_time (Optional[float], optional) – Start time (ordinal time in seconds) for requested data. Defaults to None.
duration (Optional[float], optional) – Length of time (in seconds) from the data to read . Defaults to None.
scenario_name (Optional[str | list], optional) – Name(s) of scenario to read. Defaults to None.
federate_name (Optional[str | list], optional) – Name(s) of federate to read. Defaults to None.
data_name (Optional[str | list], optional) – Name(s) of data to read. Defaults to None.
data_type (Optional[str | list], optional) – Data type(s) to read. Defaults to None.

Returns:

requested data

Return type:

pd.DataFrame

class cosim_toolbox.dbms.postgresql_timeseries.PostgreSQLTimeSeriesWriter(*, location: str | None = None, port: int | None = None, database: str | None = None, user: str | None = None, password: str | None = None, analysis_name: str | None = None, use_timescale: bool = False, batch_size: int = 1000, helper: _PostgresConnectionHelper | None = None)

Bases: TSDataWriter

PostgreSQL-based time-series data writer.

connect() → bool

Establishes connection to the PostgreSQL server.

Returns:: Flag indicating success of connection to MongoDB
Return type:: bool

disconnect() → None

Closes the connection to the PostgreSQL server.

Returns:: None

write_records(records: List[TSRecord]) → bool

Writes records (rows) to PostgreSQL database

Parameters:

records (List[TSRecord]) – data to be written to database

Returns:

Flag indicating data was successfully written to: PostgreSQL database.

Return type:

bool

Validation utilities for data management components.

@author Nathan Gray

exception cosim_toolbox.dbms.validation.ValidationError

Bases: ValueError

Custom exception for validation errors.

cosim_toolbox.dbms.validation.safe_name_log(name: str, max_length: int = 50) → str

Create a safe version of a name for logging (truncate if too long).

Parameters:

name (str) – Name to make safe for logging
max_length (int) – Maximum length for log output

Returns:

Safe name for logging

Return type:

str

cosim_toolbox.dbms.validation.validate_connection_string(connection_string: str, expected_analysis: str | None = None) → None

Validate a connection string format.

Parameters:

connection_string (str) – Connection string to validate
expected_analysis (Optional[str]) – Expected URL analysis (e.g., “postgres”)

Raises:

ValidationError – If connection string is invalid

Returns:

None

cosim_toolbox.dbms.validation.validate_database_identifier(name: str, context: str = 'identifier', max_length: int = 63) → None

Validate that a name is safe for use as database identifier (table, schema, etc.).

Parameters:

name (str) – Identifier to validate
context (str) – Context for error messages
max_length (int) – Maximum allowed length (PostgreSQL limit is 63)

Raises:

ValidationError – If identifier is invalid

Returns:

None

cosim_toolbox.dbms.validation.validate_name(name: str, context: str = 'name', max_length: int = 255) → None

Validate that a name is safe for filesystem and database use.

Parameters:

name (str) – Name to validate
context (str) – Context for error messages (e.g., “federation”, “scenario”)
max_length (int) – Maximum allowed length

Raises:

ValidationError – If name is invalid

Returns:

None