plans.databases#

SQLite-based database classes for storing and managing hydro-environmental time series.

The base class DataBase structures a database around two table categories: catalogue tables (variables, flags, sources, statistics) seeded from shipped CSV files, and operational tables (specs, records) populated per project.

A new database is created from a TOML (or JSON) setup file:

[database]
path = "hydro.db"

[specs]
path = "my_specs.json"      # .csv also accepted; format inferred from extension

[catalogs]                  # optional overrides; omitted keys fall back to defaults
variables = "my_vars.csv"

The specs table describes every dataset stored in records. A JSON specs file is a list of objects, one per dataset:

[
  {
    "name": "ppt_monthly_chirps",
    "abstract": "Monthly precipitation from CHIRPS v2.0",
    "method": "spatial average over watershed",
    "timestep": "1M",
    "extent": "watershed",
    "scale": 1,
    "offset": 0,
    "start": "1995-01-01",
    "end": null,
    "variable_name": "precipitation",
    "statistic_name": "sum",
    "source_name": "CHIRPS v2.0"
  }
]

Values in records are stored as stored = actual * scale + offset.

Classes

DataBase(name[, alias])

Base SQLite database with a structured schema.

class plans.databases.DataBase(name, alias=None)[source]#

Bases: MbaE

Base SQLite database with a structured schema.

Manages catalogue and operational tables for hydro-environmental datasets. Child classes inherit and extend this via SQL_DIR and override methods where the schema diverges.

Parameters:

name (str) – database name
alias (str, optional) – short alias

SQL_DIR = PosixPath('/home/runner/work/plans/plans/src/plans/data/sql/base')#

__init__(name, alias=None)[source]#

connect(file_db)[source]#

Open a connection to a SQLite database file.

Enables foreign key enforcement on the connection.

Parameters:: file_db (str or Path) – path to the SQLite file
Returns:: active connection
Return type:: sqlite3.Connection

close()[source]#

Close the active connection and release the file lock.

Sets conn to None after closing.

Returns:: None
Return type:: None

inspect_schema(table=None, quiet=False)[source]#

Print an overview of the database to stdout.

Without arguments, all tables are rendered three per row in a fixed display order: operational tables first (records, specs, flags), then catalogue tables (variables, statistics, sources). When table is given, only that table is shown. Each block shows the table name, row count, and column names with types. Requires an active connection.

Parameters:

table (str, optional) – name of a single table to inspect; if None all tables are shown
quiet (bool) – if True, suppress printing and only return the string

Returns:

formatted schema string

Return type:

str

inspect_specs(quiet=False)[source]#

Print a summary of all specs defined in the database.

Each spec is shown as a compact card with its identifying fields, time range, linear transform parameters, and abstract. Requires an active connection.

Parameters:: quiet (bool) – if True, suppress printing and only return the string
Returns:: formatted specs string
Return type:: str

inspect_records(head=10, tail=None, quiet=False)[source]#

Inspect the records table.

Always includes the table schema. Then shows the first head rows and/or the last tail rows with human-readable timestamps. Set either to None to skip that section. Requires an active connection.

Parameters:

head (int, optional) – number of rows to show from the top; None skips
tail (int, optional) – number of rows to show from the bottom; None skips
quiet (bool) – if True, suppress printing and only return the string

Returns:

formatted records string

Return type:

str

static _fmt_records(df)[source]#

execute_sql(sql)[source]#

Execute SQL from a string or a .sql file path.

Parameters:: sql (str or Path) – SQL text or path to a file with .sql extension
Returns:: None
Return type:: None

query_sql(sql, params=None)[source]#

Execute a SELECT query and return the result as a DataFrame.

Accepts SQL text or a path to a .sql file.

Parameters:

sql (str or Path) – SQL text or path to a file with .sql extension
params (list or tuple, optional) – positional parameters bound to ? placeholders

Returns:

query result; no datetime conversion is applied — epoch columns remain as integers unless the SQL itself handles the transformation

Return type:

pandas.DataFrame

Example — convert epoch to a readable UTC string directly in SQLite:

df = db.query_sql(
    sql="SELECT datetime(datetime, 'unixepoch') AS dt, value FROM records LIMIT 10"
)

static parse_setup(setup)[source]#

Parse a database setup configuration into a plain dict.

Accepts a dict, a path to a TOML file, or a path to a JSON file.

Parameters:: setup (dict, str, or Path) – configuration source
Returns:: configuration dict
Return type:: dict
Raises:: ValueError – if the file extension is not supported

new(config, overwrite=False)[source]#

Create a new database file from a setup configuration.

Expected configuration structure (TOML example):

[database]
path = "path/to/hydro.db"

[specs]
path = "path/to/my_specs.csv"   # or .json — format detected by extension

[catalogs]           # optional overrides; omitted keys fall back to defaults
variables = "path/to/my_variables.csv"

