watt42-public/watt42_viewlib
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Compare commits

base: watt42-public:main
Branches Tags
watt42-public:main
watt42-public:silly-sample
..
compare: watt42-public:silly-sample
Branches Tags
watt42-public:main
watt42-public:silly-sample

No commits in common. "main" and "silly-sample" have entirely different histories.
main ... silly-samp

9 changed files with 56 additions and 369 deletions
Download patch file Download diff file Expand all files Collapse all files

106
README.md
View file

@ -1,105 +1,3 @@
# Watt42 Viewlib
Watt42 Viewlib supports building browser-based front ends for Watt42 systems.
The purpose of those front ends is to visualize the state of a Watt42 system,
and to allow users to interact with it.
# Quickstart
Given a Watt42 system that exposes the status of a smart device (`is_geyser_on`
is either `true` or `false`), the following code creates a simple view that
shows the status of the device.
```python
import os
import panel
from typing import Any
from watt42_viewlib import attach_w42_state
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)
def get_geyser_state(state: dict[str, Any]) -> bool:
if not state:
return False
return state.get("is_geyser_on", False)
geyser_state = panel.rx(get_geyser_state)(w42_state)
indicator = panel.indicators.BooleanStatus(value=geyser_state, name="W42 Connected", color="success")
_ = indicator.servable()
```
# Prerequisites
You should be familiar with Python.
You need poetry installed:
- [Poetry](https://python-poetry.org/docs/#installing-with-the-official-installer)
# Demo
You can [run the sample online](https://viewlib-demo.watt42.com). The sample code is in [sample.py](./sample.py).
TODO: publish demo, provide more sophisticated demos online.
# Documentation
## Installation
Make sure you have poetry installed (see above), then run:
```bash
poetry install
```
You can now run the sample view in dev mode with:
```bash
poetry run panel serve sample.py --dev
```
Dev mode enables hot reloading, so any changes you make to the code will be
reflected immediately in the browser.
Your view will be available at `http://localhost:5006/sample`. In order for the
view to work, you need to configure an access token, and the Watt42 system that
you want to use. See #how-to-use below.
## How to use
Build your own front end for a Watt42 system by creating a Python script that
uses Viewlib. Start off with the sample code in [sample.py](./sample.py). You
can then extend the view by adding more widgets and logic.
1. Follow the installation instructions above.
2. Make sure you have a Watt42 system set up, and you have an access token for it. (TODO: link on how to do this)
3. Set the following environment variables:
- `WATT42_SYSTEM_ID`: The ID of your Watt42 system.
- `WATT42_API_TOKEN`: Your access token.
## Howtos
- See above on how to use Viewlib to build your own view.
- [How to add a diagram](https://source.c3.uber5.com/watt42-public/watt42_viewlib/src/branch/main/docs/howto_add_diagram.md)
- Learn more about how to build views with Panel:
- [Panel Tutorials](https://panel.holoviz.org/tutorials/index.html)
- [Panel Explanation](https://panel.holoviz.org/explanation/index.html)
## Reference
- [Watt42 Viewlib Documentation](https://source.c3.uber5.com/watt42-public/watt42_viewlib/src/branch/main/docs/reference.md)
- Visual components:
- [Panel Component Gallery](https://panel.holoviz.org/reference/index.html)
- [Panel App Gallery](https://panel.holoviz.org/gallery/index.html)
- [Panel API Reference](https://panel.holoviz.org/api/index.html)
# View1 Library
Supports building a view for Watt42 systems

45
docs/howto_add_diagram.md
View file

@ -1,45 +0,0 @@
# Tutorial: How to Add a Diagram to Your Viewlib App
## Prerequisites
- You should be familiar with Python.
- You are able to run the sample as per the ../README.md#installation.
## Introduction
Before adding a diagram, let's add something simpler: a Boolean indicator that shows whether the geyser is on or off.
## Add a Boolean Indicator
```diff
--- a/sample.py
+++ b/sample.py
@@ -21,6 +21,15 @@ state_as_text = panel.bind(state_to_text, w42_state)
state_pane = panel.pane.Markdown(state_as_text, sizing_mode='stretch_width')
+def get_geyser_state(state: dict[str, Any]) -> bool:
+ if not state:
+ return False
+ return state.get("is_geyser_on", False)
+
+geyser_state = panel.rx(get_geyser_state)(w42_state)
+
+indicator = panel.indicators.BooleanStatus(value=geyser_state, name="W42 Connected", color="success")
+
sidebar_content = """
This example shows how to build a front end for a Watt42 system: Watt42 API
manages the state of the system, this app visualizes it.
@@ -34,6 +43,7 @@ _ = panel.template.FastListTemplate(
title="Sample W42 App",
sidebar=[panel.pane.Markdown(sidebar_content, sizing_mode='stretch_width')],
main=[
+ indicator,
state_pane,
w42_state,
],
```
## Add a Diagram
(TODO)

97
dumb-inverter.py
View file

@ -1,97 +0,0 @@
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()

23
geyser_on_off.py
View file

@ -1,23 +0,0 @@
import os
import panel
from typing import Any
from watt42_viewlib import attach_w42_state
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)
def get_geyser_state(state: dict[str, Any]) -> bool:
if not state:
return False
return state.get("is_geyser_on", False)
geyser_state = panel.rx(get_geyser_state)(w42_state)
indicator = panel.indicators.BooleanStatus(value=geyser_state, name="W42 Connected", color="success")
_ = indicator.servable()

2
pyproject.toml
View file

@ -38,8 +38,6 @@ build-backend = "poetry.core.masonry.api"
[tool.pyright]
venvPath = "."
venv = ".venv"
reportAny = false
reportExplicitAny = false
[tool.pytest.ini_options]
pythonpath = ["."]

69
sample-old.py
View file

@ -1,69 +0,0 @@
import json
import panel
from random import randint
from typing import Any
from watt42_viewlib import attach_w42_state
w42_state = panel.rx(None)
attach_w42_state(rx_var=w42_state, system_id="fb2b91ce-383e-4356-96b3-b6405dacb353")
state_as_text = panel.bind(lambda s: f"W42 State:\n\n```\n{json.dumps(s, indent=2)}\n```\n\nReplace this with some awesome visuals", w42_state)
state_pane = panel.pane.Markdown(state_as_text, sizing_mode='stretch_width')
value = w42_state.rx.value
multiplier = panel.widgets.IntSlider(name='Multiplier', start=1, end=10, step=1, value=1)
def value_display(state: dict[str, Any], multiplier: int) -> str:
if state and 'value' in state:
return f"Current W42 State Value times multiplier: {state['value'] * multiplier}"
return "Current W42 State Value: N/A"
def value_or_zero(state: dict[str, Any]) -> int:
if state and 'value' in state:
return state['value']
return 0
def array_from_value(x: int) -> list[int]:
return [x * randint(1, 20) for _ in range(25)]
def times2(x: int) -> int:
return x * 2
times2_rx = panel.rx(times2)(panel.rx(value_or_zero)(w42_state))
markdown_rx = panel.rx(lambda v: f"## Value times 2 is: {v}, array={array_from_value(v)}")(times2_rx)
array_rx = panel.rx(array_from_value)(panel.rx(value_or_zero)(w42_state))
echart_bar_rx = panel.pane.ECharts(
panel.rx(lambda arr: {
'xAxis': {
'type': 'category',
'data': [f'Item {i+1}' for i in range(len(arr))]
},
'yAxis': {
'type': 'value'
},
'series': [{
'data': arr,
'type': 'bar'
}]
})(array_rx),
sizing_mode='stretch_width',
height=400
)
_ = panel.template.FastListTemplate(
title="Sample W42 View App",
sidebar=[panel.pane.Markdown("This is a sample sidebar.")],
main=[
panel.pane.Markdown("# Welcome to the Main Content Area"),
echart_bar_rx,
# state_pane, panel.pane.Markdown(f"Current W42 State Value: {value}"),
panel.ReactiveExpr(panel.rx(value_display)(w42_state, multiplier), widget_location='top'),
panel.pane.Markdown(markdown_rx),
],
).servable()

73
sample.py
View file

@ -1,40 +1,69 @@
import json
import os
import panel
from random import randint
from typing import Any
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="fb2b91ce-383e-4356-96b3-b6405dacb353")
attach_w42_state(rx_var=w42_state, system_id=SYSTEM_ID, token=API_TOKEN)
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_as_text = panel.bind(lambda s: f"W42 State:\n\n```\n{json.dumps(s, indent=2)}\n```\n\nReplace this with some awesome visuals", w42_state)
state_pane = panel.pane.Markdown(state_as_text, sizing_mode='stretch_width')
sidebar_content = """
This example shows how to build a front end for a Watt42 system: Watt42 API
manages the state of the system, this app visualizes it.
value = w42_state.rx.value
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).
multiplier = panel.widgets.IntSlider(name='Multiplier', start=1, end=10, step=1, value=1)
"""
def value_display(state: dict[str, Any], multiplier: int) -> str:
if state and 'value' in state:
return f"Current W42 State Value times multiplier: {state['value'] * multiplier}"
return "Current W42 State Value: N/A"
def value_or_zero(state: dict[str, Any]) -> int:
if state and 'value' in state:
return state['value']
return 0
def array_from_value(x: int) -> list[int]:
return [x * randint(1, 20) for _ in range(25)]
def times2(x: int) -> int:
return x * 2
times2_rx = panel.rx(times2)(panel.rx(value_or_zero)(w42_state))
markdown_rx = panel.rx(lambda v: f"## Value times 2 is: {v}, array={array_from_value(v)}")(times2_rx)
array_rx = panel.rx(array_from_value)(panel.rx(value_or_zero)(w42_state))
echart_bar_rx = panel.pane.ECharts(
panel.rx(lambda arr: {
'xAxis': {
'type': 'category',
'data': [f'Item {i+1}' for i in range(len(arr))]
},
'yAxis': {
'type': 'value'
},
'series': [{
'data': arr,
'type': 'bar'
}]
})(array_rx),
sizing_mode='stretch_width',
height=400
)
_ = panel.template.FastListTemplate(
title="Sample Watt42 App",
sidebar=[panel.pane.Markdown(sidebar_content, sizing_mode='stretch_width')],
title="Sample W42 View App",
sidebar=[panel.pane.Markdown("This is a sample sidebar.")],
main=[
state_pane,
w42_state,
panel.pane.Markdown("# Welcome to the Main Content Area"),
echart_bar_rx,
# state_pane, panel.pane.Markdown(f"Current W42 State Value: {value}"),
panel.ReactiveExpr(panel.rx(value_display)(w42_state, multiplier), widget_location='top'),
panel.pane.Markdown(markdown_rx),
],
).servable()

5
sample2.py
View file

@ -1,5 +1,4 @@
import json
import os
import panel
from datetime import datetime, timedelta
@ -9,8 +8,8 @@ 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")
SYSTEM_ID = "79476e53-dea6-44fa-976c-eff6260baeb6"
API_TOKEN = "d0vA6CsrY69N3JAOGtuZMEb9QpbJWPcoxhxRyBXZn8SIisB3weLKjMZwQRo8c2k9BRDtK0qHYnsvUMnqeO7Xog"
w42_state = panel.rx(None)

5
watt42_viewlib/__init__.py
View file

@ -1,7 +1,6 @@
import asyncio
import json
import logging
import os
import panel
from websockets import ConnectionClosed
@ -9,11 +8,9 @@ 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 = os.environ.get("WATT42_WS_URL", default_ws_url)
WS_URL = "ws://localhost:8000/ws/systems" # TODO: make configurable
must_reconnect = True