Adding a New Data Source

This guide explains how to add a new backend data adapter so the dashboard can ingest sensor readings from a source other than the existing WebSocket. The WebSocket implementation in app/backend/websocket.py and the SQLite database in app/backend/database.py are the reference patterns.

Overview of What You Will Create

Step	Layer / Responsibility	File Path	Purpose
1	Data Adapter	app/backend/my_source.py	Fetches raw data and returns a standardized DataFrame
2	Configuration	app/configs/database_config.yaml	Defines connection details and selects active data source
3	Settings Integration	app/services/settings_service.py	Loads and persists credentials for the data source
4	Data Routing	app/services/model_manager.py	Selects and calls the correct data adapter
5	Settings UI	app/ui/components/settings_tab.py	Provides UI inputs for configuring the data source
6	Caching Layer	(inside adapter module)	Stores fetched data locally for faster reloads

How Data Flows Into the Dashboard

New data source
      ↓
app/backend/my_source.py        ← fetch raw rows → pd.DataFrame
      ↓
app/services/model_manager.py   ← fetch_selected_experiment_data()
      ↓
app/models/<model>.py           ← run_pipeline(cfg, data=df)
      ↓
app/backend/database.py         ← save_prediction() / log_metrics()
      ↓
Performance Metrics / Experimental Insight pages

The key contract is simple: your adapter must return a pd.DataFrame with the same column schema that the model pipelines expect.

Required DataFrame Schema

All model pipelines read these columns from the raw data:

Column	Type	Description
entity	str	Reactor identifier (e.g. pioreactor_A1)
experiment_id	str	Experiment identifier
time	datetime (UTC)	Timestamp of the reading
metric	str	Parameter name (e.g. optical_density)
value	float	Measured value

These column names are configured in cfg["data"] in each model’s YAML, so they can be changed per-model if needed.

Step 1 — Create the Adapter Module

Create app/backend/my_source.py. The module must expose at minimum one async fetch function that returns a pd.DataFrame:

"""Data adapter for MySource — fetches sensor readings in long format."""

import pandas as pd
from app.backend.logger_config import setup_logger

logger = setup_logger(__name__)


async def fetch_from_my_source(
    cfg: dict,
    experiment_ids: list = None,
    experiment_list_only: bool = False,
) -> pd.DataFrame:
    """Fetch sensor data from MySource.

    Args:
        cfg (dict): Configuration dict. Must include
            cfg["my_source"]["connection_string"] (or similar).
        experiment_ids (list, optional): Filter to these experiment IDs.
            Defaults to None (fetch all).
        experiment_list_only (bool): If True, return only distinct
            experiment_id and entity pairs. Defaults to False.

    Returns:
        pd.DataFrame: Long-format DataFrame with columns:
            entity, experiment_id, time, metric, value.
            Empty DataFrame if no data is available.
    """
    try:
        df = pd.DataFrame()  # replace with real fetch
        logger.info(f"Fetched {len(df)} rows from MySource")
        return df

    except Exception as e:
        logger.error(f"MySource fetch failed: {e}", exc_info=True)
        return pd.DataFrame()

The function signature matches fetch_from_websocket() exactly — this lets model_manager.py swap sources without changing the service layer.

Step 2 — Add Config Support

Add a section to app/configs/database_config.yaml for your source’s connection details:

websocket:
  uri: wss://maple.containers.wur.nl/ws
  base_query: SELECT * FROM public.pioreactor

my_source:
  connection_string: "your://connection@string"
  table: sensor_readings
  poll_interval_seconds: 30

data_source:
  type: websocket     # change to "my_source" to switch adapters

Step 3 — Route the Config in settings_service.py

Open app/services/settings_service.py and update the credential/config loading logic to save and read credentials for your new source. Follow the existing pattern for the WebSocket credentials.

Step 4 — Wire Into model_manager.py

Open app/services/model_manager.py and update fetch_selected_experiment_data():

async def fetch_selected_experiment_data(config_path: str):
    cfg = load_config(config_path)
    source_type = cfg.get("data_source", {}).get("type", "websocket")

    if source_type == "my_source":
        from app.backend.my_source import fetch_from_my_source
        fetch_fn = fetch_from_my_source
    else:
        from app.backend.websocket import fetch_from_websocket
        fetch_fn = fetch_from_websocket

    selected_experiments = get_selected_experiments()

    if not selected_experiments:
        return await fetch_fn(cfg)

    selected_exp_ids = list({exp["experiment_id"] for exp in selected_experiments})
    df_raw = await fetch_fn(cfg, experiment_ids=selected_exp_ids)

    if df_raw.empty:
        return None

    return df_raw[df_raw["experiment_id"].isin(selected_exp_ids)]

Step 5 — Add a Settings UI Panel

Open app/ui/components/settings_tab.py and add input fields for your source’s connection parameters alongside the existing WebSocket fields. Follow the same pattern: inputs save to ~/.graft/credentials.json via settings_service.py.

Data Caching

Write a local Feather cache after every full fetch so the dashboard can reload quickly on repeated runs:

import os
cache_path = os.path.join(USER_DATA_DIR, "data", "my_source_cache.feather")
df.to_feather(cache_path)
logger.info(f"Cached data to {cache_path}")

Checklist

□ app/backend/my_source.py
    □ fetch_from_my_source() is async
    □ Matches fetch_from_websocket() signature
    □ Returns pd.DataFrame with required schema (entity, experiment_id,
      time, metric, value)
    □ Returns empty DataFrame on error, never raises

□ app/configs/database_config.yaml
    □ New section added for your source's connection details
    □ data_source.type key added

□ app/services/settings_service.py
    □ Credential loading/saving updated for new source

□ app/services/model_manager.py
    □ fetch_selected_experiment_data() routes to new adapter when
      data_source.type matches

□ app/ui/components/settings_tab.py
    □ Input fields added for connection parameters

□ Caching
    □ Feather cache written after full fetch

After completing these steps, change data_source.type in database_config.yaml to your source name, restart the application, and all model runs will use your new adapter automatically.