diff --git a/dumb-inverter.py b/dumb-inverter.py new file mode 100644 index 0000000..a26d200 --- /dev/null +++ b/dumb-inverter.py @@ -0,0 +1,97 @@ +import json +import os +import panel +from datetime import datetime +from typing import Any + +from pydantic import BaseModel + +from watt42_viewlib import attach_w42_state + +panel.extension('echarts', 'ace', 'jsoneditor') + +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] = [] + +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"]}, + "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, + } + ], + } + return option + +chart1_rx = panel.rx(chart1)(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). + +""" + +_ = panel.template.FastListTemplate( + title="Watt42: Dumb Inverter Visualization Example", + sidebar=[panel.pane.Markdown(sidebar_content, sizing_mode='stretch_width')], + main=[ + panel.pane.ECharts(chart1_rx, sizing_mode='stretch_width', height=400, theme='light'), + state_pane, + w42_state, + ], +).servable() diff --git a/sample2.py b/sample2.py index 96b4463..d6dfbf0 100644 --- a/sample2.py +++ b/sample2.py @@ -1,4 +1,5 @@ import json +import os import panel from datetime import datetime, timedelta @@ -8,8 +9,8 @@ from watt42_viewlib import attach_w42_state panel.extension('echarts', 'ace', 'jsoneditor') -SYSTEM_ID = "79476e53-dea6-44fa-976c-eff6260baeb6" -API_TOKEN = "d0vA6CsrY69N3JAOGtuZMEb9QpbJWPcoxhxRyBXZn8SIisB3weLKjMZwQRo8c2k9BRDtK0qHYnsvUMnqeO7Xog" +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) diff --git a/watt42_viewlib/__init__.py b/watt42_viewlib/__init__.py index 7f733d0..ca8dd0d 100644 --- a/watt42_viewlib/__init__.py +++ b/watt42_viewlib/__init__.py @@ -1,6 +1,7 @@ import asyncio import json import logging +import os import panel from websockets import ConnectionClosed @@ -8,9 +9,11 @@ from websockets.asyncio.client import connect logger = logging.getLogger(__name__) +default_ws_url = "wss://service.watt42.com/ws/systems" + def attach_w42_state(rx_var: panel.rx, system_id: str, token: str): - WS_URL = "ws://localhost:8000/ws/systems" # TODO: make configurable + WS_URL = os.environ.get("WATT42_WS_URL", default_ws_url) must_reconnect = True