Files
stroblmeandClaude Opus 5 6bc71ddaf3 Record flags in history and draw them as steps
A bool was excluded from the history as "not a measurement", so a true/false
port had no curve in the node panel and none on an edge — only the word. It is
recorded as 0/1 now and drawn as steps, since a bezier through two states
slopes through readings that never happened. The axis is pinned to 0..1, so a
flag that was never on sits at the floor rather than mid-box.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-08-27 15:54:06 +02:00

115 lines
3.7 KiB
Python

"""Messages: what a dashboard binds to, across every flow.
The flow API is scoped to one flow, which is the wrong shape here — a wall
panel shows the heating alongside the solar. These endpoints are the whole
namespace at once: what exists, what it last was, and a way to put a value
into it without owning a node.
"""
from typing import Any
from fastapi import APIRouter, Depends, HTTPException
from fastapi.concurrency import run_in_threadpool
from pydantic import BaseModel
from fluksio.api.deps import FlowControllerDep, get_current_user
from fluksio.flow.messages import flow_of
from fluksio.flow.pipeline import ValueSource
from fluksio.flow.state import as_number
router = APIRouter(
prefix="/messages", tags=["messages"], dependencies=[Depends(get_current_user)]
)
class MessageInfo(BaseModel):
"""One message, as something to bind a widget to."""
name: str
flow: str
dtype: str
#: Nodes publishing it. Empty means the flow declares it as an input.
providers: list[str] = []
#: Whether a dashboard may publish to it — that is, whether it is declared.
writable: bool = True
numeric: bool = False
value: Any = None
ts: float | None = None
class MessagesPublic(BaseModel):
data: list[MessageInfo]
count: int
class PublishRequest(BaseModel):
value: Any
#: Where this came from, so the canvas can show it arriving from outside
#: rather than crediting whichever node is drawn as a producer.
source_kind: str = "api"
source_id: str = ""
source_label: str = ""
source_detail: str = ""
class MessageValue(BaseModel):
name: str
value: Any
ts: float | None = None
class MessagePoints(BaseModel):
message: str
numeric: bool
points: list[dict[str, float]]
@router.get("/", response_model=MessagesPublic)
async def read_messages(controller: FlowControllerDep) -> Any:
"""Every message any published flow declares, with its last value."""
infos = await run_in_threadpool(controller.message_catalog)
return MessagesPublic(data=infos, count=len(infos))
@router.post("/{name}", response_model=MessageValue)
async def publish_message(
name: str, body: PublishRequest, controller: FlowControllerDep
) -> Any:
"""Put a value into the graph, as a dashboard control does.
Only a message some flow declares can be published to: flows own the
namespace, and a dashboard is a client of it rather than a second author.
"""
source = ValueSource(
kind=body.source_kind,
id=body.source_id,
label=body.source_label or body.source_id or "API",
detail=body.source_detail,
)
try:
await run_in_threadpool(controller.publish_message, name, body.value, source)
except KeyError:
raise HTTPException(
status_code=404, detail=f"No flow declares a message named '{name}'"
)
except TypeError as exc:
raise HTTPException(status_code=422, detail=str(exc))
values = controller.values(flow_of(name))
current = values.get(name, {})
return MessageValue(name=name, value=current.get("value"), ts=current.get("ts"))
@router.get("/{name}/history", response_model=MessagePoints)
def read_message_history(name: str, controller: FlowControllerDep) -> Any:
"""The series behind a chart. Numbers and flags — nothing else plots."""
if controller.pipeline is None:
return MessagePoints(message=name, numeric=False, points=[])
series = controller.state.history(name)
numeric = as_number(controller.state.get(name)) is not None
return MessagePoints(
message=name,
numeric=numeric or bool(series),
points=[{"ts": ts, "value": value} for ts, value in series],
)