Sources¶
A Source wraps a device, file, or transport behind a uniform interface. Built-in sources live here; custom sources implement the Source protocol below. See Add a custom source for the recipe.
The protocol¶
Bases: Protocol
Protocol every data source must implement.
Three methods, no inheritance. Stream wraps any object matching
this Protocol and runs it on a daemon acquisition thread, so read
must not block.
See Add a custom source for worked examples and the full contract.
Poll for the next chunk of samples.
Return (data, ts) where data is sample-major
(n_samples, n_channels) and ts is (n_samples,)
float64 LSL clock timestamps. Return (None, None) if no
new data is available.
Release the device.
Idempotent - may be called multiple times during shutdown.
Built-in sources¶
Pull samples from a Lab Streaming Layer outlet by name.
The default real-time source for MyoGestic: name the outlet you want
on your local LSL network, drop the source into a Stream,
and the framework's acquisition thread handles the rest. Uses
mne_lsl under the hood.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
stream_name
|
str
|
LSL outlet name to subscribe to (e.g. |
required |
dtype
|
DTypeLike | None
|
Dtype the samples are stored as (one of
|
'float32'
|
Examples:
>>> from myogestic import Stream
>>> from myogestic.sources import LSLSource
>>> stream = Stream("emg", source=LSLSource("TestEMG1"),
... window_ms=1000)
>>> # keep a 16-bit amp's native format to halve memory / disk:
>>> raw = LSLSource("TestEMG1", dtype=None)
The source is non-blocking: read pulls whatever is
immediately available from the inlet and returns (None, None)
when the outlet has produced nothing new since the last call. The
acquisition thread paces itself by waiting for the inlet to fill,
so a fast spin loop is harmless.
connect() -> StreamInfo
Resolve the outlet by name and open an inlet.
Returns a StreamInfo whose channel count and sample rate
come from the outlet's metadata. dtype is the value requested
at construction (default float32), or the outlet's native wire
format when constructed with dtype=None. Blocks up to 10 s
waiting for the outlet to appear on the network.
Raises:
| Type | Description |
|---|---|
RuntimeError
|
if no outlet with |
Pull whatever samples are immediately available.
Non-blocking. Returns (data, timestamps) where data is
(n_samples, n_channels) in the configured dtype (default
float32) and timestamps is a 1-D float64 array of LSL clock
seconds. Returns (None, None) if the inlet hasn't been opened
or no new samples are pending.
reconnect(target: str | None = None) -> StreamInfo
Reconnect to the same or a different LSL stream.
Replays a recorded session as if it were a live stream.
Accepts either a folder-format session or a .session.zip archive —
delegates to open_session_store so both layouts work transparently.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
session_path
|
str
|
Session folder or |
required |
stream_name
|
str
|
Which recorded stream to replay. |
required |
speed
|
float
|
Playback rate multiplier. |
1.0
|
Examples:
>>> from myogestic import Stream
>>> from myogestic.sources import ReplaySource
>>> source = ReplaySource("sessions/demo.session.zip", "emg")
>>> stream = Stream("emg", source, window_ms=1000)
connect() -> StreamInfo
Open the recorded session and return its StreamInfo.
Raises ValueError if the requested stream name is not
present in the session.
Return the next recorded chunk, paced to wall-clock time x speed.
Returns (None, None) when no samples are due yet; loops back
to the start of the recording once the end is reached.
Close the session and rewind the replay position.
Closing releases the ZipStore — on Windows an open handle keeps the
.session.zip locked, so the file couldn't be deleted or re-recorded
until the source was garbage-collected.
SyntheticSource(n_channels: int = 8, fs: float = 2048.0, *, noise: float = 0.12, hum: float = 0.35, hum_hz: float = 50.0, activation: float = 1.0, direction: float = 0.0, require_target: bool = False)
In-process synthetic EMG — sine waves plus noise, paced by the wall clock.
A stand-in for a real amplifier while you build an app, demonstrate one, or
run a test. It implements the same connect / read / disconnect
contract as LSLSource and the OTB sources, plus the optional
discover / reconnect extensions, so every widget behaves exactly as
it would against hardware.
Each channel carries one distinct sine (5 Hz, 6 Hz, …) with a shared mains
hum on top, so the signal viewer's notch filter and per-channel controls
have something real to act on. noise, hum and hum_hz are public
and can be changed while streaming.
This is a test signal, not a model of EMG: real EMG is a stochastic, burst-modulated process, and this is a fixed tone with white noise on it. It exercises the plot, the filters, the scaling and the recording path faithfully; a classifier trained on it learns which frequency a channel is.
Timestamps come from the mne_lsl local_clock() domain rather than a
relative counter, so "now" and last-sample age read correctly.
read blocks until the next chunk is due, the way a real inlet's blocking
pull does. Without that the acquire thread spins and squashes thousands of
chunks onto one x-position.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
n_channels
|
int
|
Channel count the |
8
|
fs
|
float
|
Sample rate in Hz. |
2048.0
|
noise
|
float
|
Gaussian noise standard deviation, against a sine of amplitude 1. Also a live attribute — assign to it while streaming and the next chunk follows, no reconnect. |
0.12
|
hum
|
float
|
Mains-hum amplitude, common-mode across every channel. Live, like
|
0.35
|
hum_hz
|
float
|
Mains-hum frequency, 50 Hz in most of the world and 60 Hz in the Americas. Live — drag it away from whichever the viewer's Notch is set to and the hum comes back, which is the notch's bandwidth made visible. |
50.0
|
require_target
|
bool
|
When |
False
|
Examples:
>>> from myogestic import Stream
>>> from myogestic.sources import SyntheticSource
>>> stream = Stream("emg", source=SyntheticSource(n_channels=8), window_ms=1000)
>>> stream.reconnect() # nothing attaches a stream on its own
True
Return the next chunk, blocking until it is due.
Report the single fake device, so Scan → Connect flows work.
reconnect(target: str | None = None) -> StreamInfo
Reconnect, optionally selecting a discovered target first.
SyntheticForceSource(*, fs: float = 100.0, zero: float = 0.3, span: float = 2.0, effort: float = 0.0, noise: float = 0.01, lag_s: float = 0.25)
A synthetic force channel you drive by hand, for testing without a transducer.
It stands in for the transducer, not for the subject. You are the
subject: watch the target and move effort, exactly as the person on a
real load cell watches the plot and pushes. Nothing here tracks the target
on its own — a channel that did would draw a tracking error nobody produced,
and that error is the number the whole task exists to measure.
SyntheticSource cannot stand in for this. Its channels are fixed sine
waves, so there is no resting level to zero and no peak to calibrate against.
It emits in arbitrary volts, not in % MVC, precisely so the calibration is
not free: the resting reading is zero and a full effort is
zero + span, so capturing Zero and MVC does real work and a task run
against a bad calibration reads wrong, exactly as it would with hardware.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
fs
|
float
|
Sample rate in Hz. |
100.0
|
zero
|
float
|
Resting reading in volts, the transducer's own offset. Non-zero on purpose: a calibration that assumes the rest reading is 0 is the mistake this exists to let you make and see. |
0.3
|
span
|
float
|
Volts between rest and a full (100 % MVC) effort. |
2.0
|
effort
|
float
|
Live, 0..1. What you are producing, and the only thing driving the channel. Raise it to capture an MVC, then move it to follow the target once a block is running. |
0.0
|
noise
|
float
|
Live. Gaussian noise standard deviation in volts. |
0.01
|
lag_s
|
float
|
Live. First-order smoothing, in seconds, so dragging the slider reads as
a contraction rather than as a step. |
0.25
|
Examples:
TargetSource takes any Trajectory — a Trapezoid in percent of MVC, a Pursuit in signed [-1, +1] control units, or your own shape with the same three members. The channel is named target_pct whichever it is: the name predates the signed trajectories and renaming it would orphan every recording made so far, so the unit is the trajectory's, not the channel name's.
TargetSource(trajectory: Trajectory | None = None, fs: float = 100.0)
Streams a Trajectory target as a recordable two-channel signal.
Implements the same connect / read / disconnect contract as the
amplifier sources, so a Stream records it exactly as it records EMG and the two
line up sample for sample. Timestamps come from the mne_lsl local_clock()
domain, the same domain every other source stamps in — that shared clock is the
whole point, and is what lets an offline script put target and EMG on one axis.
The stream runs from connect() to disconnect(); start and stop control
the task, not the stream. While stopped it keeps emitting baseline rather than
going quiet, because a source that stops producing leaves a hole in the recording
exactly where the operator was setting the block up — and a hole is indistinguishable
from a dropout at analysis time.
read blocks until the next chunk is due, the way a real inlet's blocking pull
does, so the acquire thread paces to fs instead of spinning.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
trajectory
|
Trajectory | None
|
The shape to follow — a |
None
|
fs
|
float
|
Sample rate in Hz. The target is smooth and slow, so this does not need to match the amplifier's rate; the timestamps are what align the two, not the rate. |
100.0
|
Attributes:
| Name | Type | Description |
|---|---|---|
trajectory |
Live. Assign while streaming and the next chunk follows the new shape — what the editing UI does as a segment is dragged. Read once per chunk, and a plain assignment is atomic under the GIL, so no lock is needed. |
Examples:
>>> from myogestic import Stream
>>> from myogestic.sources.target import TargetSource
>>> from myogestic.tracking import Trapezoid
>>> source = TargetSource(Trapezoid(level_pct=30.0))
>>> stream = Stream("target", source=source, window_ms=10_000)
>>> source.start() # begins the block; the stream records throughout either way
property
¶
elapsed: float
Task time of the newest emitted sample, in seconds. 0.0 while stopped.
Reported from the sample's own timestamp rather than from a fresh clock reading, so it names the point on the trajectory that was actually recorded.
Return the next chunk of target samples, blocking until it is due.
Stop the block. Nothing to release — the generator holds no handle.
module-attribute
¶
PHASE_CODES: dict[str, int] = {'rest': 0, 'ramp_up': 1, 'hold': 2, 'ramp_down': 3, 'recover': 4, 'done': 5, 'idle': 6}
Optional — requires the serial extra
SerialSource is import-only from myogestic.sources.serial_source (needs pyserial).
Reads fixed-width binary frames from a serial port.
Each frame is n_channels float32 values (little-endian). The source self-paces via serial blocking reads; timestamps are stamped on arrival with mne_lsl.lsl.local_clock().
Examples:
>>> from myogestic import Stream
>>> from myogestic.sources.serial_source import SerialSource
>>> source = SerialSource("/dev/ttyACM0", 115200, 8, 2000.0)
>>> stream = Stream("emg", source, window_ms=1000)
Read one n_channels-float32 frame, timestamped on arrival.
Returns (None, None) when the port is closed or a short read
yields fewer than n_channels values.
reconnect(target: str | None = None) -> StreamInfo
Reconnect to the same or a different serial port.