Bridges

A Bridge is a subprocess MyoGestic spawns alongside the app for heavy-data acquisition that doesn't fit the LSL pull model - typically a webcam decoder that writes frames straight to Zarr and publishes an LSL clock outlet so the rest of the app can align timestamps.

Bridges are registered via app.bridges(...), started before the GUI loop, and terminated as part of App.run()'s cleanup hook chain.

Bridge

Bridge(name: str, command: list[str])

A subprocess MyoGestic spawns alongside the app and tears down on exit.

The escape hatch for heavy-data acquisition that doesn't fit the LSL pull model - a webcam decoder that writes frames straight to Zarr, an ultrasound capture daemon, a custom script that owns its own buffer. The bridge subprocess runs whatever it wants; MyoGestic only cares that it stays alive and exits cleanly.

The bridge pattern is intentionally minimal: no IPC contract beyond "the subprocess exists, is alive, and stops on terminate". For data flowing back into the app, the subprocess publishes an LSL outlet (or writes to a Zarr file the app reads) - the same machinery every other source uses.

Registered via app.bridges(...); the app starts them after streams and tears them down on cleanup.

Parameters:

Name	Type	Description	Default
name	str	Human label, used in the bridge panel and logs.	required
command	list[str]	argv passed to subprocess.Popen. Stdout and stderr are captured to PIPE; nothing reads them by default.	required

Source code in myogestic/bridges/__init__.py

def __init__(self, name: str, command: list[str]):
    self.name = name
    self.command = command
    self.process: subprocess.Popen | None = None
    self.status = "stopped"

alive property

alive: bool

True while the subprocess is running.

start

start() -> None

Spawn the subprocess. Idempotent only if you check :attr:alive first.

Source code in myogestic/bridges/__init__.py

def start(self) -> None:
    """Spawn the subprocess. Idempotent only if you check :attr:`alive` first."""
    self.process = subprocess.Popen(
        self.command,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
    )
    self.status = "running"

stop

stop() -> None

Terminate the subprocess (SIGTERM, then SIGKILL after 5 s).

Source code in myogestic/bridges/__init__.py

def stop(self) -> None:
    """Terminate the subprocess (SIGTERM, then SIGKILL after 5 s)."""
    if self.process:
        self.process.terminate()
        try:
            self.process.wait(timeout=5)
        except subprocess.TimeoutExpired:
            self.process.kill()
    self.status = "stopped"

WebCamBridge

WebCamBridge(name: str, device: int = 0, zarr_path: str = 'session/cam.zarr')

Bases: Bridge

Bridge that runs the built-in webcam decoder subprocess.

Wraps python -m myogestic.bridges.webcam: captures frames from an OpenCV device, writes them to a Zarr array, and publishes the per-frame LSL clock so the rest of the app can align webcam time with EMG time.

Parameters:

Name	Type	Description	Default
name	str	Bridge label. The published LSL clock outlet is named "{name}_clock" (e.g. WebCamBridge("cam") publishes "cam_clock").	required
device	int	OpenCV device index. 0 is the system default camera; secondary cameras get 1, 2, ... in the order the OS enumerates them.	0
zarr_path	str	Where to write the frame array. Created if missing.	'session/cam.zarr'

Source code in myogestic/bridges/__init__.py

def __init__(self, name: str, device: int = 0, zarr_path: str = "session/cam.zarr"):
    super().__init__(
        name=name,
        command=[
            sys.executable,
            "-m",
            "myogestic.bridges.webcam",
            "--device",
            str(device),
            "--zarr",
            zarr_path,
            "--lsl-name",
            f"{name}_clock",
        ],
    )

CustomBridge

CustomBridge(name: str, script: str)

Bases: Bridge

Bridge that runs an arbitrary user Python script as a subprocess.

The unstructured escape hatch: when the heavy-data source you want doesn't fit :class:WebCamBridge and you'd rather write the decoder yourself than subclass :class:Bridge. The script runs with the same Python interpreter as the app (sys.executable); the rest is up to you (publish LSL, write Zarr, talk to a custom message bus, ...).

Parameters:

Name	Type	Description	Default
name	str	Bridge label.	required
script	str	Path to the Python script to spawn (e.g. "capture/ultrasound.py").	required

Source code in myogestic/bridges/__init__.py

def __init__(self, name: str, script: str):
    super().__init__(
        name=name,
        command=[sys.executable, script],
    )

The webcam runner

WebCamBridge invokes python -m myogestic.bridges.webcam as a subprocess. The same runner can be launched directly for testing:

uv run python -m myogestic.bridges.webcam --device 0 --zarr session/cam.zarr --lsl-name webcam_clock

Flags:

• --device N - OpenCV device index (default 0).
• --zarr PATH - where to write the Zarr array. Frames are appended one chunk per capture.
• --lsl-name NAME - LSL outlet name for the per-frame timestamp clock; the app subscribes to this to align webcam frames with EMG.