app/scripts/seed_example_chart.py

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