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. SemiSpatialDB adds a feature_id to every record, for statistics on regular grids whose features are kept outside the database. SpatialDB is a GeoPackage file that also holds the features, in a features layer of any geometry type. Its children ZonesDB, PathsDB and SitesDB accept polygons, lines or points only, and records point to them by zone_id, path_id or site_id. SitesDB branches into FieldDB (in situ measurements), itself extended by HydrologyDB (river reaches with optional catchments) and DenseFieldDB (a wide layout for dense series), joined in HydrologyDenseDB, and LabDB (laboratory analyses of samples taken at the sites).

Table and column names are lowercase words joined by _; tables are plural, and a foreign key is the singular table name plus the target column (variable_name points to variables.name). In SemiSpatialDB, feature_id is the exception: it points to features outside the database.

A new database is created from a setup, given as a dict or as the path to a TOML or JSON file with the same structure:

TOML

[database]
path = "hydro.db"

[specs]
path = "my_specs.json"

[catalogs]
variables = "my_vars.csv"

JSON

{
  "database": {"path": "hydro.db"},
  "specs": {"path": "my_specs.json"},
  "catalogs": {"variables": "my_vars.csv"}
}

The specs file may be JSON or CSV, told apart by its extension. The catalogs section is optional, and omitted keys fall back to the shipped catalogs. Relative paths follow the working directory, not the setup file. Both formats refuse single backslashes, so Windows paths are written with forward slashes. SpatialDB adds a spatial section.

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 catchment",
    "timestep": "1M",
    "extent": "catchment",
    "scale": 1,
    "offset": 0,
    "start": "1995-01-01",
    "end": null,
    "variable_name": "precipitation",
    "statistic_name": "sum",
    "source_name": "CHIRPS v2.0"
  }
]

Spec names are unique and uppercase (ASCII letters, digits and _, starting with a letter), so they never read like column names. Input names are stripped and uppercased, so ppt_monthly_chirps resolves to PPT_MONTHLY_CHIRPS. Public methods take spec_name (preferred) or spec_id; ids are only stable inside one database file.

Values in records are stored as stored = actual * scale + offset and returned as actual = (stored - offset) / scale. The records table has no uniqueness constraint on purpose: replicates at the same datetime are valid data (eg, three lab analyses on one day). Preventing a batch from being loaded twice is the job of the ingestion script, not of the database.

Each insert call is atomic and commits on its own. Several calls are grouped into one all-or-nothing block with DataBase.transaction():

with db.transaction():
    for name, df in batches.items():
        db.insert_records(dataframe=df, spec_name=name, flag=1)

Every record carries a flag from the flags table. By default, NaN values are not stored. With nan_policy="void" they are stored as NULL with the void flag (0), so a gap reported by the source can be told apart from data that was never loaded; NULL is the only placeholder, so SQL aggregates and pandas skip voids without extra handling. On insert, values outside the variable domain turn regular (1) rows into suspect (2): the value is kept, and a warning reports the count.

Flags of value 0 or below mean “no measured value” (void, not analysed, not applicable, not detected, …) and are the only ones that allow a NULL value; nondetect may instead hold the detection limit. Positive flags always carry a value (regular, suspect, raw, consisted). Censored records hold a limit, not a measurement, so analyses should filter by flag (eg, "flag_value": [1, 4]) rather than average everything.

Domain notation#

The domain column of variables holds the valid range of actual values as one or more intervals joined by U (union). Inside an interval, U separates the lower and upper bounds. [ and ] include the bound, ( and ) exclude it, and an empty bound is unbounded.

Notation	Values accepted	Example use
`[0U)`	`0 <= x`	precipitation
`(0U)`	`0 < x`	strictly positive
`[0U1]`	`0 <= x <= 1`	fractions
`[-1U1]`	`-1 <= x <= 1`	NDVI
`(0U1]`	`0 < x <= 1`	relative humidity
`[-273.15U)`	`-273.15 <= x`	temperature in C
`(U)`	any value	no constraint
`[5U5]`	`x == 5`	a single point
`[-90U-60]U[60U90]`	`-90 <= x <= -60` or `60 <= x <= 90`	polar latitudes

Domains are checked when the variables table is seeded, so a malformed domain fails at database creation.

Classes

`DataBase`(name[, alias])	Base SQLite database with a structured schema.
`DenseFieldDB`(name[, alias])	Field database in a wide layout, for dense and regular series (eg, hourly).
`FieldDB`(name[, alias])	Sites database for in situ measurements (eg, a climate station network).
`HydrologyDB`(name[, alias])	Sites database for water body reaches, with an optional catchment per site.
`HydrologyDenseDB`(name[, alias])	Hydrology database in the wide layout: reaches, optional catchments, dense series.
`LabDB`(name[, alias])	Sites database for laboratory analyses of field samples.
`PathsDB`(name[, alias])	Spatial database of lines (eg, rivers or transects), in the `paths` layer.
`SemiSpatialDB`(name[, alias])	Database whose records carry a `feature_id`.
`SitesDB`(name[, alias])	Spatial database of points (eg, stations or wells), in the `sites` layer.
`SpatialDB`(name[, alias])	GeoPackage database whose records point to features stored in the same file.
`ZonesDB`(name[, alias])	Spatial database of polygons (eg, catchments or municipalities), in the `zones` layer.

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 set SQL_DIR to a folder holding only the .sql files that differ (the rest is found in the parents’ folders, down to base/), extend the class attributes that describe the tables and records (tables, required columns, query keys, row order, wide columns), 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')#

