stroblmeandClaude Opus 5
6bc71ddaf3
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>
63 lines
2.0 KiB
Python
63 lines
2.0 KiB
Python
"""Message history: the series a node panel draws as a sparkline."""
|
|
|
|
from fluksio.flow.messages import MessageSpec
|
|
from fluksio.flow.nodes import Node
|
|
from fluksio.flow.pipeline import Pipeline
|
|
from fluksio.flow.state import HISTORY_LIMIT, MemoryState
|
|
|
|
|
|
def emitter(values) -> Node:
|
|
"""A node publishing the given values, one per execution."""
|
|
remaining = list(values)
|
|
|
|
def process(params):
|
|
return {"out": remaining.pop(0)}
|
|
|
|
node = Node(f=process, requires=[], provides=[MessageSpec(name="out")], name="src")
|
|
node.assign_flow("f", "src")
|
|
return node
|
|
|
|
|
|
def test_the_series_follows_the_order_values_were_published():
|
|
node = emitter([1.0, 2.0, 3.0])
|
|
pipeline = Pipeline(nodes=[node])
|
|
|
|
# An injecting node publishes directly, an executed one goes through the
|
|
# executor; both paths land in the same series.
|
|
node.inject()
|
|
pipeline.run()
|
|
pipeline.run()
|
|
|
|
points = pipeline.state.history("f.out")
|
|
assert [value for _, value in points] == [1.0, 2.0, 3.0]
|
|
assert [ts for ts, _ in points] == sorted(ts for ts, _ in points)
|
|
|
|
|
|
def test_the_series_is_capped():
|
|
state = MemoryState()
|
|
for i in range(HISTORY_LIMIT + 20):
|
|
state.append_history({"f.out": float(i)}, ts=float(i))
|
|
|
|
points = state.history("f.out")
|
|
assert len(points) == HISTORY_LIMIT
|
|
# The oldest twenty fell off the end.
|
|
assert points[0] == (20.0, 20.0)
|
|
assert points[-1] == (139.0, 139.0)
|
|
|
|
|
|
def test_numbers_and_flags_are_recorded_and_text_is_not():
|
|
"""A flag plots as the step between 0 and 1; text has no axis to sit on."""
|
|
state = MemoryState()
|
|
state.append_history(
|
|
{"f.text": "warm", "f.flag": True, "f.off": False, "f.temp": 21}, ts=1.0
|
|
)
|
|
|
|
assert state.history("f.text") == []
|
|
assert state.history("f.flag") == [(1.0, 1.0)]
|
|
assert state.history("f.off") == [(1.0, 0.0)]
|
|
assert state.history("f.temp") == [(1.0, 21.0)]
|
|
|
|
|
|
def test_a_message_without_history_comes_back_empty():
|
|
assert MemoryState().history("f.never") == []
|