# Streaming

This guide covers working with Python generators from Erlang.

## Overview

Python generators allow processing large datasets or infinite sequences
efficiently by yielding values one at a time. erlang_python supports
two modes of streaming:

1. **Batch streaming** (`py:stream/3,4`, `py:stream_eval/1,2`) - Collects all values into a list
2. **True streaming** (`py:stream_start/3,4`) - Sends events as values are yielded

## True Streaming (Event-driven)

For real-time processing where you need values as they arrive (e.g., LLM tokens,
live data feeds), use `py:stream_start/3,4`:

```erlang
%% Start streaming from a Python iterator
{ok, Ref} = py:stream_start(builtins, iter, [[1,2,3,4,5]]),

%% Receive events as values are yielded
receive_loop(Ref).

receive_loop(Ref) ->
    receive
        {py_stream, Ref, {data, Value}} ->
            io:format("Got: ~p~n", [Value]),
            receive_loop(Ref);
        {py_stream, Ref, done} ->
            io:format("Complete~n");
        {py_stream, Ref, {error, Reason}} ->
            io:format("Error: ~p~n", [Reason])
    after 30000 ->
        timeout
    end.
```

### Events

The stream sends these messages to the owner process:

- `{py_stream, Ref, {data, Value}}` - Each yielded value
- `{py_stream, Ref, done}` - Stream completed successfully
- `{py_stream, Ref, {error, Reason}}` - Stream error

### Options

```erlang
%% Send events to a different process
{ok, Ref} = py:stream_start(Module, Func, Args, #{owner => OtherPid}).
```

### Cancellation

Cancel an active stream:

```erlang
{ok, Ref} = py:stream_start(my_module, long_generator, []),
%% ... receive some values ...
ok = py:stream_cancel(Ref).
%% Stream will stop on next iteration
```

### Async Generators

`stream_start` supports both sync and async generators:

```erlang
%% Async generator (e.g., streaming from an async API)
ok = py:exec(<<"
async def async_gen():
    for i in range(5):
        await asyncio.sleep(0.1)
        yield i
">>),
{ok, Ref} = py:stream_start('__main__', async_gen, []).
```

## Batch Streaming (Collecting All Values)

For simpler use cases where you want all values at once:

### Generator Expressions

```erlang
%% Stream squares of numbers 0-9
{ok, Squares} = py:stream_eval(<<"(x**2 for x in range(10))">>).
%% Squares = [0,1,4,9,16,25,36,49,64,81]

%% Stream uppercase characters
{ok, Upper} = py:stream_eval(<<"(c.upper() for c in 'hello')">>).
%% Upper = [<<"H">>,<<"E">>,<<"L">>,<<"L">>,<<"O">>]

%% Stream filtered values
{ok, Evens} = py:stream_eval(<<"(x for x in range(20) if x % 2 == 0)">>).
%% Evens = [0,2,4,6,8,10,12,14,16,18]
```

### Iterator Objects

Any Python iterator can be streamed:

```erlang
%% Stream from range
{ok, Numbers} = py:stream_eval(<<"iter(range(5))">>).
%% Numbers = [0,1,2,3,4]

%% Stream dictionary items
{ok, Items} = py:stream_eval(<<"iter({'a': 1, 'b': 2}.items())">>).
%% Items = [{<<"a">>, 1}, {<<"b">>, 2}]
```

### Generator Functions

Define generator functions with `yield`:

```erlang
%% Define a generator function
ok = py:exec(<<"
def fibonacci(n):
    a, b = 0, 1
    for _ in range(n):
        yield a
        a, b = b, a + b
">>).

%% Stream from it
%% Note: Functions defined with exec are local to the worker that executes them.
%% Subsequent calls may go to different workers in the pool.
{ok, Fib} = py:stream('__main__', fibonacci, [10]).
%% Fib = [0,1,1,2,3,5,8,13,21,34]
```

For reliable inline generators, use lambda with walrus operator (Python 3.8+):

```erlang
%% Fibonacci using inline lambda - works reliably across workers
{ok, Fib} = py:stream_eval(<<"(lambda: ((fib := [0, 1]), [fib.append(fib[-1] + fib[-2]) for _ in range(8)], iter(fib))[-1])()">>).
%% Fib = [0,1,1,2,3,5,8,13,21,34]
```

## When to Use Each Mode

| Use Case | Recommended API |
|----------|-----------------|
| LLM token streaming | `stream_start/3,4` |
| Real-time data feeds | `stream_start/3,4` |
| Live progress updates | `stream_start/3,4` |
| Batch processing | `stream/3,4` or `stream_eval/1,2` |
| Small datasets | `stream/3,4` or `stream_eval/1,2` |
| One-time collection | `stream/3,4` or `stream_eval/1,2` |

## Memory Considerations

- `stream_start`: Low memory - values processed as they arrive
- `stream/stream_eval`: Values collected into a list - memory grows with output size
- Generators are garbage collected after exhaustion

## Use Cases

### LLM Token Streaming

```erlang
%% Stream tokens from an LLM
{ok, Ref} = py:stream_start(llm_client, generate_tokens, [Prompt]),
stream_to_client(Ref, WebSocket).

stream_to_client(Ref, WS) ->
    receive
        {py_stream, Ref, {data, Token}} ->
            websocket:send(WS, Token),
            stream_to_client(Ref, WS);
        {py_stream, Ref, done} ->
            websocket:send(WS, <<"[DONE]">>);
        {py_stream, Ref, {error, _}} ->
            websocket:send(WS, <<"[ERROR]">>)
    end.
```

### Data Processing Pipelines

```erlang
%% Process file lines (if defined in Python)
{ok, Lines} = py:stream(mymodule, read_lines, [<<"data.txt">>]).

%% Transform each line
Results = [process_line(L) || L <- Lines].
```

### Batch Processing

```erlang
%% Process in batches
ok = py:exec(<<"
def batches(data, size):
    for i in range(0, len(data), size):
        yield data[i:i+size]
">>).
```