_TABLES = ('variables', 'flags', 'sources', 'statistics', 'specs', 'records')#

_RECORD_COLUMNS = ('datetime', 'value')#

_QUERY_KEYS = ('spec_name', 'spec_id', 'start', 'end', 'flag_value', 'transform_values', 'wide', 'aggfunc', 'with_meta')#

_ORDER_BY = ('spec_id', 'datetime')#

_WIDE_COLUMNS = ('spec_name',)#

_TIME_SQL = 'r.datetime'#

_EPOCH_OUTPUT = ('datetime', 'inserted')#

_WIDE_INDEX = 'datetime'#

_ROW_ORDER = ('id',)#

__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]#

static _read_sql(sql)[source]#

_sql_path(table)[source]#

_create_table(table)[source]#

_prepare_file(cfg, file_db)[source]#

_finalize_file(cfg)[source]#

_prepare_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
Raises:: FileNotFoundError – if sql has a .sql suffix but the file does not exist

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

Raises:

FileNotFoundError – if sql has a .sql suffix but the file does not exist

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.

Creation is all or nothing: on any error the partial file is removed.

The setup has the same structure as a dict or as a file. Besides the sections below, spatial databases need the one in SpatialDB:

TOML

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

[specs]
path = "path/to/my_specs.csv"

[catalogs]
variables = "path/to/my_variables.csv"

JSON

{
  "database": {"path": "path/to/hydro.db"},
  "specs": {"path": "path/to/my_specs.csv"},
  "catalogs": {"variables": "path/to/my_variables.csv"}
}

The catalogs section is optional, and omitted keys fall back to the shipped catalogs.

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 or a table’s SQL file does not exist
ValueError – if a spec has an invalid or duplicated name, unknown columns, unparseable dates, or points to unknown catalogue entries

static _to_epoch(series)[source]#

static _apply_columns_map(df, columns_map)[source]#

_seed_catalog(table, file_path)[source]#

static _as_list(x)[source]#

static _normalize_spec_name(name)[source]#

_spec_lookup()[source]#

_check_specs(df)[source]#

_resolve_spec_ids(spec_name=None, spec_id=None, required=True)[source]#

Resolve spec names or ids to a list of spec ids.

spec_name always wins: if spec_id is also given and points to other specs, a warning is issued and spec_id is ignored.

Parameters:

spec_name (str or list[str], optional) – one or more spec names; case and surrounding whitespace are ignored
spec_id (int or list[int], optional) – one or more spec ids, used only if spec_name is None
required (bool) – raise if both are None; otherwise return None

Returns:

spec ids, or None if nothing was given and not required

Return type:

list[int] or None

Raises:

ValueError – if a name or id is unknown, or both are None while required

_flag_value(name, required=True)[source]#

static _parse_domain(domain)[source]#

Parse a domain string into a list of intervals.

See the module notes for the notation. Single points are allowed ([5U5]); empty intervals raise.

Parameters:: domain (str) – domain notation, eg [0U) or [-90U-60]U[60U90]
Returns:: intervals as (lower, upper, lower_closed, upper_closed); unbounded sides are -inf and inf
Return type:: list[tuple[float, float, bool, bool]]
Raises:: ValueError – if the notation is invalid or an interval is empty

static _in_domain(values, domain)[source]#

_check_variables(df)[source]#

_flag_out_of_domain(df, spec_id)[source]#

transaction()[source]#

Group several inserts into one atomic block.

The block commits on success and rolls back on any exception, which is re-raised. Blocks can be nested: a failed inner block rolls back only its own changes, and only the outermost block commits. Outside a block, each public insert commits on its own.

Do not call execute_sql() inside a block: executescript commits any pending transaction first.

Returns:: this database
Return type:: DataBase
Example:

with db.transaction():
    for name, df in batches.items():
        db.insert_records(dataframe=df, spec_name=name)

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

Insert DataFrame rows into a table.

Columns the target table does not have raise an error, unless drop_extra is True. Rows for specs get their name normalized and checked for duplicates and unknown catalogue entries.

The insert is atomic: a failure leaves no rows. It commits at the end, unless it runs inside transaction().

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)
drop_extra (bool) – drop columns the target table does not have instead of raising

Returns:

None

Return type:

None

Raises:

ValueError – if there are extra columns and drop_extra is False, or if a spec is invalid, duplicated or points to unknown catalogue entries

insert_row(row_dict, on_table, columns_map=None, drop_extra=False)[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=None, spec_name=None, flag=None, columns_map=None, transform_values=True, nan_policy='drop', drop_extra=False, check_domain=True)[source]#

