app.data_loader

data_loader

DuckDB-based data engine for lazy querying of BirdCurve data sources.

DataEngine

DataEngine(settings: Settings)

Unified query engine. Attaches the on-disk DuckDB read-only and discovers model/forecast dirs.

Source code in dashboard/backend/app/data_loader.py

def __init__(self, settings: Settings):
    self._settings = settings
    # Connect to a transient DB; ATTACH the on-disk DuckDB read-only.
    # Why ATTACH not direct connect: lets us keep sidecar tables (below)
    # alongside on-disk data without touching it.
    self._conn = duckdb.connect()
    self._conn.execute(
        f"ATTACH '{settings.duckdb_path}' AS db (READ_ONLY)"
    )
    # Sidecar CSVs get their own attached in-memory catalog rather than
    # TEMP tables: TEMP objects are session-local and invisible to the
    # per-call cursors that make concurrent request handling safe.
    self._conn.execute("ATTACH ':memory:' AS sidecars")
    # Pin tz to UTC per project convention (init-time queries only;
    # request-time cursors set their own session state in _cursor()).
    self._conn.execute("SET TimeZone='UTC'")
    self._conn.execute("SET search_path='db,sidecars'")

    # Discover model results
    self._model_dir = settings.model_results_dir
    if not self._model_dir.is_dir():
        logger.warning(
            "model_results_dir %s does not exist — no models or forecast "
            "scenarios will be available", self._model_dir,
        )
    self._latest_production = self._find_latest_dir("Production_Ensemble_")
    self._forecast_dirs = self._discover_forecasts()

    # Load small files into memory
    self._small_files_cache: dict[str, Any] = {}
    self._load_small_files()

    # Register optional out-of-DB sidecars (e.g. EUR/USD daily CSV).
    self._eur_usd_registered = self._try_register_eur_usd(settings)
    self._coal_api2_registered = self._try_register_coal_api2(settings)

query_wide

query_wide(table: str, columns: list[str], start: str | None = None, end_exclusive: str | None = None, timestamp_col: str = 'timestamp_utc') -> list[dict]

Select named wide-schema columns + timestamp, optionally filtered by [start, end_exclusive). Callers with an inclusive calendar-date end promote it via _helpers.end_exclusive() first.

Source code in dashboard/backend/app/data_loader.py

def query_wide(
    self,
    table: str,
    columns: list[str],
    start: str | None = None,
    end_exclusive: str | None = None,
    timestamp_col: str = "timestamp_utc",
) -> list[dict]:
    """Select named wide-schema columns + timestamp, optionally filtered
    by [start, end_exclusive). Callers with an inclusive calendar-date end
    promote it via `_helpers.end_exclusive()` first.
    """
    quoted = ", ".join(f'"{c}"' for c in columns)
    where, params = [], []
    if start is not None:
        where.append(f'"{timestamp_col}" >= ?')
        params.append(start)
    if end_exclusive is not None:
        where.append(f'"{timestamp_col}" < ?')
        params.append(end_exclusive)
    where_sql = ("WHERE " + " AND ".join(where)) if where else ""
    sql = f'SELECT "{timestamp_col}", {quoted} FROM {table} {where_sql} ORDER BY "{timestamp_col}"'
    return self.query(sql, params)

query_forecast_file

query_forecast_file(scenario: str, filename_pattern: str, start: str | None = None, end_exclusive: str | None = None, datetime_col: str | tuple[str, ...] | None = None) -> list[dict]

Query a forecast .feather or .csv file from a scenario directory, filtered to [start, end_exclusive). Callers with an inclusive calendar-date end promote it via _helpers.end_exclusive() first.

NOTE: DuckDB read_parquet() CANNOT read .feather (Arrow IPC) files. Feather files are loaded via pandas.read_feather() and registered as cursor-local DuckDB tables for SQL filtering.

Returns [] when the file or expected datetime column is missing (so callers don't need to special-case partial scenario dirs).

Source code in dashboard/backend/app/data_loader.py

def query_forecast_file(
    self,
    scenario: str,
    filename_pattern: str,
    start: str | None = None,
    end_exclusive: str | None = None,
    datetime_col: str | tuple[str, ...] | None = None,
) -> list[dict]:
    """Query a forecast .feather or .csv file from a scenario directory,
    filtered to [start, end_exclusive). Callers with an inclusive
    calendar-date end promote it via `_helpers.end_exclusive()` first.

    NOTE: DuckDB read_parquet() CANNOT read .feather (Arrow IPC) files.
    Feather files are loaded via pandas.read_feather() and registered as
    cursor-local DuckDB tables for SQL filtering.

    Returns [] when the file or expected datetime column is missing
    (so callers don't need to special-case partial scenario dirs).
    """
    fdir = self.forecast_dir(scenario)
    if fdir is None:
        return []

    feather_matches = list(fdir.glob(f"{filename_pattern}.feather"))
    csv_matches = list(fdir.glob(f"{filename_pattern}.csv"))

    # Resolve which datetime column the file actually uses.
    candidates: tuple[str, ...]
    if datetime_col is None:
        candidates = self._DATETIME_COL_CANDIDATES
    elif isinstance(datetime_col, str):
        candidates = (datetime_col, *self._DATETIME_COL_CANDIDATES)
    else:
        candidates = tuple(datetime_col)

    # Register either source as a pandas-backed table so the SQL below
    # never f-strings a file path. Eliminates the residual SQL-injection
    # vector if `filename_pattern` ever comes from user input (today all
    # callers pass literals, but defence in depth is cheap).
    if feather_matches:
        df = pd.read_feather(feather_matches[0])
    elif csv_matches:
        df = pd.read_csv(csv_matches[0])
    else:
        return []

    actual_col = next((c for c in candidates if c in df.columns), None)
    if actual_col is None:
        return []

    where_parts = []
    params = []
    if start:
        where_parts.append(f'"{actual_col}" >= ?')
        params.append(start)
    if end_exclusive:
        where_parts.append(f'"{actual_col}" < ?')
        params.append(end_exclusive)

    where_clause = f"WHERE {' AND '.join(where_parts)}" if where_parts else ""
    sql = f'SELECT * FROM _forecast_src {where_clause} ORDER BY "{actual_col}"'

    # The registration is cursor-local, so concurrent requests can't
    # collide on the name and nothing leaks past the cursor's lifetime.
    with self._cursor() as cur:
        cur.register("_forecast_src", df)
        return _records_from_df(cur.execute(sql, params).fetchdf())