dashboard: pace a querying chart by its window, and an example to evaluate it
A chart is drawn in buckets, and nothing it can show changes until the bucket it is drawing closes — so the resolution sets the refresh rather than a flat five-second floor. A week at quarter-hour buckets now asks four times an hour instead of sixty, for the same picture. Leaving the field empty follows the window; a slower rate is still honoured. `make seed-example` builds the thing to evaluate it with: a flow that logs a temperature to InfluxDB, a flow that answers a chart's request by turning the window into Flux and the rows back into a series, and a dashboard holding the chart. The reading flow declares the request as an input with a starting value, which is how a flow says a value reaches it from a panel rather than from a node upstream. Axis labels keep enough decimals to stay distinct — `si` rounds to three figures, so every tick of a chart living inside one degree read "19". Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Claude Opus 5
@@ -0,0 +1,340 @@
|
||||
#!/usr/bin/env python
|
||||
"""Seed the querying-chart example: two flows, a dashboard, and data to draw.
|
||||
|
||||
What it builds, and why it is split the way it is:
|
||||
|
||||
climate_log inject -> sample -> influxdb writes the measurements
|
||||
climate_chart build -> influxdb -> parse answers a chart's request
|
||||
climate a dashboard with one querying chart
|
||||
|
||||
The reading half is the point. A chart publishes ``{range_s, interval_s}`` and
|
||||
draws the ``series`` that comes back; between the two sit a Python node that
|
||||
turns the window into Flux and another that turns rows into lines. The database
|
||||
node only holds the credentials and runs what it is handed, so the widget never
|
||||
learns it was InfluxDB — swapping in Postgres means rewriting those two Python
|
||||
nodes and nothing else.
|
||||
|
||||
The writing half exists so the chart has something to show. It samples every
|
||||
30 seconds; delete ``climate_log`` when you are done evaluating.
|
||||
|
||||
Run it against a stack that is already up::
|
||||
|
||||
make -C app seed-example
|
||||
|
||||
Environment (the Makefile passes these):
|
||||
API_URL, FIRST_SUPERUSER, FIRST_SUPERUSER_PASSWORD
|
||||
INFLUX_URL, INFLUX_ORG, INFLUX_BUCKET, INFLUX_TOKEN
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import sys
|
||||
from typing import Any
|
||||
|
||||
import httpx
|
||||
|
||||
API = os.environ.get("API_URL", "http://api.localhost")
|
||||
EMAIL = os.environ.get("FIRST_SUPERUSER", "")
|
||||
PASSWORD = os.environ.get("FIRST_SUPERUSER_PASSWORD", "")
|
||||
|
||||
INFLUX_URL = os.environ.get("INFLUX_URL", "http://influxdb:8086")
|
||||
INFLUX_ORG = os.environ.get("INFLUX_ORG", "fluksio")
|
||||
INFLUX_BUCKET = os.environ.get("INFLUX_BUCKET", "fluksio")
|
||||
INFLUX_TOKEN = os.environ.get("INFLUX_TOKEN", "")
|
||||
|
||||
SECRET = "influx_eval_token"
|
||||
LOG_FLOW = "climate_log"
|
||||
CHART_FLOW = "climate_chart"
|
||||
PANEL = "climate"
|
||||
|
||||
MEASUREMENT = "climate"
|
||||
FIELD = "temperature"
|
||||
|
||||
SAMPLE_SOURCE = '''"""A plausible indoor temperature, so the example has a curve to draw."""
|
||||
|
||||
import math
|
||||
import time
|
||||
|
||||
|
||||
def process(tick, params):
|
||||
# A slow daily swing plus a faster one, so any window shows some shape.
|
||||
now = time.time()
|
||||
daily = 3.0 * math.sin(now / 86400.0 * 2 * math.pi)
|
||||
churn = 0.4 * math.sin(now / 900.0 * 2 * math.pi)
|
||||
return {"temperature": round(20.5 + daily + churn, 2)}
|
||||
'''
|
||||
|
||||
BUILD_SOURCE = f'''"""Turn a chart's window into Flux. This is the database-specific half."""
|
||||
|
||||
|
||||
def process(chart_request, params):
|
||||
span = int(chart_request["range_s"])
|
||||
every = int(chart_request["interval_s"])
|
||||
flux = "\\n".join(
|
||||
[
|
||||
'from(bucket: "{INFLUX_BUCKET}")',
|
||||
f" |> range(start: -{{span}}s)",
|
||||
' |> filter(fn: (r) => r["_measurement"] == "{MEASUREMENT}")',
|
||||
' |> filter(fn: (r) => r["_field"] == "{FIELD}")',
|
||||
f" |> aggregateWindow(every: {{every}}s, fn: mean, createEmpty: false)",
|
||||
]
|
||||
)
|
||||
# Everything beside "flux" is echoed back by the node, and the widget
|
||||
# checks it against what it asked for — so it has to travel with the query.
|
||||
return {{
|
||||
"query": {{
|
||||
"flux": flux,
|
||||
"range_s": chart_request["range_s"],
|
||||
"interval_s": chart_request["interval_s"],
|
||||
}}
|
||||
}}
|
||||
'''
|
||||
|
||||
PARSE_SOURCE = '''"""Turn rows into the series a chart draws. Nothing here is InfluxDB-specific."""
|
||||
|
||||
|
||||
def process(rows, params):
|
||||
points = [
|
||||
[row["ts"], float(row["value"])]
|
||||
for row in rows["rows"]
|
||||
if row.get("ts") is not None and row.get("value") is not None
|
||||
]
|
||||
return {
|
||||
"temperature_series": {
|
||||
# The echo the widget matches against its own request.
|
||||
"range_s": rows["range_s"],
|
||||
"interval_s": rows["interval_s"],
|
||||
"lines": [{"label": "Indoor", "points": points}],
|
||||
}
|
||||
}
|
||||
'''
|
||||
|
||||
|
||||
def influx_params() -> dict[str, Any]:
|
||||
"""Credentials for a database node, with the token kept out of the flow."""
|
||||
return {
|
||||
"url": INFLUX_URL,
|
||||
"token": {"$secret": SECRET},
|
||||
"org": INFLUX_ORG,
|
||||
"bucket": INFLUX_BUCKET,
|
||||
}
|
||||
|
||||
|
||||
LOG_NODES = [
|
||||
{
|
||||
"id": "every_30s",
|
||||
"type": "inject",
|
||||
"title": "Every 30 seconds",
|
||||
"position": {"x": 40, "y": 80},
|
||||
"params": {"interval": 30, "at_start": True, "payload": 1},
|
||||
"requires": [],
|
||||
"provides": [{"name": "tick", "dtype": "float"}],
|
||||
},
|
||||
{
|
||||
"id": "sample",
|
||||
"type": "python",
|
||||
"title": "Read the room",
|
||||
"position": {"x": 340, "y": 80},
|
||||
"requires": [{"name": "tick", "dtype": "float"}],
|
||||
"provides": [{"name": "temperature", "dtype": "float"}],
|
||||
},
|
||||
{
|
||||
"id": "store",
|
||||
"type": "influxdb",
|
||||
"title": "Write to InfluxDB",
|
||||
"position": {"x": 640, "y": 80},
|
||||
"params": {
|
||||
**influx_params(),
|
||||
"writes": {
|
||||
"temperature": {"measurement": MEASUREMENT, "field": FIELD},
|
||||
},
|
||||
},
|
||||
"requires": [{"name": "temperature", "dtype": "float"}],
|
||||
"provides": [],
|
||||
},
|
||||
]
|
||||
|
||||
CHART_NODES = [
|
||||
{
|
||||
"id": "build",
|
||||
"type": "python",
|
||||
"title": "Window to Flux",
|
||||
"position": {"x": 320, "y": 80},
|
||||
"requires": [{"name": "chart_request", "dtype": "record"}],
|
||||
"provides": [{"name": "query", "dtype": "record"}],
|
||||
},
|
||||
{
|
||||
"id": "read",
|
||||
"type": "influxdb",
|
||||
"title": "Run the query",
|
||||
"position": {"x": 680, "y": 80},
|
||||
"params": influx_params(),
|
||||
"requires": [{"name": "query", "dtype": "record"}],
|
||||
"provides": [{"name": "rows", "dtype": "json"}],
|
||||
},
|
||||
{
|
||||
"id": "parse",
|
||||
"type": "python",
|
||||
"title": "Rows to a series",
|
||||
"position": {"x": 1040, "y": 80},
|
||||
"requires": [{"name": "rows", "dtype": "json"}],
|
||||
"provides": [{"name": "temperature_series", "dtype": "series"}],
|
||||
},
|
||||
]
|
||||
|
||||
WIDGETS = [
|
||||
{
|
||||
"id": "indoor",
|
||||
"type": "chart",
|
||||
"title": "Indoor temperature",
|
||||
"layout": {"lg": {"x": 0, "y": 0, "w": 8, "h": 5}},
|
||||
"config": {
|
||||
"source": "query",
|
||||
"request": f"{CHART_FLOW}.chart_request",
|
||||
"request_dtype": "record",
|
||||
"message": f"{CHART_FLOW}.temperature_series",
|
||||
"dtype": "series",
|
||||
"range_s": 3600,
|
||||
"unit": "°C",
|
||||
},
|
||||
},
|
||||
{
|
||||
"id": "how",
|
||||
"type": "markdown",
|
||||
"title": "",
|
||||
"layout": {"lg": {"x": 8, "y": 0, "w": 4, "h": 5}},
|
||||
"config": {
|
||||
# The widget renders headings and bullets, nothing inline.
|
||||
"content": (
|
||||
"## How this works\n"
|
||||
"- The chart publishes a request: the window and the"
|
||||
" resolution it wants.\n"
|
||||
"- climate_chart turns that into Flux, runs it, and answers"
|
||||
" with a series.\n"
|
||||
"- The answer says which window it was computed for, and the"
|
||||
" chart ignores one that does not match.\n"
|
||||
"- Nothing in the widget knows it was InfluxDB. Swapping the"
|
||||
" database means rewriting two Python nodes.\n"
|
||||
"- climate_log writes a sample every 30 s. Delete that flow"
|
||||
" when you are done evaluating."
|
||||
)
|
||||
},
|
||||
},
|
||||
]
|
||||
|
||||
|
||||
class Api:
|
||||
def __init__(self) -> None:
|
||||
self.http = httpx.Client(base_url=f"{API}/api/v1", timeout=30)
|
||||
token = self.http.post(
|
||||
"/login/access-token",
|
||||
data={"username": EMAIL, "password": PASSWORD},
|
||||
).json()["access_token"]
|
||||
self.http.headers["Authorization"] = f"Bearer {token}"
|
||||
|
||||
def __call__(self, method: str, path: str, body: Any = None) -> Any:
|
||||
response = self.http.request(method, path, json=body)
|
||||
response.raise_for_status()
|
||||
return response.json() if response.content else None
|
||||
|
||||
|
||||
def seed_flow(
|
||||
api: Api,
|
||||
name: str,
|
||||
title: str,
|
||||
nodes: list,
|
||||
sources: dict,
|
||||
inputs: list | None = None,
|
||||
) -> None:
|
||||
"""Write a flow and publish it, replacing whatever was there before."""
|
||||
try:
|
||||
api("DELETE", f"/flows/{name}")
|
||||
except httpx.HTTPStatusError:
|
||||
pass
|
||||
api(
|
||||
"PUT",
|
||||
f"/flows/{name}",
|
||||
{"name": name, "title": title, "nodes": nodes, "inputs": inputs or []},
|
||||
)
|
||||
for node_id, code in sources.items():
|
||||
api("PUT", f"/flows/{name}/nodes/{node_id}/source", {"code": code})
|
||||
version = api("GET", f"/flows/{name}?draft=true")["definition"]["version"]
|
||||
api("POST", f"/flows/{name}/publish", {"version": version})
|
||||
print(f" {name}: {len(nodes)} nodes, published")
|
||||
|
||||
|
||||
def main() -> int:
|
||||
if not EMAIL or not PASSWORD:
|
||||
print("FIRST_SUPERUSER / FIRST_SUPERUSER_PASSWORD are unset.", file=sys.stderr)
|
||||
return 1
|
||||
if not INFLUX_TOKEN:
|
||||
print(
|
||||
"INFLUX_TOKEN is unset — set it to a token that can read and write "
|
||||
f"the '{INFLUX_BUCKET}' bucket.",
|
||||
file=sys.stderr,
|
||||
)
|
||||
return 1
|
||||
|
||||
api = Api()
|
||||
|
||||
# The flows reference the token by name, so it never sits in the document.
|
||||
api("PUT", f"/secrets/{SECRET}", {"value": INFLUX_TOKEN})
|
||||
print(f" secret '{SECRET}' set")
|
||||
|
||||
seed_flow(
|
||||
api,
|
||||
LOG_FLOW,
|
||||
"Climate log (evaluation)",
|
||||
LOG_NODES,
|
||||
{"sample": SAMPLE_SOURCE},
|
||||
)
|
||||
seed_flow(
|
||||
api,
|
||||
CHART_FLOW,
|
||||
"Climate chart (evaluation)",
|
||||
CHART_NODES,
|
||||
{"build": BUILD_SOURCE, "parse": PARSE_SOURCE},
|
||||
# The request arrives from the panel, not from a node upstream. Saying
|
||||
# so is what stops the canvas reporting `build` as waiting on something
|
||||
# nothing provides — a flow declares what reaches it from outside. The
|
||||
# initial value is the widget's own default window, so the flow has an
|
||||
# answer ready before anyone opens the dashboard.
|
||||
inputs=[
|
||||
{
|
||||
"spec": {"name": "chart_request", "dtype": "record"},
|
||||
"initial": {"range_s": 3600, "interval_s": 60},
|
||||
}
|
||||
],
|
||||
)
|
||||
|
||||
try:
|
||||
api("DELETE", f"/dashboards/{PANEL}")
|
||||
except httpx.HTTPStatusError:
|
||||
pass
|
||||
api("POST", f"/dashboards/{PANEL}", {"name": PANEL, "title": "Climate"})
|
||||
current = api("GET", f"/dashboards/{PANEL}")
|
||||
api(
|
||||
"PUT",
|
||||
f"/dashboards/{PANEL}",
|
||||
{
|
||||
**current,
|
||||
"pages": [
|
||||
{
|
||||
"id": "main",
|
||||
"title": "Overview",
|
||||
"sections": [{"id": "main", "widgets": WIDGETS}],
|
||||
}
|
||||
],
|
||||
},
|
||||
)
|
||||
version = api("GET", f"/dashboards/{PANEL}?draft=true")["version"]
|
||||
api("POST", f"/dashboards/{PANEL}/publish", {"version": version})
|
||||
print(f" dashboard '{PANEL}': published")
|
||||
|
||||
print(f"\nOpen it at /view/{PANEL} — the first samples land within 30 s.")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user