Real-Time Dashboard

Build a dashboard showing current grid conditions with automatic refresh

Build a dashboard showing current grid conditions with automatic refresh.

import time


def _get_latest(client, dataset, **extra_params):
    """Fetch the latest data point for a dataset using the time=latest API parameter."""
    params = {"time": "latest", **extra_params}
    df, _meta, _dataset_metadata = client.get(
        f"{client.host}/datasets/{dataset}/query",
        params=params,
    )
    return df


def get_current_grid_status(client):
    """Fetch current grid conditions across multiple datasets."""

    # Get latest fuel mix
    fuel_mix = _get_latest(client, "ercot_fuel_mix")

    # Get latest load
    load = _get_latest(client, "ercot_load")

    # Get latest hub prices
    prices = _get_latest(
        client,
        "ercot_lmp_by_settlement_point",
        filter_column="location",
        filter_value="HB_HOUSTON,HB_NORTH,HB_WEST",
        filter_operator="in",
    )

    return {
          "timestamp": fuel_mix["interval_start_utc"].iloc[0],
          "total_load_mw": load["load"].iloc[0],
          "solar_mw": fuel_mix["solar"].iloc[0],
          "wind_mw": fuel_mix["wind"].iloc[0],
          "renewable_pct": (fuel_mix["solar"].iloc[0] + fuel_mix["wind"].iloc[0])
          / (
              fuel_mix["coal_and_lignite"].iloc[0]
              + fuel_mix["hydro"].iloc[0]
              + fuel_mix["nuclear"].iloc[0]
              + fuel_mix["power_storage"].iloc[0]
              + fuel_mix["solar"].iloc[0]
              + fuel_mix["wind"].iloc[0]
              + fuel_mix["natural_gas"].iloc[0]
              + fuel_mix["other"].iloc[0]
          )
          * 100,
          "prices": prices[["location", "lmp"]].to_dict("records"),
      }


# Dashboard refresh loop
while True:
    status = get_current_grid_status(client)
    print(f"\n=== Grid Status at {status['timestamp']} ===")
    print(f"Total Load: {status['total_load_mw']:,.0f} MW")
    print(f"Solar: {status['solar_mw']:,.0f} MW | Wind: {status['wind_mw']:,.0f} MW")
    print(f"Renewable Penetration: {status['renewable_pct']:.1f}%")
    print("Hub Prices:")
    for p in status["prices"]:
        print(f"  {p['location']}: ${p['lmp']:.2f}/MWh")

    time.sleep(300)  # Refresh every 5 minutes