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]

TODO

Parameters:: collection_type (str) – TODO
Returns:: TODO
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

Parameters:

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

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) – TODO
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) – TODO
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

Parameters:

collection_type (str) – TODO
name (str) – TODO
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) – TODO
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) – TODO
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.

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.

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