Streaming between recipes in Python

How one recipe feeds another: the ports, the wire, who decides when the consumer runs, and what a DataBuffer between them changes.

This is the Python side of the mechanism described in src/Core/uprops/docs/streaming.md, which is the C++ one. You do not need that document to read this one, but everything here rests on it, and a question this page does not answer is very likely answered there.

Two scenes ship with the library, built by the examples of the same names, and every listing below is taken from them:

Scene	What it is
assets/examples/recipes/01_recipes_pull.gcx	producer → consumer, read on demand
assets/examples/recipes/02_recipes_push.gcx	producer → DataBuffer → consumer, live

Both are also two clicks away inside gcompose: Recipes panel → + Add → Connected Recipes (pull) and (push), which write the same two recipes into the scene you have open, wired and ready to run.

---

1. What a stream is here

A stream is a sequence of scene objects, read one at a time, through a port that says so. That is all it is. Two quite different things wear it:

• a collection — everything is already there, reading it twice reads the same twice. A recipe that returns a list of boxes produces one of these.
• a queue of what is in flight — items arrive over time, and reading takes them. A DataBuffer relaying a producer is one of these; so is a MuonScatterSource filling from a running simulation.

The port is the same in both cases and so is the wire. What differs is what reading means, and that difference is the single thing to keep in mind on this page: reading a collection is a replay, reading a queue is a consumption.

