#!/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): # 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): 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): 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())