Enterprise

# Stream data

> Get samples as the Core receives them with a streamer in the Python and TypeScript clients: read frames, change channels, slow the stream, and close the streamer.

A streamer gets samples from the Core as the Core receives them. Use it to process live data, to show live values in your own app, or to react to a value in a script. A streamer gets only new samples. To get stored samples, [read data](https://docs.synnaxlabs.com/reference/client/read-data). Samples from `client.write` and from writers in `persist` mode do not reach streamers, as [writer modes](https://docs.synnaxlabs.com/reference/client/write-data#writer-modes) says.

## Open a streamer and read frames

[Enterprise](https://docs.synnaxlabs.com/reference/get-started/editions "Needs Synnax Enterprise")

**Python**

```python
import synnax as sy

client = sy.Synnax()

with client.open_streamer(["time", "temperature"]) as streamer:
    for frame in streamer:
        print(frame["temperature"].to_numpy())
```

The loop waits for each frame. The `with` block closes the streamer, also when the code raises an error.

**TypeScript**

```typescript
const streamer = await client.openStreamer(["time", "temperature"]);
try {
  for await (const frame of streamer) {
    console.log(Array.from(frame.get("temperature")));
  }
} finally {
  streamer.close();
}
```

The loop waits for each frame. The `finally` block closes the streamer, also when the code throws an error.

Each frame holds the samples of one write, as a [frame](https://docs.synnaxlabs.com/reference/client/series-and-frames#frames). A frame can hold samples for only some of the channels, so check before you use a channel:

[Enterprise](https://docs.synnaxlabs.com/reference/get-started/editions "Needs Synnax Enterprise")

**Python**

```python
for frame in streamer:
    if "temperature" in frame:
        latest = frame["temperature"][-1]
```

**TypeScript**

```typescript
for await (const frame of streamer) {
  const temperature = frame.get("temperature");
  if (temperature.length > 0) {
    const latest = temperature.at(-1);
  }
}
```

### Read one frame

[Enterprise](https://docs.synnaxlabs.com/reference/get-started/editions "Needs Synnax Enterprise")

**Python**

`read` waits for the next frame. With a timeout, it returns `None` when no frame comes in time:

```python
frame = streamer.read(timeout=5)
if frame is None:
    print("No data in 5 seconds")
```

The timeout is a number of seconds or a `TimeSpan`.

**TypeScript**

`read` waits for the next frame:

```typescript
const frame = await streamer.read();
```

`read` has no timeout. When the Core sends nothing for 15 seconds, `read` throws, so a lost connection does not stop your code.

## Stream with asyncio in Python

`open_async_streamer` gives a streamer for `asyncio` code:

```python
async with await client.open_async_streamer(["time", "temperature"]) as streamer:
    async for frame in streamer:
        print(frame["temperature"].to_numpy())
```

The async streamer has no per-call timeout on `read`, and it cannot change its channels. Both Python streamers detect a lost connection through keep-alive messages. With the default `keep_alive` of 5 seconds, a read fails after 15 seconds without a message. The TypeScript streamer is always async.

## Change the channels

To stream other channels, give the streamer a new list. The list replaces the old channels:

[Enterprise](https://docs.synnaxlabs.com/reference/get-started/editions "Needs Synnax Enterprise")

**Python**

```python
streamer.update_channels(["time", "temperature", "pressure"])
```

**TypeScript**

```typescript
await streamer.update(["time", "temperature", "pressure"]);
```

## Slow the stream

When your code reads more slowly than frames come, the Core drops frames. To get all the samples, slow the stream, or read the stored data later. Two options reduce the data:

[Enterprise](https://docs.synnaxlabs.com/reference/get-started/editions "Needs Synnax Enterprise")

**Python**

```python
streamer = client.open_streamer(
    ["time", "temperature"], downsample_factor=10, throttle_rate=10
)
```

**TypeScript**

```typescript
import { Rate } from "@synnaxlabs/client";

const streamer = await client.openStreamer({
  channels: ["time", "temperature"],
  downsampleFactor: 10,
  throttleRate: Rate.hz(10),
});
```

* `downsample_factor` -> keep one of each `N` samples in each series. The Core drops the other samples and does not average them. A write of fewer than `N` samples is not reduced.
* `throttle_rate` -> send at most this many frames each second. The Core joins the samples between two sends into one frame, so the streamer gets all samples in fewer frames.

## Close the streamer

[Enterprise](https://docs.synnaxlabs.com/reference/get-started/editions "Needs Synnax Enterprise")

**Python**

```python
streamer.close()
```

**TypeScript**

```typescript
streamer.close();
```

A streamer that you do not close keeps its connection to the Core open.

## Streamer options

[Enterprise](https://docs.synnaxlabs.com/reference/get-started/editions "Needs Synnax Enterprise")

**Python**

* [`channels`](https://docs.synnaxlabs.com/reference/client/stream-data#python-channels)`list[str] | list[int]`

  The names or keys of the channels to stream.

* [`downsample_factor`](https://docs.synnaxlabs.com/reference/client/stream-data#python-downsample-factor)`int = 1`

  Keep one of each `downsample_factor` samples in each series.

* [`throttle_rate`](https://docs.synnaxlabs.com/reference/client/stream-data#python-throttle-rate)`float = 0`

  The largest number of frames each second, in Hz. With 0, the Core sends each frame when it receives it.

* [`keep_alive`](https://docs.synnaxlabs.com/reference/client/stream-data#python-keep-alive)`CrudeTimeSpan = 5 s`

  How often the Core sends a keep-alive message. A read raises a timeout error after three intervals with no message. Use at least 2 seconds, or 0 to turn off the check.

**TypeScript**

* [`channels`](https://docs.synnaxlabs.com/reference/client/stream-data#typescript-channels)`string[] | number[]`

  The names or keys of the channels to stream.

* [`downsampleFactor`](https://docs.synnaxlabs.com/reference/client/stream-data#typescript-downsample-factor)`number = 1`

  Keep one of each `downsampleFactor` samples in each series.

* [`throttleRate`](https://docs.synnaxlabs.com/reference/client/stream-data#typescript-throttle-rate)`Rate = 0`

  The largest number of frames each second. With 0, the Core sends each frame when it receives it.

* [`keepAlive`](https://docs.synnaxlabs.com/reference/client/stream-data#typescript-keep-alive)`TimeSpan = 5 s`

  How often the Core tells the streamer that the connection is alive. When three intervals pass with no message, `read` throws. The value must be at least 2 s. With `TimeSpan.ZERO`, `read` waits with no limit.