A stream is not a thread, and does not imply one. Nothing here starts a thread unless you ask for one (StreamMode = "Async", or a decorator's mode="async").

---

2. Declaring the ports

A recipe declares its ports at the top of its own source, as ordinary Python that runs when the recipe is loaded:

from uLib.recipe_agent import StreamInputPort, StreamOutputPort

In  = StreamInputPort("In",  description="what this recipe reads")
Out = StreamOutputPort("Out", description="what this recipe makes")

Each line puts a port on the node, in the node graph and in the Properties panel, coloured as a stream port. The names on the left (In, Out) are the names your run(ctx) body uses; the strings are the labels shown in the UI.

In the panel these two lines are + Add → Stream Input and Stream Output, which write them for you.

When the recipe runs, every declared name is replaced by the value its port actually holds — so In inside run() is not the StreamInputPort object you declared, it is what is on the wire.

---

3. Reading a stream

An input port arrives in run() as a Stream, or as None when nothing is wired to it. Always check:

def run(ctx):
    if In is None:
        print("nothing is wired to In")
        return

There are two ways to read it, and choosing correctly is the whole of section 1 applied:

for item in stream — replay

for item in In:
    ...

Rewinds the producer to its start and walks all of it. Right for a collection.

Wrong for a queue: the rewind is ResetStream(), and on a DataBuffer that empties the queue before reading. A live queue has nothing behind its cursor to go back to, so this reads what happens to be there after the clear — often nothing.

stream.drain() — live

for item in In.drain():
    ...

batch = list(In.drain())        # same thing, as a list
batch = In.take(100)            # at most 100

Takes what has arrived now, one item at a time, and leaves the cursor where it stopped. When the producer has nothing more at this moment, drain() returns — which is not the end of the stream, only the end of this batch. A consumer meant to run again returns here and is called again.

Items drained are gone from the queue. That is the same rule as C++: reading a queue consumes it.

Looking without reading

len(In)        # how many are available (queued, or held by a collection)
In[3]          # the item at that index, without consuming anything

Rewinding on purpose

In.reset()     # start a collection over -- and empty a queue

Called for you by for item in stream. Rarely what you want explicitly.

---

4. Writing a stream

Assign anything iterable to a declared stream output port and it becomes that port's stream:

Out = StreamOutputPort("Out")

def make_items(ctx, count):
    for index in range(int(count)):
        yield ctx.create("ContainerBox", "item_%d" % index)

def run(ctx):
    global Out                       # `global`, or the assignment is local
    Out = make_items(ctx, Count)

A list, a tuple, a generator, or a Stream you were given — all work. Three things are worth knowing:

• global is required. Without it Out = ... binds a local name and the port keeps whatever it had.
• The collection happens when run() returns, not while it yields. The generator above is walked once, at the end of the run, and what it yielded becomes the port's stream. A recipe therefore cannot hand over its first item before its last — which is exactly why a live chain puts a DataBuffer downstream to pace what the recipe made (section 7).
• At most 50 000 items are taken from one assignment; the rest is dropped. A run that means to produce more than that wants a producer object, not a recipe.

You may also assign an existing scene object that is itself a stream producer (a DataBuffer, a MuonScatterSource, a collection), in which case the port simply points at it.

---

5. Making the connection

A wire from one recipe's output port to another's input port is what makes the two a chain. Three ways to make it:

In the node graph — drag from Out to In. What you get is a connection, described below.

From a script —

ctx.connect(producer, "Out", consumer, "In")     # wire it
ctx.disconnect(consumer, "In")                   # remove whatever drives it

In C++ — ReferencePropertyBase::ConnectFrom, which the two above end in; uLib::BuildRecipeStreamChain builds either of the shipped chains in one call.

Connecting, and the one thing you cannot assign

consumer.In = producer.Out        # ValueError: a Stream is not assignable
ctx.connect(producer, "Out", consumer, "In")    # this is how two ports meet

A port's value read from Python is a Stream wrapper, and a wrapper cannot be written into another port — attempting it raises. ctx.connect is how one port is attached to another, and the node graph's edge is the same thing.

Assignment is for an object, and is right when what the consumer should read is a thing in the scene rather than another recipe's port:

consumer.In = buffer              # a DataBuffer *is* a stream producer
consumer.In = ctx["some_collection"]

What the connection adds over pointing at an object:

• it takes what the driving port holds now, so a wire made after the producer has run means something immediately;
• it keeps following that port — including when a later run puts a different object there, which an assignment made once would not see;
• it is what the node graph draws, and what a saved scene restores.

One thing that is easy to over-read: a recipe keeps one producer object per stream output port and refills it on every run, so a consumer pointed straight at that object also sees each new run's items. The connection is still what you want between two recipes — it is the wire the scene stores and the graph shows — but if you are debugging a chain, the items arriving is not by itself proof that the wire is there.

6. Who runs the consumer

Wiring says where the items go. It does not say when anything runs. Two answers:

Explicitly

ctx.run_recipe(producer)
ctx.run_recipe(consumer)

or the Run button, or obj.run(). This is the pull chain: the consumer reads the whole stream because you asked it to.

By announcement — AutoRun

consumer.AutoRun = True

A recipe follows its input ports (Recipe::FollowsInputPorts), so anything announcing itself on a wired input port runs it. Two things announce:

• the producer's port being written at the end of a run (so running a producer runs an AutoRun consumer straight after it), and
• a DataBuffer reaching its NotifyEvery threshold (section 7).

This is the push chain. Runs are coalesced and posted to the application thread: several announcements in quick succession become one run on the latest state, rather than a queue of runs (see docs/code/core/autorun.md).

---

7. Putting a DataBuffer in between

Data/DataBuffer in the Add menu. Between producer and consumer it does three things worth having: it decouples them onto different threads, it bounds how far ahead the producer may get, and it is what wakes the consumer.

producer --(Out -> Input)--> DataBuffer --> consumer
                             Async, 256, every 8    AutoRun

Wire the producer's output into the buffer's Input, and point the consumer's stream input at the buffer object itself — a DataBuffer is a stream producer, so there is no output port to wire from:

ctx.connect(producer, "Out", buffer, "Input")
consumer.In = buffer

Its properties, all in the Stream group:

Property	What it does
StreamMode	"Inline" — pass straight through, no queue, no thread. "Auto" — the same until someone pushes into it. "Async" — a worker thread takes from the producer into the queue; this is the live one.
Capacity	How many items may wait. The bound on memory, and the backpressure: a full buffer holds the producer up.
OverflowPolicy	What a push into a full buffer does: "Block" (wait for room, the default), "DropOldest", "DropNewest".
NotifyEvery	Announce once this many items are queued; 0 never announces. The rate at which an AutoRun consumer runs.
StreamQueueSize	Read-only: how many are waiting right now.

NotifyEvery is the dial worth understanding. It announces once per crossing of the threshold, not once per item: the queue must fall back below it before it announces again. So it sets the pace, not an exact batch size — a consumer draining while the relay keeps filling takes whatever has arrived, which is usually more than the threshold. The shipped push chain relays 400 items through a buffer holding 32, announcing every 8, and the consumer runs about 26 times, averaging 15 items a run.

Set it to 1 and the consumer runs per item, which is correct and mostly overhead. Set it larger than the whole run and it never runs at all.

---

8. A live chain, end to end

This is 02_recipes_push.gcx, which the + Add → Connected Recipes (push) menu entry also writes. Producer:

from uLib.recipe_agent import StreamOutputPort, Property

Out = StreamOutputPort("Out", description="items for the buffer to pace")
Count = Property("Count", default=64, type="int", min=1, max=100000)

def make_items(ctx, count):
    for index in range(int(count)):
        yield ctx.create("ContainerBox", "item_%d" % index)

def run(ctx):
    global Out
    Out = make_items(ctx, Count)

Consumer:

from uLib.recipe_agent import StreamInputPort, Property

In = StreamInputPort("In", description="the buffer feeding this recipe")

_state = {"runs": 0, "total": 0, "last": 0}

def get_runs():  return _state["runs"]
def get_total(): return _state["total"]
def get_last():  return _state["last"]

Runs      = Property("Runs",      getter=get_runs,  type="int")
Total     = Property("Total",     getter=get_total, type="int")
LastBatch = Property("LastBatch", getter=get_last,  type="int")

def run(ctx):
    if In is None:
        return
    batch = list(In.drain())
    _state["runs"] += 1
    _state["last"] = len(batch)
    _state["total"] += len(batch)
    print("run %d: %d new, %d in all"
          % (_state["runs"], len(batch), _state["total"]))

and the scene between them:

buffer.StreamMode = "Async"
buffer.Capacity = 256
buffer.NotifyEvery = 8
consumer.AutoRun = True

ctx.connect(producer, "Out", buffer, "Input")
consumer.In = buffer

Run the producer — that is the only thing anybody runs — and the consumer's Runs, Total and LastBatch climb while it goes. Each run took only what had arrived; no item is counted twice; nothing ever holds more than Capacity items.

Where you run it changes what you see. In the window an AutoRun run is posted to the event loop, so the consumer runs between other things and its counters climb while the buffer is still relaying — the live update. In a batch session (gcompose --batch --recipe ...) there is no event loop to post to, so the announcements are delivered inline and the whole chain has already finished by the time the producer's run returns. The pacing is the same and the counters say so afterwards; there is simply nothing to watch.

Four things have to be true at once for that, and removing any one of them turns it back into a pull chain that happens to have an extra node in it:

1. StreamMode = "Async" — otherwise the buffer relays only when asked, and nothing arrives over time to announce.
2. NotifyEvery > 0 — otherwise nothing announces.
3. AutoRun on the consumer — otherwise the announcement reaches a recipe that does nothing with it.
4. In.drain() in the consumer — otherwise each run rewinds the buffer, which empties it.

---

9. Results the node can show

A plain Property assigned inside run() changes only the script's variable. It is not written back to the node: the write-back at the end of a run covers output ports, not properties.

Seen = Property("Seen", default=0, type="int")

def run(ctx):
    global Seen
    Seen = 42            # the node still shows 0

Keep the value in the script and declare a computed property whose getter returns it — the getter is called every time anything reads the port, so the panel, the node and any wire see the current value:

_state = {"seen": 0}

def get_seen():
    return _state["seen"]

Seen = Property("Seen", getter=get_seen, type="int")

def run(ctx):
    _state["seen"] = 42          # the node shows 42

The recipe's namespace outlives the run — the source is re-executed only when it changes — which is what lets the getter still answer afterwards, and what lets _state accumulate across runs, as the live consumer above does.

+ Add → Computed Property writes this shape for you.

---

10. The streaming decorators

uLib.recipe_agent offers three decorators for the shape of a stream stage. They are conveniences over plain generators, not a separate mechanism:

from uLib.recipe_agent import stream_producer, stream_filter, stream_consumer

@stream_producer
def make(ctx, n):
    for i in range(n):
        yield ctx.create("ContainerBox", "item_%d" % i)

@stream_filter
def big_enough(item):           # bool -> keep or drop
    return item.Size[0] > 1.0

@stream_filter
def doubled(item):              # anything else -> the transformed item
    item.Size = [2 * c for c in item.Size]
    return item

@stream_consumer
def count_it(item):
    _state["seen"] += 1

def run(ctx):
    global Out
    Out = big_enough(make(ctx, int(Count)))

A @stream_filter function is used as a predicate when it returns a bool, as a transformation when it returns anything else, and as a stage when it is itself a generator over the incoming stream.

Each takes mode=:

• mode="inline" (the default) — runs on the caller's thread, no queue, no thread overhead.
• mode="async" — runs the body on a worker thread feeding a bounded queue.Queue (capacity=, default 1024). This is Python-side decoupling, entirely separate from a DataBuffer, and the two are rarely both needed.

@stream_filter(numba=True) compiles the predicate with @njit(fastmath=True) when numba is installed, and is a plain call when it is not. See docs/NUMBA_RECIPES.md.

+ Add writes any of the three, and StreamIO writes an input port, an output port and a numba filter joining them.

---

11. Threads, and what is safe

What	Which thread
Run button / ctx.run_recipe	the application (GUI) thread
Play / obj.start()	a worker thread
An AutoRun run	posted to the application thread, coalesced
DataBuffer in "Async" mode	its own worker thread, for the relay only
@stream_*(mode="async")	a Python worker thread

Two rules follow, and both have cost real crashes:

• Scene objects belong to the application thread. Creating, deleting or re-pointing objects from a worker is not safe; read what you need there and do the editing on the application thread. An AutoRun consumer is already on the right thread, which is one more reason to prefer it to a thread of your own.
• A Geant::Scene belongs to the thread that built it, and refuses to run from another — it says so in its StreamProblem property rather than crashing. A reconstruction driving a scene's stream from a worker has its work posted back automatically; there is nothing to do but not fight it.

---

12. Reading a live C++ stream

Nothing above is specific to recipes. Any object that produces a stream reads the same way, including the ones a simulation fills:

poca = ctx["poca"]                     # its Stream port is wired to the source
for muon in poca.Stream.drain():       # what has been measured so far
    ...

# or reach an object that is a producer directly, with no port in between:
source = ctx["muon_stream"]            # a MuonScatterSource
for muon in Stream(source).drain():
    ...

Stream is in a recipe's namespace already; the wrapper is only needed when you have the producing object rather than a port pointing at it.

MuonScatterSource has the same dial as the buffer — NotifyEvery, plus Capacity, BatchSize, WaitTimeout — and reports Queued, Pushed, Dropped and Driven for the panel. Dropped above zero means the consumer is slower than the producer and the queue was full; 19_streaming_poca and 20_driven_stream are the worked examples.

---

13. Things that catch people

Symptom	Cause
The consumer reads nothing	Nothing wired (In is None), or the producer has not run yet.
The consumer reads the same items again every run	It uses for item in In where it wants In.drain().
The buffer empties when the consumer runs	The same: iterating rewinds, and a rewind on a queue is a clear.
The consumer never runs by itself	AutoRun off, or NotifyEvery is 0, or StreamMode is not "Async".
The consumer runs once per item	NotifyEvery is 1.
The node shows a stale value	A plain Property was assigned in run(); use a computed one (section 9).
The port empties after the producer runs again	The two were assigned, not connected (section 5).
Items go missing under load	Capacity too small with OverflowPolicy dropping; "Block" instead, or a bigger capacity.
A generator assigned to an output yields nothing	It was consumed already — a generator can only be walked once.

---

14. DataSource, DataSink, and DataBuffer are different tools

Their names are similar but they move different things:

Type	Moves	Where to use it
Core.DataSource	one byte payload from memory or a file	standalone Python, or as an input asset for an image/mesh
Core.DataSink	one byte payload to memory or a file	standalone Python, or as an output asset
DataBuffer	scene objects through a bounded live queue, and optionally an encoded scene payload	a gcompose scene/Recipe pipeline

A source and sink have direct pybind APIs for bytes:

from uLib import Core

source = Core.DataSource("calibration")
source.set_data(b'{"gain": 1.25}')

sink = Core.DataSink("copy")
sink.write_source(source)
assert sink.get_data() == b'{"gain": 1.25}'

This does not make each byte a stream item. A stream item is a scene Object, such as a reconstructed muon or a box made by a Recipe. To connect a byte source to a domain object, give that object's source port a DataSource; VoxImage and TessellatedMesh expose SetSource(), ReadFromDataSource(), SetSink(), and WriteToDataSink() in the direct Math binding.

DataBuffer is a registered scene class, not a standalone Core.DataBuffer class. Configure it through the gcompose ctx facade:

# gcompose console or Recipe
buffer = ctx.create("DataBuffer", "muon_queue")
buffer.StreamMode = "Async"
buffer.Capacity = 256
buffer.OverflowPolicy = "Block"
buffer.NotifyEvery = 16

ctx.connect(producer, "Out", buffer, "Input")
consumer.In = buffer
consumer.AutoRun = True

The consumer then calls In.drain(). It receives scene objects from the buffer queue; it does not read the bytes held by a DataSource. See Core, contexts and data I/O for the file/embedded-data lifecycle.

15. Reference

Declarations (uLib.recipe_agent): StreamInputPort(name, description=), StreamOutputPort(name, description=), Property(name, default=, type=, min=, max=, step=, getter=, setter=, description=), Action(name, handler=).

On a Stream: for item in s (replay), s.drain(limit=None) (live), s.take(n), s.reset(), len(s), s[i], s.producer.

On ctx: ctx.connect(source, "OutPort", target, "InPort"), ctx.disconnect(target, "InPort"), ctx.run_recipe(recipe), ctx.set_recipe_source(recipe, source), ctx.recipe_ports(recipe), ctx.create(class_name, name), ctx.find(name), ctx["name"].

Decorators: @stream_producer(mode=, capacity=), @stream_filter(mode=, numba=, capacity=), @stream_consumer(mode=).

C++: uLib::BuildRecipeStreamChain(context, kind) builds either chain; uLib::StreamInput is how a C++ consumer reads one of these ports; the implementation reference is src/Core/uprops/docs/streaming.md.