Parameters:

config (dict, str, or Path) – setup dict or path to a TOML/JSON file
overwrite (bool) – delete and recreate the file if it already exists

Returns:

path to the created database file

Return type:

Path

Raises:

FileExistsError – if the database file already exists and overwrite is False
FileNotFoundError – if the specs file does not exist

static _to_epoch(series)[source]#

static _apply_columns_map(df, columns_map)[source]#

_seed_catalog(table, file_path)[source]#

insert_rows(dataframe, on_table, columns_map=None)[source]#

Insert DataFrame rows into a table, keeping only columns that exist in the target.

Parameters:

dataframe (pandas.DataFrame) – source data
on_table (str) – target table name
columns_map (dict, optional) – column mapping {db_col: csv_name_or_index}; values may be a string (source column name) or an int (source column position, 0-based)

Returns:

None

Return type:

None

insert_row(row_dict, on_table, columns_map=None)[source]#

Insert a single row from a plain dict into any table.

Convenience wrapper around insert_rows() for one-row inserts. Epoch conversion is applied automatically for date columns in tables listed in _EPOCH_COLS (e.g. start and end in specs).

Parameters:

row_dict (dict) – row fields as a dict
on_table (str) – target table name
columns_map (dict, optional) – column mapping {db_col: csv_name_or_index}

Returns:

None

Return type:

None

insert_records(dataframe, spec_id, flag=None, columns_map=None, transform_values=True, fill_nan=None)[source]#

Insert records into the records table for a given spec.

spec_id is broadcast to all rows in the batch. The datetime column is converted from text to epoch seconds automatically. The optional linear transform (stored = actual * scale + offset) is resolved from the matching row in specs.

NaN handling in value is controlled by fill_nan: None drops rows with NaN values; any other value fills them before inserting.

Parameters:

dataframe (pandas.DataFrame) – source data; must contain at least datetime and value; flag_value is also required unless flag is provided
spec_id (int) – foreign key referencing the specs table
flag (int, optional) – if given, overwrite or create the flag_value column with this constant
columns_map (dict, optional) – column mapping {db_col: csv_name_or_index}; values may be a string (source column name) or an int (source column position, 0-based)
transform_values (bool) – apply linear transform if True
fill_nan (float or None) – fill value for NaN entries in value; if None, NaN rows are dropped

Returns:

None

Return type:

None

insert_record(row_dict, spec_id, flag=None, columns_map=None, transform_values=True, fill_nan=None)[source]#

Insert a single record from a plain dict.

Convenience wrapper around insert_records() for one-row inserts.

Parameters:

row_dict (dict) – record fields, e.g. {"datetime": "2024-01-01", "value": 12.3}
spec_id (int) – foreign key referencing the specs table
flag (int, optional) – if given, overwrite or create the flag_value column with this constant
columns_map (dict, optional) – column mapping {db_col: csv_name_or_index}; values may be a string (source column name) or an int (source column position, 0-based)
transform_values (bool) – apply linear transform if True
fill_nan (float or None) – fill value for NaN entries in value; if None, NaN rows are dropped

Returns:

None

Return type:

None

load_records(files, spec_id, sep=None, flag=None, columns_map=None, transform_values=True, fill_nan=None)[source]#

Load records from a list of CSV files into the records table.

All files are read and concatenated, then inserted via insert_records(). See that method for column requirements and transform behaviour.

Parameters:

files (list[str or Path]) – paths to CSV files to load
spec_id (int) – foreign key referencing the specs table
sep (str, optional) – column separator; None uses file_csv_sep
flag (int, optional) – if given, overwrite or create the flag_value column with this constant
columns_map (dict, optional) – column mapping {db_col: csv_name_or_index}; values may be a string (source column name) or an int (source column position, 0-based)
transform_values (bool) – apply linear transform if True
fill_nan (float or None) – fill value for NaN entries in value; if None, NaN rows are dropped

Returns:

None

Return type:

None

query_records(query_dict=None, quiet=True)[source]#

Query the records table with optional filters combined with AND logic.

Filters and options are passed as a single dict so the interface stays stable as new keys are added. All keys are optional; an empty or absent dict returns the full table.

Supported keys:

start (str) — lower datetime bound (inclusive); any format parseable by pandas
end (str) — upper datetime bound (inclusive)
flag_value (int or list[int]) — one or more flag values to include
spec_id (int or list[int]) — one or more spec IDs to include
transform_values (bool, default True) — reverse the linear transform so value holds actual values via actual = (stored - offset) / scale

The returned datetime column is always a timezone-aware (UTC) pandas datetime series.

Parameters:

query_dict (dict, optional) – filter and option keys (see above)
quiet (bool) – if False, prints elapsed query time to stdout

Returns:

matching rows with datetime as UTC-aware pandas Timestamps

Return type:

pandas.DataFrame