04 — Streaming: process data in chunks

g.forward(x, stream=True, chunk_size=N) splits the input along the first dimension and pushes chunks through the filter chain — memory stays proportional to the chunk, not the dataset. Each filter declares how it behaves per chunk via _stream_mode:

"fixed" (default): state is frozen from fit(); chunks are independent.
"evolving": state updates chunk to chunk (online learning).
"barrier": must see everything → forces materialization at that node.

from soma import Filter, Graph

class Scaler(Filter):
    _cache_version = "nb04-scaler-v1"

    def fit(self, x, y=None):
        return {"max": max(abs(v) for v in x) or 1.0}
    def forward(self, x, state):
        return [v / state["max"] for v in x]

g = Graph()
g.node("scaler", Scaler())
g.fit([100.0, 200.0, 300.0, 400.0, 500.0])

out = g.forward([50.0, 150.0, 250.0, 350.0, 450.0, 550.0], stream=True, chunk_size=2)
print([round(v, 2) for v in out])

[0.1, 0.3, 0.5, 0.7, 0.9, 1.1]

4.1 — Chunk-level caching

In streaming mode each chunk’s result is cached by hash(config + state + chunk). Re-streaming overlapping data reuses the chunks already seen — useful for sliding-window inference.

CALLS = {"n": 0}

class Counting(Filter):
    _cache_version = "nb04-v1"
    def forward(self, x, state):
        CALLS["n"] += 1
        return [v * 2 for v in x]

g = Graph()
g.node("counting", Counting())
g.fit([1.0])

data = [float(i) for i in range(8)]
g.forward(data, stream=True, chunk_size=2)
first = CALLS["n"]
g.forward(data, stream=True, chunk_size=2)   # same chunks again
print(f"first pass: {first} chunk executions; second pass: {CALLS['n'] - first} new executions")

first pass: 5 chunk executions; second pass: 0 new executions

4.2 — Barrier filters

Some computations (sorting, global statistics) can’t be chunked. Mark them _stream_mode = "barrier": upstream still streams, the barrier node materializes.

class GlobalSort(Filter):
    _stream_mode = "barrier"
    _cache_version = "nb04-sort-v1"
    def forward(self, x, state):
        return sorted(x)

g = Graph.somatize(Counting() >> GlobalSort())
g.fit([1.0])
print(g.forward([5.0, 1.0, 4.0, 2.0, 3.0], stream=True, chunk_size=2))

[2.0, 4.0, 6.0, 8.0, 10.0]

Streaming composes with everything else: distribution to workers (chunks travel as WebSocket binary frames) and the persistent cache (chunk results survive restarts).