> For the complete documentation index, see [llms.txt](https://docs.anthriq.com/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://docs.anthriq.com/bxi-studio/developer/overview/features/exg.md).

# Stream EXG

> **Applies to:** [Instinct](/bxi-studio/developer/overview/devices/instinct.md) and [xSys](/bxi-studio/developer/overview/devices/xsys.md). Not available on [xBud](/bxi-studio/developer/overview/devices/xbud.md).

The `eeg` feature delivers continuous ADC samples. Confirm electrode contact first: a stream from a poorly seated electrode records noise that no later processing recovers. See [Measure contact impedance](/bxi-studio/developer/overview/features/impedance.md).

The lifecycle and frame format are described in [Stream lifecycle](/bxi-studio/developer/overview/concepts/streams.md). This page applies them.

> **Note:** The feature name passed to the SDK is `"eeg"`, not `"exg"`. The signal is EXG; the API identifier kept the older spelling.

## Start a stream

{% tabs %}
{% tab title="Node.js" %}

```typescript
const streamId = 20;

await client.features.executeOperation("eeg", "add_stream", {
  stream: {
    stream_id: streamId,
    protocol: "websocket",
    host: "127.0.0.1",
    port: 9020,
    elements_before_flush: 80,
  },
});

// Let the stream server bind.
await new Promise((resolve) => setTimeout(resolve, 1000));
```

{% endtab %}

{% tab title="Python" %}

```python
import asyncio

stream_id = 20

await client.execute_operation("eeg", "add_stream", {
    "stream": {
        "stream_id": stream_id,
        "protocol": "websocket",
        "host": "127.0.0.1",
        "port": 9020,
        "elements_before_flush": 80,
    },
})

# Let the stream server bind.
await asyncio.sleep(1)
```

{% endtab %}
{% endtabs %}

| Field                   | Type          | Description                              |
| ----------------------- | ------------- | ---------------------------------------- |
| `stream_id`             | `number`      | Identifier used by every later operation |
| `protocol`              | `"websocket"` | Transport for stream frames              |
| `host`                  | `string`      | Bind address, usually `127.0.0.1`        |
| `port`                  | `number`      | Port for the stream server               |
| `buffer_size`           | `number`      | Optional transport buffer size           |
| `elements_before_flush` | `number`      | Samples per batch                        |

At 4 kHz, `elements_before_flush: 80` gives a batch every 20 ms.

> **Note:** The SDK sets `stream_type` from the feature name. Do not set it yourself.

## Receive samples

{% tabs %}
{% tab title="Node.js" %}

```typescript
let framesReceived = 0;

const subscription = await client.features.subscribeToOperation(
  "eeg",
  "stream",
  (update) => {
    if (!update.success) return;
    framesReceived++;

    for (const frame of update.data.details.payload.frames ?? []) {
      for (const channel of frame.channels ?? []) {
        if (channel.has_error) {
          console.warn(`Channel ${channel.channel_id}: error ${channel.error_info}`);
          continue;
        }
        process(channel.channel_id, channel.adc_data);
      }
    }
  },
  { stream_id: streamId }
);

const subscriptionId = subscription.data.subscriptionId;

// Let the WebSocket client connect.
await new Promise((resolve) => setTimeout(resolve, 2000));

await client.features.executeOperation("eeg", "start_stream", { stream_id: streamId });
```

{% endtab %}

{% tab title="Python" %}

```python
import asyncio

frames_received = 0

def on_frame(update):
    global frames_received
    if not update.success:
        return

    frames_received += 1
    payload = update.data["details"]["payload"]
    for frame in payload.get("frames", []):
        for channel in frame.get("channels", []):
            if channel["has_error"]:
                print(f"Channel {channel['channel_id']}: error {channel['error_info']}")
                continue
            process(channel["channel_id"], channel["adc_data"])

subscription = await client.subscribe_to_operation(
    "eeg", "stream", on_frame, {"stream_id": stream_id}
)
subscription_id = subscription.subscription_id

# Let the WebSocket client connect.
await asyncio.sleep(2)

await client.execute_operation("eeg", "start_stream", {"stream_id": stream_id})
```

{% endtab %}
{% endtabs %}

`adc_data` is a signed 24-bit sample, already sign-extended, so a negative value reads as negative.

`channel_id` is the absolute channel, combining the ADC and its local channel. Which physical electrode that corresponds to is set by the Synap register's `connected_nerv_id` and `connected_nerv_channel_id`. See [Configure channels](/bxi-studio/developer/overview/features/channels.md).

## Keep the callback fast

At 4 kHz with 80-sample batches the callback runs about 50 times a second. Work done inside it delays every other operation on the event loop.

Hand frames off rather than processing them in place:

{% tabs %}
{% tab title="Node.js" %}

```typescript
const queue = [];
let draining = false;

await client.features.subscribeToOperation(
  "eeg",
  "stream",
  (update) => {
    if (update.success) queue.push(update);
    if (!draining) void drain();
  },
  { stream_id: streamId }
);

async function drain() {
  draining = true;
  while (queue.length > 0) {
    await analyse(queue.shift());
  }
  draining = false;
}
```

{% endtab %}

{% tab title="Python" %}

```python
queue: asyncio.Queue = asyncio.Queue()

await client.subscribe_to_operation(
    "eeg", "stream", queue.put_nowait, {"stream_id": stream_id}
)

async def drain():
    while True:
        update = await queue.get()
        await analyse(update)

task = asyncio.create_task(drain())
```

{% endtab %}
{% endtabs %}

An unbounded queue grows without limit if the consumer is slower than the device. Cap it and count drops rather than letting memory climb through a long recording.

## Stop and clean up

{% tabs %}
{% tab title="Node.js" %}

```typescript
await client.features.executeOperation("eeg", "stop_stream", { stream_id: streamId });

// Buffered frames keep arriving; let them drain.
await new Promise((resolve) => setTimeout(resolve, 1500));

await client.features.unsubscribeFromOperation("eeg", "stream", { subscriptionId });
await client.features.executeOperation("eeg", "remove_stream", { stream_id: streamId });
```

{% endtab %}

{% tab title="Python" %}

```python
import asyncio

await client.execute_operation("eeg", "stop_stream", {"stream_id": stream_id})

# Buffered frames keep arriving; let them drain.
await asyncio.sleep(1.5)

await client.unsubscribe_from_operation(
    "eeg", "stream", {"subscriptionId": subscription_id}
)
await client.execute_operation("eeg", "remove_stream", {"stream_id": stream_id})
```

{% endtab %}
{% endtabs %}

Unsubscribe with the feature and operation used to subscribe, which are `"eeg"` and `"stream"`.

> **Warning:** A stream that is not removed keeps the device transmitting after the process exits. Put these four calls in a `finally` block.

## Change configuration without restarting

`update_stream` adjusts a running stream:

{% tabs %}
{% tab title="Node.js" %}

```typescript
await client.features.executeOperation("eeg", "update_stream", {
  stream_id: streamId,
  updates: { elements_before_flush: 60 },
});
```

{% endtab %}

{% tab title="Python" %}

```python
await client.execute_operation("eeg", "update_stream", {
    "stream_id": stream_id,
    "updates": {"elements_before_flush": 60},
})
```

{% endtab %}
{% endtabs %}

## Complete example

{% tabs %}
{% tab title="Node.js" %}

```typescript
import { BxiClient } from "@anthriq_dev/bxi-interface";

async function recordEeg(durationMs: number): Promise<number> {
  const client = new BxiClient<"anthriq-instinct">({ deviceType: "anthriq-instinct" });
  const streamId = 20;
  let subscriptionId: string | undefined;
  let frames = 0;

  await client.initialize("localhost");
  await client.connect();

  try {
    await client.features.executeOperation("eeg", "add_stream", {
      stream: {
        stream_id: streamId,
        protocol: "websocket",
        host: "127.0.0.1",
        port: 9020,
        elements_before_flush: 80,
      },
    });
    await new Promise((r) => setTimeout(r, 1000));

    const subscription = await client.features.subscribeToOperation(
      "eeg",
      "stream",
      (update) => {
        if (update.success) frames++;
      },
      { stream_id: streamId }
    );
    subscriptionId = subscription.data.subscriptionId;
    await new Promise((r) => setTimeout(r, 2000));

    await client.features.executeOperation("eeg", "start_stream", { stream_id: streamId });
    await new Promise((r) => setTimeout(r, durationMs));
    await client.features.executeOperation("eeg", "stop_stream", { stream_id: streamId });
    await new Promise((r) => setTimeout(r, 1500));

    return frames;
  } finally {
    if (subscriptionId) {
      await client.features.unsubscribeFromOperation("eeg", "stream", { subscriptionId });
    }
    await client.features.executeOperation("eeg", "remove_stream", { stream_id: streamId });

    if (client.isConnected()) await client.disconnect();
    await client.shutdown();
  }
}
```

{% endtab %}

{% tab title="Python" %}

```python
import asyncio
from anthriq_bxi_interface import BxiClient

async def record_exg(duration_s: float) -> int:
    client = BxiClient(device_type="anthriq-instinct")
    stream_id = 20
    subscription_id = None
    frames = 0

    def on_frame(update):
        nonlocal frames
        if update.success:
            frames += 1

    await client.initialize("localhost")
    await client.connect()

    try:
        await client.execute_operation("eeg", "add_stream", {
            "stream": {
                "stream_id": stream_id,
                "protocol": "websocket",
                "host": "127.0.0.1",
                "port": 9020,
                "elements_before_flush": 80,
            },
        })
        await asyncio.sleep(1)

        subscription = await client.subscribe_to_operation(
            "eeg", "stream", on_frame, {"stream_id": stream_id}
        )
        subscription_id = subscription.subscription_id
        await asyncio.sleep(2)

        await client.execute_operation("eeg", "start_stream", {"stream_id": stream_id})
        await asyncio.sleep(duration_s)
        await client.execute_operation("eeg", "stop_stream", {"stream_id": stream_id})
        await asyncio.sleep(1.5)

        return frames
    finally:
        if subscription_id:
            await client.unsubscribe_from_operation(
                "eeg", "stream", {"subscriptionId": subscription_id}
            )
        await client.execute_operation("eeg", "remove_stream", {"stream_id": stream_id})
        await client.shutdown()
```

{% endtab %}
{% endtabs %}

## Next steps

* [Run parallel streams](/bxi-studio/developer/overview/features/parallel.md)
* [Measure contact impedance](/bxi-studio/developer/overview/features/impedance.md)
* [Record and play back](/bxi-studio/developer/overview/backend-services/recording.md)