database.db_manager module

Database manager for storing and querying simulation data (SQLite).

class database.db_manager.DatabaseManager(db_path: str)

Bases: object

Manages SQLite database operations for the simulator.

Uses in-memory caches for source_id and metric_id to avoid repeated lookups.

clear_caches()

Clear in-memory source_id and metric_id caches.

clear_records(table: str, source_id: int = None, from_time: str = None)

Delete records from the table, optionally by source_id and/or from_time.

Parameters:

• table – One of ‘measurements’, ‘forecast’, ‘scheduling’.

• source_id – If set, delete only records for this source.

• from_time – If set, delete only records with timestamp >= from_time (ISO string).

close()

Close the persistent database connection if open.

connect()

Open a persistent database connection and set row_factory to sqlite3.Row.

delete_metric(metric_id: int)

Delete the metric row with the given metric_id.

delete_source(source_id: int)

Delete the source row with the given source_id.

get_all_metrics(data_type: str = None)

Return all metric rows, optionally filtered by data_type.

get_all_sources(source_type: str = None)

Return all source rows, optionally filtered by source_type.

get_connection()

Context manager: yields a connection with foreign keys on; commits on success, rolls back on exception.

get_metric(metric_id: int = None, metric_name: str = None)

Return one metric by metric_id or metric_name; at least one must be given.

Returns:

sqlite3.Row or None if not found.

get_metric_id(metric_name: str) → int

Return metric_id for the given metric name. Uses caching. The metric must already exist (e.g. from DEFAULT_METRICS or manual creation).

Parameters:

metric_name – Metric name (e.g. ‘power_active’, ‘soc’).

Returns:

The metric_id.

Raises:

ValueError – If no metric with that name exists.

get_or_create_source(source_name: str, source_type: str) → int

Return source_id for the given source; insert the source if it does not exist. Results are cached to avoid repeated lookups.

Parameters:

• source_name – Unique source name (e.g. ‘port’, ‘SeaBreeze’, ‘FastCharger_A’).

• source_type – Type (e.g. ‘port’, ‘boat’, ‘charger’, ‘pv’, ‘bess’, ‘weather’).

Returns:

The source_id.

get_records(table: str, source_id: int = None, metric_id: int = None, start_time: str = None, end_time: str = None)

Return records from the table with optional filters.

Parameters:

• table – One of ‘measurements’, ‘forecast’, ‘scheduling’.

• source_id – Optional filter by source_id.

• metric_id – Optional filter by metric_id.

• start_time – Optional inclusive start timestamp (ISO string).

• end_time – Optional inclusive end timestamp (ISO string).

Returns:

List of rows (sqlite3.Row), ordered by timestamp.

get_source(source_id: int = None, source_name: str = None)

Return one source by source_id or source_name; at least one must be given.

Returns:

sqlite3.Row or None if not found.

initialize_default_metrics()

Insert all DEFAULT_METRICS into the metric table (INSERT OR IGNORE).

initialize_schema()

Create all tables (source, metric, measurements, forecast, scheduling) and their indexes.

save_metric(metric_name: str, unit: str, data_type: str) → int

Insert a metric row.

Parameters:

• metric_name – Unique name (e.g. ‘power’, ‘soc’, ‘temperature’).

• unit – Unit string (e.g. ‘kW’, ‘%’, ‘°C’).

• data_type – One of ‘float’, ‘int’, ‘str’, etc.

Returns:

The inserted metric_id.

save_record(table: str, timestamp: str, source_id: int, metric_id: int, value: str) → int

Insert a single record into the given table.

Parameters:

• table – One of ‘measurements’, ‘forecast’, ‘scheduling’.

• timestamp – ISO-format UTC timestamp string.

• source_id – Foreign key to source.

• metric_id – Foreign key to metric.

• value – Value stored as string.

Returns:

The measurement_id of the inserted row.

save_records_batch(table: str, records: List[Tuple[str, int, int, str]])

Insert multiple records in one transaction.

Parameters:

• table – One of ‘measurements’, ‘forecast’, ‘scheduling’.

• records – List of (timestamp, source_id, metric_id, value) tuples.

save_source(source_name: str, source_type: str) → int

Insert a source row.

Parameters:

• source_name – Unique name (e.g. ‘boat_001’, ‘charger_01’).

• source_type – Type (e.g. ‘boat’, ‘charger’, ‘battery’).

Returns:

The inserted source_id.