144 lines
5.6 KiB
Python
144 lines
5.6 KiB
Python
import json
|
|
import os
|
|
import panel
|
|
from datetime import datetime
|
|
from typing import Any
|
|
|
|
import pandas as pd
|
|
|
|
from pydantic import BaseModel
|
|
|
|
from watt42_viewlib import attach_w42_state
|
|
|
|
panel.extension('echarts', 'ace', 'jsoneditor', 'perspective')
|
|
|
|
SYSTEM_ID = os.environ.get("WATT42_SYSTEM_ID", "invalid-system-id")
|
|
API_TOKEN = os.environ.get("WATT42_API_TOKEN", "invalid-api-token")
|
|
|
|
w42_state = panel.rx(None)
|
|
|
|
attach_w42_state(rx_var=w42_state, system_id=SYSTEM_ID, token=API_TOKEN)
|
|
|
|
class PvForecast(BaseModel):
|
|
at: datetime = datetime.now()
|
|
slots: list[float] = []
|
|
|
|
class LoadForecast(BaseModel):
|
|
at: datetime = datetime.now()
|
|
slots: list[float] = []
|
|
|
|
|
|
class InverterForecast(BaseModel):
|
|
levels: list[float] = []
|
|
grid_import_power: list[float] = []
|
|
|
|
class SystemState(BaseModel):
|
|
pv_forecast: PvForecast = PvForecast()
|
|
load_forecast: LoadForecast = LoadForecast()
|
|
value: int = 42
|
|
inverter: InverterForecast = InverterForecast()
|
|
|
|
def state_to_text(state: dict[str, Any]) -> str:
|
|
return f"Watt42 State:\n\n```\n{json.dumps(state, indent=2)}\n```\n\nReplace this with some awesome visuals"
|
|
|
|
state_as_text = panel.bind(state_to_text, w42_state)
|
|
|
|
state_pane = panel.pane.Markdown(state_as_text, sizing_mode='stretch_width')
|
|
|
|
def chart1(state: SystemState) -> dict:
|
|
""" Return an ECharts option dict visualizing some aspect of the state."""
|
|
|
|
now = state.load_forecast.at
|
|
pv_slots = state.pv_forecast.slots
|
|
load_slots = state.load_forecast.slots
|
|
hours = list(range(len(pv_slots)))
|
|
option = {
|
|
"title": {"text": f"PV and Load Forecast at {now.strftime('%Y-%m-%d %H:%M')}"},
|
|
"tooltip": {"trigger": "axis"},
|
|
"legend": {"data": ["PV Forecast", "Load Forecast", "Inverter Level", "Grid import"]},
|
|
"xAxis": {"type": "category", "data": hours, "name": "Hours"},
|
|
"yAxis": {"type": "value", "name": "Power (kW)"},
|
|
"series": [
|
|
{
|
|
"name": "PV Forecast",
|
|
"type": "line",
|
|
"data": pv_slots,
|
|
},
|
|
{
|
|
"name": "Load Forecast",
|
|
"type": "line",
|
|
"data": load_slots,
|
|
},
|
|
{
|
|
"name": "Inverter Level",
|
|
"type": "line",
|
|
"data": state.inverter.levels,
|
|
},
|
|
{
|
|
"name": "Grid import",
|
|
"type": "line",
|
|
"data": state.inverter.grid_import_power,
|
|
}
|
|
],
|
|
}
|
|
return option
|
|
|
|
chart1_rx = panel.rx(chart1)(panel.rx(lambda s: SystemState.model_validate(s) if s else SystemState())(w42_state))
|
|
|
|
def series_as_df(state: SystemState) -> pd.DataFrame:
|
|
""" Return a pandas DataFrame representation of the series data."""
|
|
data = {
|
|
"Time": [state.load_forecast.at + pd.Timedelta(minutes=15 * i) for i in range(len(state.load_forecast.slots))],
|
|
"PV Forecast": state.pv_forecast.slots,
|
|
"Load Forecast": state.load_forecast.slots,
|
|
"Inverter Level": state.inverter.levels,
|
|
}
|
|
df = pd.DataFrame(data)
|
|
return df
|
|
|
|
series_df_rx = panel.rx(series_as_df)(panel.rx(lambda s: SystemState.model_validate(s) if s else SystemState())(w42_state))
|
|
|
|
def summary_as_markdown(state: SystemState) -> str:
|
|
""" Return a markdown summary of the system state."""
|
|
md = f"""
|
|
### Summary
|
|
|
|
- Total PV Forecast: {sum(state.pv_forecast.slots) / 4000.0:.2f} kWh
|
|
- Total Load Forecast: {sum(state.load_forecast.slots) / 4000.0:.2f} kWh
|
|
- Total grid import: {sum(state.inverter.grid_import_power) / 4000.0:.2f} kWh
|
|
"""
|
|
|
|
return md
|
|
|
|
state_summary_rx = panel.rx(summary_as_markdown)(panel.rx(lambda s: SystemState.model_validate(s) if s else SystemState())(w42_state))
|
|
|
|
sidebar_content = """
|
|
Demonstrates how to visualize the Watt42 "dumb inverter simulation" system state.
|
|
|
|
Find instructions on [how to use this example
|
|
here](https://source.c3.uber5.com/watt42-public/watt42_viewlib/src/branch/main/README.md#how-to-use).
|
|
|
|
"""
|
|
|
|
grid_config = json.loads('{"version":"3.8.0","plugin":"Datagrid","plugin_config":{"columns":{},"edit_mode":"READ_ONLY","scroll_lock":false},"columns_config":{},"settings":true,"theme":"Pro Light","title":null,"group_by":["level rounded"],"split_by":[],"sort":[["level rounded","desc"]],"filter":[],"expressions":{"level rounded":"round(\\"Inverter Level\\" / 100) * 100\\t","one":"1"},"columns":["index","one","Time","PV Forecast","Load Forecast","Inverter Level","level rounded"],"aggregates":{}}')
|
|
|
|
_ = panel.template.FastListTemplate(
|
|
title="Watt42: Dumb Inverter Visualization Example",
|
|
sidebar=[panel.pane.Markdown(sidebar_content, sizing_mode='stretch_width'), panel.pane.Markdown(state_summary_rx, sizing_mode='stretch_width')],
|
|
main=[
|
|
panel.pane.ECharts(chart1_rx, sizing_mode='stretch_width', height=400, theme='light'),
|
|
panel.pane.Perspective(series_df_rx, sizing_mode='stretch_width',
|
|
height=400, columns=grid_config['columns'],
|
|
title="Group by rounded battery level (experimental)",
|
|
expressions=grid_config['expressions'],
|
|
aggregates=grid_config['aggregates'],
|
|
group_by=grid_config['group_by'],
|
|
sort=grid_config['sort'],
|
|
filters=grid_config['filter'],
|
|
),
|
|
# warn that above pane may not work on all browsers
|
|
panel.pane.Markdown("**Note:** The above data grid uses the Perspective library which may not work on all browsers.", sizing_mode='stretch_width'),
|
|
state_pane,
|
|
w42_state,
|
|
],
|
|
).servable()
|