database package

Submodules

Module contents

Database layer for the port simulator.

Provides DatabaseManager for SQLite storage of measurements, forecasts, and scheduling data.

class database.DatabaseManager(db_path: str)[source]

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()[source]

Clear in-memory source_id and metric_id caches.

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

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()[source]

Close the persistent database connection if open.

connect()[source]

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

delete_metric(metric_id: int)[source]

Delete the metric row with the given metric_id.

delete_source(source_id: int)[source]

Delete the source row with the given source_id.

get_all_metrics(data_type: str = None)[source]

Return all metric rows, optionally filtered by data_type.

get_all_sources(source_type: str = None)[source]

Return all source rows, optionally filtered by source_type.

get_connection()[source]

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)[source]

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[source]

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[source]

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)[source]

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)[source]

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()[source]

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

initialize_schema()[source]

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

save_metric(metric_name: str, unit: str, data_type: str) → int[source]

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[source]

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]])[source]

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[source]

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.