494 lines
18 KiB
Python
494 lines
18 KiB
Python
"""
|
|
SQLite storage layer for aerodrome meteorological analytics.
|
|
|
|
Tables:
|
|
observations — Hourly analytic time-series (10 float variables per timestamp).
|
|
outlier_log — Audit trail of values treated by physical-limit enforcement.
|
|
aerodromes — Catalog of aerodromes available on the ICEA/DECEA website.
|
|
|
|
Example:
|
|
>>> from db import get_connection, ensure_schema, upsert_analytics, query_analytics
|
|
>>> conn = get_connection(Path("met.db"))
|
|
>>> ensure_schema(conn)
|
|
>>> upsert_analytics(conn, "SBGR", anl_df)
|
|
>>> df = query_analytics(conn, "SBGR", "2024-01-01", "2024-12-31")
|
|
"""
|
|
|
|
import sqlite3
|
|
from datetime import date, datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
import pandas as pd
|
|
|
|
# ── Analytic columns ──────────────────────────────────────────────────────────
|
|
ANL_COLS: list[str] = [
|
|
"T", "Td", "UR", "QNH", "WS", "WG", "WD", "VIS", "TETO", "PREC"
|
|
]
|
|
|
|
# ── Physical limits per variable ──────────────────────────────────────────────
|
|
PHYSICAL_LIMITS: dict[str, tuple[float, float]] = {
|
|
"T": (-25.0, 55.0), # °C — Brazilian climate extremes
|
|
"Td": (-30.0, 40.0), # °C — dew point
|
|
"UR": ( 0.0, 100.0), # %
|
|
"QNH": (940.0, 1060.0), # hPa — absolute world records
|
|
"WS": ( 0.0, 200.0), # kt
|
|
"WG": ( 0.0, 250.0), # kt — Cat-5 hurricane gust ≈ 175 kt
|
|
"WD": ( 0.0, 360.0), # °
|
|
"VIS": ( 0.0, 9999.0), # dam
|
|
"TETO": ( 0.0, 9999.0), # dam
|
|
"PREC": ( 0.0, 500.0), # mm — world record hourly ≈ 300 mm/h
|
|
}
|
|
|
|
# ── SQL schema ────────────────────────────────────────────────────────────────
|
|
_SCHEMA = """
|
|
CREATE TABLE IF NOT EXISTS observations (
|
|
aerodrome TEXT NOT NULL,
|
|
dt TEXT NOT NULL,
|
|
T REAL, Td REAL, UR REAL, QNH REAL,
|
|
WS REAL, WG REAL, WD REAL,
|
|
VIS REAL, TETO REAL, PREC REAL,
|
|
PRIMARY KEY (aerodrome, dt)
|
|
);
|
|
CREATE INDEX IF NOT EXISTS idx_aero_dt ON observations(aerodrome, dt);
|
|
|
|
CREATE TABLE IF NOT EXISTS outlier_log (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
aerodrome TEXT NOT NULL,
|
|
dt TEXT NOT NULL,
|
|
variable TEXT NOT NULL,
|
|
orig_value REAL NOT NULL,
|
|
treatment TEXT NOT NULL DEFAULT 'SET_NULL',
|
|
reason TEXT,
|
|
applied_at TEXT NOT NULL
|
|
);
|
|
CREATE INDEX IF NOT EXISTS idx_outlier_aero ON outlier_log(aerodrome, variable);
|
|
|
|
CREATE TABLE IF NOT EXISTS aerodromes (
|
|
icao TEXT PRIMARY KEY,
|
|
nome TEXT,
|
|
uf TEXT,
|
|
updated_at TEXT
|
|
);
|
|
"""
|
|
|
|
|
|
# ── Connection ────────────────────────────────────────────────────────────────
|
|
def get_connection(db_path: Path) -> sqlite3.Connection:
|
|
"""Opens (or creates) a SQLite database with WAL mode enabled.
|
|
|
|
Args:
|
|
db_path: Filesystem path to the ``.db`` file. Parent directories are
|
|
created automatically if they do not exist.
|
|
|
|
Returns:
|
|
An open :class:`sqlite3.Connection` configured with WAL journal mode,
|
|
NORMAL synchronous writes, and a 32 MB page cache.
|
|
"""
|
|
db_path.parent.mkdir(parents=True, exist_ok=True)
|
|
conn = sqlite3.connect(str(db_path), check_same_thread=False)
|
|
conn.execute("PRAGMA journal_mode=WAL")
|
|
conn.execute("PRAGMA synchronous=NORMAL")
|
|
conn.execute("PRAGMA cache_size=-32000")
|
|
return conn
|
|
|
|
|
|
def ensure_schema(conn: sqlite3.Connection) -> None:
|
|
"""Creates all tables and indices if they do not already exist.
|
|
|
|
Args:
|
|
conn: Open SQLite connection.
|
|
"""
|
|
conn.executescript(_SCHEMA)
|
|
conn.commit()
|
|
|
|
|
|
# ── Physical-limit clipping ───────────────────────────────────────────────────
|
|
def _clip_to_limits(df: pd.DataFrame) -> pd.DataFrame:
|
|
"""Sets values outside physical limits to NaN (no audit log entry).
|
|
|
|
Args:
|
|
df: DataFrame whose columns may include any subset of ANL_COLS.
|
|
|
|
Returns:
|
|
Copy of *df* with out-of-range values replaced by ``float("nan")``.
|
|
"""
|
|
df = df.copy()
|
|
for col, (lo, hi) in PHYSICAL_LIMITS.items():
|
|
if col in df.columns:
|
|
mask = df[col].notna() & ((df[col] < lo) | (df[col] > hi))
|
|
df.loc[mask, col] = float("nan")
|
|
return df
|
|
|
|
|
|
# ── Upsert analytics ──────────────────────────────────────────────────────────
|
|
def upsert_analytics(
|
|
conn: sqlite3.Connection,
|
|
aerodrome: str,
|
|
anl_df: pd.DataFrame,
|
|
) -> int:
|
|
"""Inserts or replaces rows in the observations table.
|
|
|
|
Physical limits are applied automatically before insertion; out-of-range
|
|
values are silently set to NULL.
|
|
|
|
Args:
|
|
conn: Open SQLite connection with schema already applied.
|
|
aerodrome: ICAO code of the aerodrome (e.g. ``"SBGR"``).
|
|
anl_df: DataFrame with a ``_dt`` column (datetime-like) and any subset
|
|
of :data:`ANL_COLS` as float columns.
|
|
|
|
Returns:
|
|
Number of rows processed (inserted or replaced).
|
|
"""
|
|
if anl_df.empty:
|
|
return 0
|
|
|
|
anl_df = _clip_to_limits(anl_df)
|
|
|
|
cols_present = [c for c in ANL_COLS if c in anl_df.columns]
|
|
placeholders = ", ".join(["?"] * (2 + len(cols_present)))
|
|
col_names = ", ".join(["aerodrome", "dt"] + cols_present)
|
|
sql = (
|
|
f"INSERT OR REPLACE INTO observations ({col_names}) VALUES ({placeholders})"
|
|
)
|
|
|
|
def _dt_str(v: object) -> str:
|
|
if isinstance(v, (datetime, pd.Timestamp)):
|
|
return v.strftime("%Y-%m-%d %H:%M:%S")
|
|
return str(v)[:19]
|
|
|
|
rows = []
|
|
for _, row in anl_df.iterrows():
|
|
vals: list = [aerodrome, _dt_str(row["_dt"])]
|
|
for c in cols_present:
|
|
v = row[c]
|
|
vals.append(None if pd.isna(v) else float(v))
|
|
rows.append(tuple(vals))
|
|
|
|
with conn:
|
|
conn.executemany(sql, rows)
|
|
return len(rows)
|
|
|
|
|
|
# ── Physical-limit repair on existing rows ────────────────────────────────────
|
|
def apply_physical_limits(
|
|
conn: sqlite3.Connection,
|
|
aerodrome: Optional[str] = None,
|
|
) -> int:
|
|
"""Audits and corrects out-of-range values already stored in the database.
|
|
|
|
Every corrected value is logged in ``outlier_log`` before being set to
|
|
NULL. Pass ``aerodrome=None`` to process all aerodromes.
|
|
|
|
Args:
|
|
conn: Open SQLite connection with schema already applied.
|
|
aerodrome: ICAO code to restrict processing, or ``None`` for all.
|
|
|
|
Returns:
|
|
Total number of outlier records created and corrected.
|
|
"""
|
|
now = datetime.now(timezone.utc).isoformat()
|
|
total = 0
|
|
aero_filter = "AND aerodrome=?" if aerodrome else ""
|
|
aero_params_suffix = (aerodrome,) if aerodrome else ()
|
|
|
|
for col, (lo, hi) in PHYSICAL_LIMITS.items():
|
|
rows = conn.execute(
|
|
f"SELECT aerodrome, dt, {col} FROM observations "
|
|
f"WHERE {col} IS NOT NULL AND {col} < ? {aero_filter}",
|
|
(lo,) + aero_params_suffix,
|
|
).fetchall()
|
|
if rows:
|
|
conn.executemany(
|
|
"INSERT INTO outlier_log"
|
|
"(aerodrome,dt,variable,orig_value,treatment,reason,applied_at)"
|
|
" VALUES(?,?,?,?,?,?,?)",
|
|
[(r[0], r[1], col, r[2], "SET_NULL", f"below_limit:{lo}", now)
|
|
for r in rows],
|
|
)
|
|
total += len(rows)
|
|
|
|
rows = conn.execute(
|
|
f"SELECT aerodrome, dt, {col} FROM observations "
|
|
f"WHERE {col} IS NOT NULL AND {col} > ? {aero_filter}",
|
|
(hi,) + aero_params_suffix,
|
|
).fetchall()
|
|
if rows:
|
|
conn.executemany(
|
|
"INSERT INTO outlier_log"
|
|
"(aerodrome,dt,variable,orig_value,treatment,reason,applied_at)"
|
|
" VALUES(?,?,?,?,?,?,?)",
|
|
[(r[0], r[1], col, r[2], "SET_NULL", f"above_limit:{hi}", now)
|
|
for r in rows],
|
|
)
|
|
total += len(rows)
|
|
|
|
where_clause = f"WHERE aerodrome='{aerodrome}'" if aerodrome else ""
|
|
conn.execute(f"""
|
|
UPDATE observations SET
|
|
T = CASE WHEN T IS NOT NULL AND (T < -25 OR T > 55 ) THEN NULL ELSE T END,
|
|
Td = CASE WHEN Td IS NOT NULL AND (Td < -30 OR Td > 40 ) THEN NULL ELSE Td END,
|
|
UR = CASE WHEN UR IS NOT NULL AND (UR < 0 OR UR > 100 ) THEN NULL ELSE UR END,
|
|
QNH = CASE WHEN QNH IS NOT NULL AND (QNH < 940 OR QNH > 1060) THEN NULL ELSE QNH END,
|
|
WS = CASE WHEN WS IS NOT NULL AND (WS < 0 OR WS > 200 ) THEN NULL ELSE WS END,
|
|
WG = CASE WHEN WG IS NOT NULL AND (WG < 0 OR WG > 250 ) THEN NULL ELSE WG END,
|
|
WD = CASE WHEN WD IS NOT NULL AND (WD < 0 OR WD > 360 ) THEN NULL ELSE WD END,
|
|
VIS = CASE WHEN VIS IS NOT NULL AND (VIS < 0 OR VIS > 9999) THEN NULL ELSE VIS END,
|
|
TETO = CASE WHEN TETO IS NOT NULL AND (TETO < 0 OR TETO > 9999) THEN NULL ELSE TETO END,
|
|
PREC = CASE WHEN PREC IS NOT NULL AND (PREC < 0 OR PREC > 500 ) THEN NULL ELSE PREC END
|
|
{where_clause}
|
|
""")
|
|
conn.commit()
|
|
return total
|
|
|
|
|
|
# ── Outlier audit queries ─────────────────────────────────────────────────────
|
|
def get_outlier_summary(
|
|
conn: sqlite3.Connection,
|
|
aerodrome: Optional[str] = None,
|
|
) -> pd.DataFrame:
|
|
"""Returns per-variable outlier counts and statistics from the audit log.
|
|
|
|
Args:
|
|
conn: Open SQLite connection.
|
|
aerodrome: Filter to a single ICAO code, or ``None`` for all.
|
|
|
|
Returns:
|
|
DataFrame with columns ``variable``, ``aerodrome``, ``n_outliers``,
|
|
``min_orig``, ``max_orig``, ``reason``, ``last_applied``.
|
|
"""
|
|
aero_filter = "WHERE aerodrome=?" if aerodrome else ""
|
|
params = (aerodrome,) if aerodrome else ()
|
|
return pd.read_sql_query(
|
|
f"""
|
|
SELECT
|
|
variable,
|
|
aerodrome,
|
|
COUNT(*) AS n_outliers,
|
|
MIN(orig_value) AS min_orig,
|
|
MAX(orig_value) AS max_orig,
|
|
reason,
|
|
MAX(applied_at) AS last_applied
|
|
FROM outlier_log
|
|
{aero_filter}
|
|
GROUP BY variable, aerodrome, reason
|
|
ORDER BY variable, aerodrome
|
|
""",
|
|
conn, params=params,
|
|
)
|
|
|
|
|
|
def get_outlier_detail(
|
|
conn: sqlite3.Connection,
|
|
aerodrome: Optional[str] = None,
|
|
variable: Optional[str] = None,
|
|
limit: int = 500,
|
|
) -> pd.DataFrame:
|
|
"""Returns row-level outlier records from the audit log.
|
|
|
|
Args:
|
|
conn: Open SQLite connection.
|
|
aerodrome: Filter by ICAO code, or ``None`` for all.
|
|
variable: Filter by variable name (e.g. ``"T"``), or ``None`` for all.
|
|
limit: Maximum number of rows to return.
|
|
|
|
Returns:
|
|
DataFrame with columns ``aerodrome``, ``dt``, ``variable``,
|
|
``orig_value``, ``reason``, ``applied_at``.
|
|
"""
|
|
conditions: list[str] = []
|
|
params: list[object] = []
|
|
if aerodrome:
|
|
conditions.append("aerodrome=?")
|
|
params.append(aerodrome)
|
|
if variable:
|
|
conditions.append("variable=?")
|
|
params.append(variable)
|
|
where = ("WHERE " + " AND ".join(conditions)) if conditions else ""
|
|
return pd.read_sql_query(
|
|
f"SELECT aerodrome, dt, variable, orig_value, reason, applied_at "
|
|
f"FROM outlier_log {where} ORDER BY applied_at DESC, aerodrome, dt "
|
|
f"LIMIT {limit}",
|
|
conn, params=params,
|
|
)
|
|
|
|
|
|
def has_outlier_log(conn: sqlite3.Connection) -> bool:
|
|
"""Returns True if at least one outlier record exists in the audit log.
|
|
|
|
Args:
|
|
conn: Open SQLite connection.
|
|
"""
|
|
n = conn.execute("SELECT COUNT(*) FROM outlier_log").fetchone()[0]
|
|
return n > 0
|
|
|
|
|
|
# ── Aerodrome catalog ─────────────────────────────────────────────────────────
|
|
def upsert_aerodromes(
|
|
conn: sqlite3.Connection,
|
|
aerodromes: list[dict],
|
|
) -> int:
|
|
"""Inserts or updates aerodrome catalog entries.
|
|
|
|
Args:
|
|
conn: Open SQLite connection with schema already applied.
|
|
aerodromes: List of dicts, each with keys ``icao``, ``nome``, ``uf``.
|
|
|
|
Returns:
|
|
Number of records processed.
|
|
"""
|
|
if not aerodromes:
|
|
return 0
|
|
now = datetime.now(timezone.utc).isoformat()
|
|
with conn:
|
|
conn.executemany(
|
|
"INSERT OR REPLACE INTO aerodromes(icao, nome, uf, updated_at)"
|
|
" VALUES(?, ?, ?, ?)",
|
|
[(a["icao"], a.get("nome", ""), a.get("uf", ""), now)
|
|
for a in aerodromes if a.get("icao")],
|
|
)
|
|
return len(aerodromes)
|
|
|
|
|
|
def list_all_aerodromes(conn: sqlite3.Connection) -> list[dict]:
|
|
"""Returns all entries from the aerodrome catalog table.
|
|
|
|
Args:
|
|
conn: Open SQLite connection.
|
|
|
|
Returns:
|
|
List of ``{"icao": str, "nome": str, "uf": str}`` dicts ordered by
|
|
ICAO code. Returns an empty list if the catalog is empty.
|
|
"""
|
|
rows = conn.execute(
|
|
"SELECT icao, nome, uf FROM aerodromes ORDER BY icao"
|
|
).fetchall()
|
|
return [{"icao": r[0], "nome": r[1] or "", "uf": r[2] or ""} for r in rows]
|
|
|
|
|
|
def list_aerodromes_with_data(conn: sqlite3.Connection) -> list[dict]:
|
|
"""Returns catalog entries for aerodromes that have rows in observations.
|
|
|
|
Joins the ``observations`` table with the ``aerodromes`` catalog so the
|
|
caller receives full display metadata (name, state) for each code that
|
|
actually has data.
|
|
|
|
Args:
|
|
conn: Open SQLite connection with schema already applied.
|
|
|
|
Returns:
|
|
List of ``{"icao": str, "nome": str, "uf": str}`` dicts ordered by
|
|
ICAO code. Returns an empty list when no observations exist yet.
|
|
"""
|
|
rows = conn.execute(
|
|
"""SELECT DISTINCT o.aerodrome,
|
|
COALESCE(a.nome, ''),
|
|
COALESCE(a.uf, '')
|
|
FROM observations o
|
|
LEFT JOIN aerodromes a ON a.icao = o.aerodrome
|
|
ORDER BY o.aerodrome"""
|
|
).fetchall()
|
|
return [{"icao": r[0], "nome": r[1], "uf": r[2]} for r in rows]
|
|
|
|
|
|
# ── General queries ───────────────────────────────────────────────────────────
|
|
def query_analytics(
|
|
conn: sqlite3.Connection,
|
|
aerodrome: str,
|
|
start_dt: Optional[str] = None,
|
|
end_dt: Optional[str] = None,
|
|
) -> pd.DataFrame:
|
|
"""Queries analytic observations for one aerodrome over an optional date range.
|
|
|
|
Args:
|
|
conn: Open SQLite connection.
|
|
aerodrome: ICAO code to query.
|
|
start_dt: Inclusive start date/datetime string (``"YYYY-MM-DD"`` or
|
|
``"YYYY-MM-DD HH:MM:SS"``). No lower bound if ``None``.
|
|
end_dt: Inclusive end date (``"YYYY-MM-DD"``). When only a date is
|
|
given, the comparison extends to ``23:59:59`` of that day.
|
|
|
|
Returns:
|
|
DataFrame with columns ``_dt`` (``datetime64``) plus the ANL_COLS
|
|
floats. Returns an empty DataFrame if no rows match.
|
|
"""
|
|
sql = (
|
|
"SELECT dt, T, Td, UR, QNH, WS, WG, WD, VIS, TETO, PREC "
|
|
"FROM observations WHERE aerodrome=?"
|
|
)
|
|
params: list[object] = [aerodrome]
|
|
if start_dt:
|
|
sql += " AND dt >= ?"
|
|
params.append(start_dt)
|
|
if end_dt:
|
|
sql += " AND dt <= ?"
|
|
params.append(end_dt + " 23:59:59" if len(end_dt) == 10 else end_dt)
|
|
sql += " ORDER BY dt"
|
|
df = pd.read_sql_query(sql, conn, params=params)
|
|
if df.empty:
|
|
return df
|
|
df["_dt"] = pd.to_datetime(df["dt"], format="%Y-%m-%d %H:%M:%S", errors="coerce")
|
|
return df.drop(columns=["dt"]).dropna(subset=["_dt"]).reset_index(drop=True)
|
|
|
|
|
|
def get_coverage(
|
|
conn: sqlite3.Connection,
|
|
aerodrome: str,
|
|
) -> Optional[tuple[date, date]]:
|
|
"""Returns the earliest and latest observation dates for one aerodrome.
|
|
|
|
Args:
|
|
conn: Open SQLite connection.
|
|
aerodrome: ICAO code to inspect.
|
|
|
|
Returns:
|
|
``(min_date, max_date)`` tuple, or ``None`` if no observations exist.
|
|
"""
|
|
row = conn.execute(
|
|
"SELECT MIN(dt), MAX(dt) FROM observations WHERE aerodrome=?",
|
|
(aerodrome,),
|
|
).fetchone()
|
|
if not row or row[0] is None:
|
|
return None
|
|
return date.fromisoformat(row[0][:10]), date.fromisoformat(row[1][:10])
|
|
|
|
|
|
def list_aerodromes(conn: sqlite3.Connection) -> list[str]:
|
|
"""Returns ICAO codes of all aerodromes that have at least one observation.
|
|
|
|
Args:
|
|
conn: Open SQLite connection.
|
|
|
|
Returns:
|
|
Sorted list of ICAO code strings.
|
|
"""
|
|
rows = conn.execute(
|
|
"SELECT DISTINCT aerodrome FROM observations ORDER BY aerodrome"
|
|
).fetchall()
|
|
return [r[0] for r in rows]
|
|
|
|
|
|
def aerodrome_stats(conn: sqlite3.Connection, aerodrome: str) -> dict:
|
|
"""Returns basic row-count statistics for one aerodrome.
|
|
|
|
Args:
|
|
conn: Open SQLite connection.
|
|
aerodrome: ICAO code to inspect.
|
|
|
|
Returns:
|
|
Dict with keys ``min_dt``, ``max_dt``, ``n_obs``, ``n_T``, ``n_VIS``.
|
|
Returns an empty dict if no observations exist.
|
|
"""
|
|
row = conn.execute(
|
|
"SELECT MIN(dt), MAX(dt), COUNT(*), COUNT(T), COUNT(VIS) "
|
|
"FROM observations WHERE aerodrome=?",
|
|
(aerodrome,),
|
|
).fetchone()
|
|
if not row or row[0] is None:
|
|
return {}
|
|
return {
|
|
"min_dt": row[0], "max_dt": row[1],
|
|
"n_obs": row[2], "n_T": row[3], "n_VIS": row[4],
|
|
}
|