xingshu-bci 0.1.0

Official Python SDK for the XingShu BCI Platform local API.

xingshu-bci (Python SDK)

Official Python SDK for the XingShu BCI Platform local API.

Install

pip install xingshu-bci
# or, from source:
pip install -e .

Quick Start

from xingshu_bci import XingShuClient

# Auto-discover the local API, then provide a token explicitly or via
# XINGSHU_BCI_TOKEN.
client = XingShuClient.auto(token="xs_live_...")
print(client.version())
print(client.status())

client.connect(board="synthetic", transport="synthetic")
client.start_recording()
client.insert_marker(value=1, label="trial_start")

for event in client.events(types=["samples", "analysis"]):
    print(event.type, event.payload)
    break  # demo only

client.stop_recording()
client.disconnect()

See docs/SDK.md for the full guide.

Discovering the local API

When the desktop app starts, it writes a bootstrap file here:

%APPDATA%\XingShu BCI\api\bootstrap.json

It contains the dynamic baseUrl and wsUrl. XingShuClient.auto() reads that file so you do not need to hard-code a port.

For authentication, create a token in the Developer Center and pass it either as token=... or through XINGSHU_BCI_TOKEN.

Examples

Live samples

from xingshu_bci import XingShuClient

with XingShuClient.auto(token="xs_live_...") as client:
    print("API version:", client.version().apiVersion)
    print("Status:", client.status().state)

    if client.status().state == "idle":
        client.connect(board="synthetic", transport="synthetic")

    print("Subscribing to events (Ctrl+C to stop)...")
    try:
        for event in client.events(types=["samples", "analysis", "status"]):
            if event.type == "samples":
                data = event.payload.get("data") or []
                if data and data[0]:
                    print(f"samples: ch0[0]={data[0][0]:.3f}")
            elif event.type == "analysis":
                focus = event.payload.get("focus")
                if focus:
                    print(f"focus: {focus}")
            elif event.type == "status":
                print(f"status: {event.payload.get('state')}")
    except KeyboardInterrupt:
        print("Stopped.")

Record with markers

import time
from xingshu_bci import XingShuClient

with XingShuClient.auto(token="xs_live_...") as client:
    client.connect(board="synthetic", transport="synthetic")
    try:
        client.start_recording()
        client.insert_marker(value=1, label="trial_start")
        time.sleep(2)
        client.insert_marker(value=2, label="stimulus_on")
        time.sleep(1)
        client.insert_marker(value=3, label="stimulus_off")
        time.sleep(2)
        client.insert_marker(value=4, label="trial_end")
        print("Recording stopped:", client.stop_recording())
    finally:
        client.disconnect()

Poll focus

import time
from xingshu_bci import XingShuClient

with XingShuClient.auto(token="xs_live_...") as client:
    if client.status().state == "idle":
        client.connect(board="synthetic", transport="synthetic")

    try:
        while True:
            snapshot = client.analysis_focus_latest()
            focus = snapshot.get("focus")
            print("focus: --" if focus is None else f"focus: {focus}")
            time.sleep(0.5)
    except KeyboardInterrupt:
        print("Stopped.")

Focus-driven bulb

from xingshu_bci import SmoothedFocusStream, XingShuClient

class StubBulb:
    def on(self): print("[bulb] ON")
    def off(self): print("[bulb] OFF")
    def set_brightness(self, percent): print(f"[bulb] {percent}%")

with XingShuClient.auto(token="xs_live_...") as client:
    if client.status().state == "idle":
        client.connect(board="synthetic", transport="synthetic")

    bulb = StubBulb()
    for upd in SmoothedFocusStream(client, metric="concentration"):
        if upd.rising:
            bulb.on()
        elif upd.falling:
            bulb.off()
        if upd.is_on:
            bulb.set_brightness(upd.brightness)

License

Proprietary © XingShu BCI.