Insert records into the records table for a given spec.

The spec is given by spec_name (preferred) or spec_id and is broadcast to all rows in the batch. If both are given and disagree, spec_name is used and a warning is issued. 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 in value follows nan_policy: "drop" skips those rows; "void" stores them as NULL with the void flag, overriding the batch flag, so a known gap can be told apart from data never loaded. Rows whose flag is already a no-value flag (0 or below, eg notanalysed) are kept as given under both policies.

With check_domain, actual values outside the variable domain (see the module notes on domain notation) turn regular rows into suspect and issue a warning. Values are kept; other flags are left as given.

Inputs are validated before anything is written: unknown flags, unparseable dates and columns records does not have raise an error.

Warning

Spec ids depend on the order specs were created, so the same id can point to different datasets in two database files. Prefer spec_name.

Parameters:

dataframe (pandas.DataFrame) – source data; must contain the columns in _RECORD_COLUMNS (datetime and value in the base class); flag_value is also required unless flag is provided
spec_id (int, optional) – id of the spec; used only if spec_name is None
spec_name (str, optional) – name of the spec (preferred); case is ignored
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
nan_policy (str) – for NaN values with a positive flag, "drop" skips them and "void" stores them as void records
drop_extra (bool) – drop columns records does not have instead of raising
check_domain (bool) – flag values outside the variable domain as suspect

Returns:

None

Return type:

None

Raises:

ValueError – if no spec is given, the spec is unknown, more than one spec is given, a required column is missing, nan_policy is unknown, or a flag, date or column is invalid

_write_records(df, drop_extra=False)[source]#

insert_record(row_dict, spec_id=None, spec_name=None, flag=None, columns_map=None, transform_values=True, nan_policy='drop', drop_extra=False, check_domain=True)[source]#

Insert a single record from a plain dict.

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

Warning

Spec ids depend on the order specs were created, so the same id can point to different datasets in two database files. Prefer spec_name.

Parameters:

row_dict (dict) – record fields, e.g. {"datetime": "2024-01-01", "value": 12.3}
spec_id (int, optional) – id of the spec; used only if spec_name is None
spec_name (str, optional) – name of the spec (preferred); case is ignored
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
nan_policy (str) – for NaN values with a positive flag, "drop" skips them and "void" stores them as void records
drop_extra (bool) – drop columns records does not have instead of raising
check_domain (bool) – flag values outside the variable domain as suspect

Returns:

None

Return type:

None

load_records(files, spec_id=None, spec_name=None, sep=None, flag=None, columns_map=None, transform_values=True, nan_policy='drop', drop_extra=False, check_domain=True)[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.

Warning

Spec ids depend on the order specs were created, so the same id can point to different datasets in two database files. Prefer spec_name.

Parameters:

files (list[str or Path]) – paths to CSV files to load
spec_id (int, optional) – id of the spec; used only if spec_name is None
spec_name (str, optional) – name of the spec (preferred); case is ignored
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
nan_policy (str) – for NaN values with a positive flag, "drop" skips them and "void" stores them as void records
drop_extra (bool) – drop columns records does not have instead of raising
check_domain (bool) – flag values outside the variable domain as suspect

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. Unknown keys raise an error.

Filter keys:

spec_name (str or list[str]) — one or more spec names (preferred); case is ignored; wins over spec_id with a warning if both disagree
spec_id (int or list[int]) — one or more spec ids, used only without spec_name
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

Option keys:

transform_values (bool, default True) — reverse the linear transform so value holds actual values via actual = (stored - offset) / scale
wide (bool, default False) — one column per spec, indexed by datetime
aggfunc (str or callable, default None) — with wide, aggregate replicates at the same datetime (eg, "mean"); None raises on replicates
with_meta (bool, default False) — add variable, units, statistic and source; as columns in long format, as df.attrs["specs"] in wide format

The long format is ordered by spec, then datetime, and carries both spec_name and spec_id. Without spec_name and spec_id, records of all specs are returned. datetime is always a timezone-aware (UTC) pandas datetime.

Warning

Spec ids depend on the order specs were created, so the same id can point to different datasets in two database files. Prefer spec_name.

Parameters:

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

Returns:

matching records, in long or wide format

Return type:

pandas.DataFrame

Raises:

ValueError – if a key is unknown, aggfunc is given without wide, or wide meets replicates without aggfunc

Example:

df = db.query_records(
    query_dict={
        "spec_name": ["PPT_MONTHLY_CHIRPS", "Q_MONTHLY_ANA"],
        "start": "2000-01-01",
        "wide": True,
        "with_meta": True,
    }
)
meta = df.attrs["specs"]

_read_records(query_dict)[source]#

_query_source()[source]#

_query_filters(query_dict)[source]#

Build the WHERE conditions of query_records().

Child classes extend this to add their own filter keys.

Parameters:: query_dict (dict) – validated query keys
Returns:: SQL conditions (combined with AND) and their bound parameters
Return type:: tuple[list[str], list]

_spec_meta(names)[source]#

_to_wide(df, aggfunc=None)[source]#

Pivot long records to a wide table indexed by _WIDE_INDEX (datetime here).

Columns are keyed by _WIDE_COLUMNS, in order of appearance. Replicates on the same keys and datetime raise unless aggfunc is given.

Parameters:

df (pandas.DataFrame) – long-format records from query_records()
aggfunc (str or callable, optional) – pandas aggregation name (eg, "mean") or callable

Returns:

wide table, one column per key combination

Return type:

pandas.DataFrame

Raises:

ValueError – if there are replicates and aggfunc is None

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

Bases: DataBase

Database whose records carry a feature_id.

Features are not stored in the database: feature_id is an integer key (eg, an H3 cell index or a grid cell id) that points to a layer kept elsewhere. Spatial selection happens outside, and the selected ids are passed to query_records().

Only records differs from DataBase; all other tables come from the base schema.

Children rename the key by setting _FEATURE_KEY (eg, zone_id); the record, query, ordering and wide-format hooks follow it.

Parameters:

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

Example:

import geopandas as gpd

cells = gpd.read_file("grid.gpkg", layer="cells")
catchment = gpd.read_file("catchment.gpkg")
ids = gpd.sjoin(cells, catchment, predicate="intersects")["id"].tolist()

df = db.query_records(
    query_dict={"spec_name": "PPT_MONTHLY_CHIRPS", "feature_id": ids}
)
catchment_mean = df.groupby("datetime")["value"].mean()

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

_FEATURE_KEY = 'feature_id'#

_RECORD_COLUMNS = ('datetime', 'value', 'feature_id')#

_QUERY_KEYS = ('spec_name', 'spec_id', 'start', 'end', 'flag_value', 'transform_values', 'wide', 'aggfunc', 'with_meta', 'feature_id')#

_ORDER_BY = ('spec_id', 'feature_id', 'datetime')#

_WIDE_COLUMNS = ('spec_name', 'feature_id')#

classmethod __init_subclass__(**kwargs)[source]#

Swap the inherited feature key for the child’s own in the column hooks.

A child sets only _FEATURE_KEY; the hooks it inherits are renamed, and so is the matching _name key (feature_id to zone_id also turns feature_name into zone_name). Hooks written in the child’s own body are used as written.

_query_filters(query_dict)[source]#

Build the WHERE conditions of query_records(), adding the feature key.

The key (_FEATURE_KEY) takes an int or any iterable of ints (numpy included). The ids are bound as one JSON array, so their number is not limited by the SQLite cap on bound parameters.

Parameters:: query_dict (dict) – validated query keys
Returns:: SQL conditions (combined with AND) and their bound parameters
Return type:: tuple[list[str], list]

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

Bases: SemiSpatialDB

GeoPackage database whose records point to features stored in the same file.

The file holds one layer of features next to the standard tables. Each feature has a generated id, a unique name, a geometry and the attributes declared in the columns map. Records carry feature_id, a real foreign key to the layer, so a record cannot point to a missing feature.

This class stores any geometry type, mixed in one features layer. Its children ZonesDB, PathsDB and SitesDB accept one kind only and rename the layer and key after it (eg, zones and zone_id). A child sets the layer name, the record key and the accepted geometry types; the SQL templates in sql/spatial are filled from them.

The layer is written and extended through geopandas only. Plain SQLite cannot insert into it, because the GeoPackage triggers call functions that only GDAL defines. All other tables work as in DataBase.

The setup adds a spatial section to those of DataBase.new(), and the database path must end in .gpkg:

TOML

[database]
path = "catchments.gpkg"

[specs]
path = "catchments_specs.json"

[spatial]
file = "hydrology.gpkg"
layer = "catchments"

[spatial.columns_map]
name = "CatchmentCode"
catchment_code = "CatchmentCode"
catchment_title = "CatchmentName"

JSON

{
  "database": {"path": "catchments.gpkg"},
  "specs": {"path": "catchments_specs.json"},
  "spatial": {
    "file": "hydrology.gpkg",
    "layer": "catchments",
    "columns_map": {
      "name": "CatchmentCode",
      "catchment_code": "CatchmentCode",
      "catchment_title": "CatchmentName"
    }
  }
}

The file is the source of the features, in any format geopandas reads. The layer picks one layer of a file with several; without it, the first layer is read, with only a warning.

The optional columns_map declares {layer column: source column}, and a source column may be used more than once. Only mapped columns enter the layer. The name key is the unique key of the features, and defaults to a source column called name. Names are stripped text, from text or integer sources. Ids are generated and only stable inside one file, so a source id worth keeping must be mapped as an attribute.

A shapefile holds one layer, so the setup leaves layer out and file points to the .shp, next to its companion files. Shapefile field names stop at 10 characters and the map must use them as read, so a field written as CatchmentCode is mapped as CatchmentC. Without a .prj, the layer has no CRS:

TOML

[spatial]
file = "catchments.shp"

[spatial.columns_map]
name = "CatchmentC"
catchment_title = "CatchmentN"

JSON

{
  "spatial": {
    "file": "catchments.shp",
    "columns_map": {
      "name": "CatchmentC",
      "catchment_title": "CatchmentN"
    }
  }
}

Warning

Tables are STRICT, a feature of SQLite 3.37 (2021). Readers built on an older SQLite (eg, an old QGIS or GDAL) cannot open the file at all.

Parameters:

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

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

_LAYER = 'features'#

_GEOMETRY_TYPES = None#

_PROMOTE_TO_MULTI = False#

_KEY_TABLE = None#

_RESERVED_COLUMNS = ('id', 'geometry')#

_QUERY_KEYS = ('spec_name', 'spec_id', 'start', 'end', 'flag_value', 'transform_values', 'wide', 'aggfunc', 'with_meta', 'feature_id', 'feature_name')#

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

connect(file_db)[source]#

Open a connection to a GeoPackage file.

Keeps the file path, which geopandas needs to read and write the layer.

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

_fill_template(sql_text)[source]#

_create_table(table)[source]#

_prepare_file(cfg, file_db)[source]#

_finalize_file(cfg)[source]#

_layer_geometry(layer)[source]#

_clean_names(names, layer)[source]#

_select_columns(gdf, columns_map, layer)[source]#

_layer_fields(layer)[source]#

_add_columns(gdf, fields, layer)[source]#

_check_attributes(gdf, fields, layer)[source]#

_check_features(gdf, layer)[source]#

_write_layer(gdf, file_db, mode, layer)[source]#

_name_lookup(table)[source]#

_resolve_names(names, table)[source]#

property _key_table#

insert_features(gdf, columns_map=None)[source]#

Append features to the layer.

Names must be new to the layer, and ids are generated. Only mapped columns are read: they must exist in the layer, and layer columns left out are stored as NULL. Geometries are reprojected to the layer CRS. The write goes through geopandas on its own connection, so it cannot run inside a transaction() block.

Parameters:

gdf (geopandas.GeoDataFrame) – source features
columns_map (dict, optional) – column mapping {layer_col: source_col}; name defaults to a source column of that name

Returns:

None

Return type:

None

Raises:

ValueError – if names are missing, repeated or taken, a mapped column is reserved, missing or not fit for the layer, geometry types are not accepted, or gdf has no CRS while the layer has one
RuntimeError – if called inside an open transaction

_check_no_transaction(method)[source]#

static _match_crs(gdf, crs)[source]#

_append_layer(gdf, layer, add_columns=False)[source]#

_read_layer(names, layer)[source]#

get_features(names=None)[source]#

Read the layer as a GeoDataFrame.

Parameters:: names (str or list[str], optional) – one or more layer names; None returns the whole layer
Returns:: layer rows with id as the first column (the record key values)
Return type:: geopandas.GeoDataFrame
Raises:: ValueError – if a name is unknown

_prepare_records(df)[source]#

_query_filters(query_dict)[source]#

Build the WHERE conditions of query_records(), adding the name key.

The name key (eg, zone_name) takes one or more names, resolved to ids through the table the records point to; unknown names raise.

Parameters:: query_dict (dict) – validated query keys
Returns:: SQL conditions (combined with AND) and their bound parameters
Return type:: tuple[list[str], list]

_ORDER_BY = ('spec_id', 'feature_id', 'datetime')#

_RECORD_COLUMNS = ('datetime', 'value', 'feature_id')#

_WIDE_COLUMNS = ('spec_name', 'feature_id')#

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

Bases: SpatialDB

Spatial database of polygons (eg, catchments or municipalities), in the zones layer.

Records carry zone_id, and queries also accept zone_name. Polygons are written as multi-polygons, so single and multi-part zones share one layer.

Parameters:

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

_LAYER = 'zones'#

_FEATURE_KEY = 'zone_id'#

_GEOMETRY_TYPES = ('Polygon', 'MultiPolygon')#

_PROMOTE_TO_MULTI = True#

_ORDER_BY = ('spec_id', 'zone_id', 'datetime')#

_QUERY_KEYS = ('spec_name', 'spec_id', 'start', 'end', 'flag_value', 'transform_values', 'wide', 'aggfunc', 'with_meta', 'zone_id', 'zone_name')#

_RECORD_COLUMNS = ('datetime', 'value', 'zone_id')#

_WIDE_COLUMNS = ('spec_name', 'zone_id')#

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

Bases: SpatialDB

Spatial database of lines (eg, rivers or transects), in the paths layer.

Records carry path_id, and queries also accept path_name. Lines are written as multi-lines, so single and multi-part paths share one layer.

Parameters:

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

_LAYER = 'paths'#

_FEATURE_KEY = 'path_id'#

_GEOMETRY_TYPES = ('LineString', 'MultiLineString')#

_PROMOTE_TO_MULTI = True#

_ORDER_BY = ('spec_id', 'path_id', 'datetime')#

_QUERY_KEYS = ('spec_name', 'spec_id', 'start', 'end', 'flag_value', 'transform_values', 'wide', 'aggfunc', 'with_meta', 'path_id', 'path_name')#

_RECORD_COLUMNS = ('datetime', 'value', 'path_id')#

_WIDE_COLUMNS = ('spec_name', 'path_id')#

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

Bases: SpatialDB

Spatial database of points (eg, stations or wells), in the sites layer.

Records carry site_id, and queries also accept site_name.

Parameters:

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

_LAYER = 'sites'#

_FEATURE_KEY = 'site_id'#

_GEOMETRY_TYPES = ('Point',)#

_ORDER_BY = ('spec_id', 'site_id', 'datetime')#

_QUERY_KEYS = ('spec_name', 'spec_id', 'start', 'end', 'flag_value', 'transform_values', 'wide', 'aggfunc', 'with_meta', 'site_id', 'site_name')#

_RECORD_COLUMNS = ('datetime', 'value', 'site_id')#

_WIDE_COLUMNS = ('spec_name', 'site_id')#

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

Bases: SitesDB

Sites database for in situ measurements (eg, a climate station network).

The instrument measures at the site, so records point straight to it by site_id. It behaves as SitesDB for now; the name marks the branch for features specific to field data.

Parameters:

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

_ORDER_BY = ('spec_id', 'site_id', 'datetime')#

_QUERY_KEYS = ('spec_name', 'spec_id', 'start', 'end', 'flag_value', 'transform_values', 'wide', 'aggfunc', 'with_meta', 'site_id', 'site_name')#

_RECORD_COLUMNS = ('datetime', 'value', 'site_id')#

_WIDE_COLUMNS = ('spec_name', 'site_id')#

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

Bases: FieldDB

Sites database for water body reaches, with an optional catchment per site.

A site is a point that stands for a reach (eg, its gauge or outlet). The catchments layer holds the polygon draining to it, in the file next to sites, with attributes of that area. Each catchment takes the name of its site. Every catchment needs a site, a site has at most one catchment, and a site may have none, so catchments can be added as they become available.

Records carry site_id. An areal series, such as catchment rainfall, is a record of the site whose spec tells it apart (eg, extent is catchment).

The setup takes an optional catchments section, read like spatial:

TOML

[database]
path = "reaches.gpkg"

[specs]
path = "reaches_specs.json"

[spatial]
file = "gauges.gpkg"
layer = "gauges"

[spatial.columns_map]
name = "GaugeCode"
gauge_title = "GaugeName"

[catchments]
file = "hydrology.gpkg"
layer = "catchments"
skip_unmatched = true

[catchments.columns_map]
name = "GaugeCode"
area_km2 = "AreaKm2"

JSON

{
  "database": {"path": "reaches.gpkg"},
  "specs": {"path": "reaches_specs.json"},
  "spatial": {
    "file": "gauges.gpkg",
    "layer": "gauges",
    "columns_map": {
      "name": "GaugeCode",
      "gauge_title": "GaugeName"
    }
  },
  "catchments": {
    "file": "hydrology.gpkg",
    "layer": "catchments",
    "skip_unmatched": true,
    "columns_map": {
      "name": "GaugeCode",
      "area_km2": "AreaKm2"
    }
  }
}

In the map, name is the source column holding the site names. Features naming no site are skipped with a warning, unless skip_unmatched is false; no match at all is an error, as it usually points to a wrong column. Without the section the layer starts empty, with name alone, in the CRS of the sites. The columns set at creation are kept unless an append asks for more.

Parameters:

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

Example:

# a polygon drawn in a GIS: one feature, no useful columns
db.append_catchment(
    name="CODE002",
    file="drawn.gpkg",
    attributes={"area_km2": 120.5},
)
# one feature of a larger file, with attributes from its columns
db.append_catchment(
    name="CODE003",
    file="hydrology.gpkg",
    layer="catchments",
    columns_map={"name": "GaugeCode", "area_km2": "AreaKm2"},
)
db.missing_catchments()

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

_CATCHMENTS = 'catchments'#

_CATCHMENT_TYPES = ('Polygon', 'MultiPolygon')#

_SOURCE_KEY = '_site_key'#

_layer_geometry(layer)[source]#

_prepare_file(cfg, file_db)[source]#

_finalize_file(cfg)[source]#

_source_keys(names)[source]#

_match_sites(gdf, columns_map, sites, skip_unmatched)[source]#

_pick_feature(name, file, layer, columns_map)[source]#

append_catchment(name, file, layer=None, columns_map=None, attributes=None, add_columns=False, overwrite=False)[source]#

Append the catchment of one site from a spatial file.

The geometry comes from file. With name in columns_map, that source column picks the feature; without it, the file must hold exactly one feature, as in a polygon just drawn. Attributes may come from other columns_map entries or from attributes, and layer columns left out are stored as NULL. Geometries are reprojected to the layer CRS.

The write goes through geopandas on its own connection, so it cannot run inside a transaction() block.

Parameters:

name (str) – site name
file (str or Path) – source file, in any format geopandas reads
layer (str, optional) – source layer, for files with several
columns_map (dict, optional) – column mapping {layer_col: source_col}
attributes (dict, optional) – attribute values {layer_col: value}
add_columns (bool) – add attribute columns the layer does not have, typed from their values; existing rows stay NULL
overwrite (bool) – replace the catchment of the site, if any; its own id changes, which records do not depend on

Returns:

None

Return type:

None

Raises:

ValueError – if the site is unknown or has a catchment, the source does not give one feature, a column is unknown, reserved or set twice, or a value or geometry does not fit the layer
RuntimeError – if called inside an open transaction

insert_catchments(gdf, columns_map=None, skip_unmatched=True, add_columns=False)[source]#

Append several catchments, eg generated in Python.

In the map, name is the source column holding the site names. Layer columns left out are stored as NULL. Geometries are reprojected to the layer CRS. All rows are checked before the write, and sites that already have a catchment raise; use append_catchment() to replace one.

Parameters:

gdf (geopandas.GeoDataFrame) – source features
columns_map (dict, optional) – column mapping {layer_col: source_col}; name defaults to a source column of that name
skip_unmatched (bool) – skip features naming no site, with a warning, instead of raising
add_columns (bool) – add attribute columns the layer does not have, typed from their values; existing rows stay NULL

Returns:

None

Return type:

None

Raises:

ValueError – if no feature names a site, a site has a catchment or is repeated, or a column or geometry does not fit the layer
RuntimeError – if called inside an open transaction

get_catchments(names=None)[source]#

Read the catchments layer as a GeoDataFrame.

Parameters:: names (str or list[str], optional) – one or more site names; None returns the whole layer
Returns:: layer rows; id belongs to the catchment, not to the site
Return type:: geopandas.GeoDataFrame
Raises:: ValueError – if a name has no catchment

missing_catchments()[source]#

List the sites without a catchment.

Returns:: site names
Return type:: list[str]

_ORDER_BY = ('spec_id', 'site_id', 'datetime')#

_QUERY_KEYS = ('spec_name', 'spec_id', 'start', 'end', 'flag_value', 'transform_values', 'wide', 'aggfunc', 'with_meta', 'site_id', 'site_name')#

_RECORD_COLUMNS = ('datetime', 'value', 'site_id')#

_WIDE_COLUMNS = ('spec_name', 'site_id')#

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

Bases: FieldDB

Field database in a wide layout, for dense and regular series (eg, hourly).

records holds one row per site and time, with one column per spec, instead of one row per value. The columns are created from the specs, each named after its spec in lowercase. Rows are stored by site and then time, so a station series reads fast, while one variable over all stations scans the table.

Values are regular unless record_flags says otherwise, so flags take no space until a value needs one. query_records() keeps its keys and output, without the id and inserted columns of the long layout.

Each spec loads on its own, in any order, into its column: a value fills the row of its site and time, which is created if missing. There are no replicates: a second value for the same spec, site and time raises. Specs cannot be added after creation yet.

Specs may mix timesteps. A coarse value sits on one row, and the rows between two values keep an empty cell, about one byte each: a monthly spec next to daily ones costs about 30 bytes per value, close to the long layout. This suits a few coarse specs next to many fine ones (eg, yearly land use next to daily series). Many coarse specs, or a step far finer than the rest (eg, monthly next to hourly), are better kept in a separate database. Coarse values should share the timestamps of the fine rows (eg, the first day of the month at 00:00), or each adds a row of its own.

records and record_flags are WITHOUT ROWID tables, which GDAL cannot open, so a GIS shows the sites and the catalogs but reads no records; they are read through query_records() or SQL. Queries build long rows in pandas, so one variable over all sites and all times takes memory in proportion; filter by site or time for large files.

Parameters:

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

Example:

db.insert_wide(
    dataframe=pd.DataFrame(
        {
            "site_name": ["A001", "A001"],
            "datetime": ["2019-01-01 00:00", "2019-01-01 01:00"],
            "PPT_HOURLY": [1.0, 0.0],
            "TAS_HOURLY": [18.5, 18.4],
        }
    )
)
df = db.query_records(query_dict={"site_name": "A001", "wide": True})

_TABLES = ('variables', 'flags', 'sources', 'statistics', 'specs')#

_EPOCH_OUTPUT = ('datetime',)#

_ROW_ORDER = ('site_id', 'datetime')#

_spec_columns()[source]#

_finalize_file(cfg)[source]#

_write_records(df, drop_extra=False)[source]#

insert_wide(dataframe, transform_values=True, check_domain=True)[source]#

Insert a wide table, the fast path for bulk loads.

Columns are datetime, the site as site_id or site_name, and one column per spec, named after it in any case. Empty cells store nothing. Values outside the domain of their variable become suspect, and all others regular.

Parameters:

dataframe (pandas.DataFrame) – one row per site and time
transform_values (bool) – apply the spec transform; False means values are already stored numbers
check_domain (bool) – flag values outside the variable domain; it reads actual values, so it needs transform_values

Returns:

None

Return type:

None

Raises:

ValueError – if a column names no spec, the site or datetime is missing, or a site and time already exist

_read_records(query_dict)[source]#

_ORDER_BY = ('spec_id', 'site_id', 'datetime')#

_QUERY_KEYS = ('spec_name', 'spec_id', 'start', 'end', 'flag_value', 'transform_values', 'wide', 'aggfunc', 'with_meta', 'site_id', 'site_name')#

_RECORD_COLUMNS = ('datetime', 'value', 'site_id')#

_WIDE_COLUMNS = ('spec_name', 'site_id')#

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

Bases: HydrologyDB, DenseFieldDB

Hydrology database in the wide layout: reaches, optional catchments, dense series.

It joins HydrologyDB and DenseFieldDB. Sites, catchments, the catchments setup section and the catchment methods work as in the first; records, insert_wide() and queries as in the second. It suits catchment datasets with many specs over the same days (eg, rain, streamflow and evapotranspiration from several sources), where most rows are filled.

Specs may mix timesteps, as in DenseFieldDB, which suits daily series with a few pentad, monthly or yearly specs (eg, land use shares).

Parameters:

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

_ORDER_BY = ('spec_id', 'site_id', 'datetime')#

_QUERY_KEYS = ('spec_name', 'spec_id', 'start', 'end', 'flag_value', 'transform_values', 'wide', 'aggfunc', 'with_meta', 'site_id', 'site_name')#

_RECORD_COLUMNS = ('datetime', 'value', 'site_id')#

_WIDE_COLUMNS = ('spec_name', 'site_id')#

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

Bases: SitesDB

Sites database for laboratory analyses of field samples.

Data follows the chain sites -> samples -> mixtures -> records. A sample (eg, a water bottle) is taken at a site at a sampling time. The lab analyses a mixture of one or more samples; most mixtures hold a single sample, and a sample may enter several mixtures. Records carry mixture_id and the analysis time, which may be unknown (NULL). There is no aliquot table: the mixture stands for whatever the lab analysed.

Queries filter and sort by the sampling time (sample_datetime, the latest sample of the mixture) and return both times; the gap between them is how long the sample was stored. Filters by site_id or site_name match records whose mixture holds a sample from those sites, so a blend of two sites shows up under both.

Identical results are valid replicates (eg, a triplicate “not detected”); only the record id tells them apart.

Parameters:

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

Example:

db.insert_samples(
    dataframe=pd.DataFrame(
        {"name": ["B1", "B2"], "site_name": ["Gauge 1", "Gauge 2"],
         "datetime": ["2024-03-01 12:00", "2024-03-01 15:00"]}
    )
)
# one mixture per sample, named after it
db.insert_mixtures(dataframe=pd.DataFrame({"sample_name": ["B1", "B2"]}))
db.insert_records(
    dataframe=pd.DataFrame(
        {"mixture_name": ["B1", "B2"], "value": [7.1, 6.8],
         "datetime": ["2024-03-04", None]}
    ),
    spec_name="PH_LAB",
    flag=1,
)
df = db.query_records(query_dict={"site_name": "Gauge 1"})

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

_TABLES = ('variables', 'flags', 'sources', 'statistics', 'specs', 'samples', 'mixtures', 'mixture_samples', 'records')#

_FEATURE_KEY = 'mixture_id'#

_KEY_TABLE = 'mixtures'#

_QUERY_KEYS = ('spec_name', 'spec_id', 'start', 'end', 'flag_value', 'transform_values', 'wide', 'aggfunc', 'with_meta', 'mixture_id', 'mixture_name', 'site_id', 'site_name')#

_ORDER_BY = ('spec_id', 'sample_datetime', 'mixture_id', 'datetime')#

_TIME_SQL = 'm.sample_datetime'#

_EPOCH_OUTPUT = ('datetime', 'inserted', 'sample_datetime')#

_WIDE_INDEX = 'sample_datetime'#

insert_samples(dataframe)[source]#

Insert field samples.

Parameters:: dataframe (pandas.DataFrame) – one row per sample with name, datetime (sampling time) and site_id or site_name; alias and abstract are optional
Returns:: None
Return type:: None
Raises:: ValueError – if a required column is missing or a site is unknown

insert_mixtures(dataframe)[source]#

Insert mixtures with the samples they hold.

Each row links one sample to one mixture, so a blend of three samples takes three rows with the same mixture name. Without a name column, each sample becomes its own mixture, named after the sample (the common case of no mixing).

Parameters:: dataframe (pandas.DataFrame) – rows with sample_name or sample_id, and optional name (mixture) and abstract (the first one per mixture is kept)
Returns:: None
Return type:: None
Raises:: ValueError – if samples are missing or unknown

_query_source()[source]#

_query_filters(query_dict)[source]#

Build the WHERE conditions of query_records(), adding site keys.

site_id and site_name match records whose mixture holds at least one sample from those sites.

Parameters:: query_dict (dict) – validated query keys
Returns:: SQL conditions (combined with AND) and their bound parameters
Return type:: tuple[list[str], list]

_RECORD_COLUMNS = ('datetime', 'value', 'mixture_id')#

_WIDE_COLUMNS = ('spec_name', 'mixture_id')#