API Reference

Load and analyze Delsys CSV exports as pysampled.Data time series.

The package centers on Log, which reads a CSV exported from EMGworks or Trigno Discover, normalizes its many per-format quirks (header layouts, sub-channel orderings, link-device asynchrony), resamples each channel to a configurable per-modality target rate, and groups the result into per-sensor modality bundles (EMG, EKG, IMU, FSR, VO2Master).

Classes:

Log: Top-level loader; the only entry point most users need.

Sensor: One physical Delsys sensor with all its modality bundles. Signal: Per-channel time series tagged with sensor / modality / sub-channel.

EMG: Single, Duo, and Quattro EMG bundle with preprocessing and feature extraction. EKG: ECG bundle with R-peak detection and rate properties. IMU: Accelerometer / gyroscope bundle (X, Y, Z axes). FSR: Force-sensitive-resistor bundle (4 channels). VO2Master: Respiratory gas analyzer link device (8 channels).

SensorInfo, SensorLog, SigInfoDelsys: Metadata namedtuples.

Constants:

TARGET_SR: Default per-modality target sampling rate. SUBCHANNEL_MAP: Canonical sub-channel ordering per modality. APPLICATIONS: Recognized acquisition applications. LINK_DEVICE_REGISTRY: Sensor-name substring → (modality, synthetic

sensor number) for link devices (VO2 Master, HR Strap).

Example

import delsys

lf = delsys.Log("path/to/Trial.csv")
for emg in lf.emg.split_by_signal_name():
    envelope = emg.process(amp_kind="envelope2")
right_forearm = lf.find(side="R", location="Forearm")

Top-level loader

Top-level loader.

The Log class reads a Delsys CSV export, dispatches to the appropriate per-format parser (EMGworks, Trigno Discover basic, Trigno Discover with link devices), and groups the parsed signals into Sensor objects keyed by sensor number.

class delsys.log.Log(fname, sensor_map=None, target_sr=None, clock_mul=1.0, t0=0.0, sensor_name_replace=None)

Load a Delsys CSV file and expose its signals as typed bundles.

A Log is the single entry point of the package. Construct one with the path to a CSV file produced by EMGworks or Trigno Discover, and the constructor parses the header, picks the right per-format dataframe parser, resamples each channel to the target sampling rate, and assembles the results into per-sensor modality bundles (EMG, EKG, IMU, FSR, VO2Master, etc.).

After construction, three families of accessors are available:

Direct attribute access: lf.emg, lf.ekg, lf.acc, lf.gyro, lf.fsr, lf.analog, lf.vo2master, lf.hrstrap — each returns a single aggregated pysampled.Data per modality (channels stacked across all sensors that have that modality), or None if no sensor does. Use bundle.split_by_signal_name() to recover the per-Sensor list. Side accessors lf.left / lf.right / lf.center return lists of Sensor.
The find() method for filtered queries.
__getitem__ for legacy lookups by sensor number, side, modality, location, or sensor name. Deprecated as of 0.3.0 — retained indefinitely for backward compatibility, but new code should prefer find, which is more explicit about its return shape.

Parameters:

fname (str) – Path to a CSV file exported from Delsys.
sensor_map (str | List[SensorLog] | None) –
Manual sensor-to-name-and-placement map. May be:
- None (default) — auto-build from the sensor numbers found in the file.
- str — path to a channelmap text file (e.g. delsys_channelmap.txt).
- list — a pre-parsed list of SensorLog records.
target_sr (Dict[str, float | None] | None) – Per-modality target sampling rates, e.g. {"EMGS": 2000, "ACC": 200, ...}. Defaults to delsys.TARGET_SR.
clock_mul (float) – Sampling-rate multiplier used when re-clocking against another acquisition system. For example, if a Delsys recording’s duration is 190.27 s but the synchronized OptiTrack recording is 190.12 s, set clock_mul = 190.27 / 190.12 so the inferred sampling rates are scaled to the OptiTrack clock.
t0 (float) – Trial start time in the synchronized clock. If non-zero and clock_mul != 1.0, t0 is interpreted in the new clock.
sensor_name_replace (Dict[str, str] | None) – {corrupted_name: new_name} map applied to sensor names that were misspelled during data acquisition.

Variables:

fname (str) – The CSV path that was loaded.
name (str) – File stem (without extension), used as the recording’s name.
hdr (dict) – Parsed header information (see _parse_hdr()).
sensors (List[Sensor]) – One entry per physical sensor in the recording.
signals (List[Signal]) – One entry per per-channel signal (i.e. an EMG sensor contributes one Signal, an ACC sensor contributes three).
sensor_map (List[SensorLog]) – The channelmap used during loading.
signal_map (List[SigInfoDelsys]) – Per-column metadata records.
target_sr (Dict[str, Optional[float]]) – Effective target-sr map.
clock_mul (float) – Effective clock multiplier.
t0 (float) – Effective trial start time.
sensor_groups (Dict[str, Sequence[int]]) – User-defined groupings (see add_sensor_group()).
t_min (float) – Earliest valid time across all signals (seconds).
t_max (float) – Latest valid time across all signals (seconds).
sr_orig (List[float]) – Original (pre-resample) sampling rate per channel.

Example

import delsys

lf = delsys.Log("trial_01.csv")
for emg in lf.emg.split_by_signal_name():
    processed = emg.process(amp_kind="envelope2")
right_sensors = lf.find(side="R")
forearm_emg = lf.find(modality="EMG", location="Forearm")

property dur: float: Duration of the recording, in seconds (t_max - t_min).

property modality_sensors: Dict[str, Set[str]]: Map each modality to the set of sensor names that have it.

property sensor_numbers: List[int]: Sensor numbers in their order of appearance in sensors.

find(modality=None, side=None, location=None, sensor_number=None, name=None, as_='auto')

Query for sensors, modality bundles, or raw signals.

Parameters:

modality (str | None) – Modality to filter by. Case-insensitive and matches EMG variants ('emg' matches sensors with EMGS/EMGD/ EMGQ). Recognized values: 'EMG', 'EKG', 'ACC', 'GYRO', 'FSR', 'Analog', 'VO2', 'HR'.
side (str | None) – Sensor side — one of 'L', 'R', 'C'.
location (str | None) – Substring match on sensor.location (e.g. 'Forearm' matches both 'LForearm' and 'RForearm').
sensor_number (int | None) – Specific sensor number.
name (str | None) – Substring match on sensor.name.
as – Return shape — one of 'auto', 'modality', 'sensor', 'signal'. 'auto' returns modality bundles when modality is set, otherwise sensors. 'signal' returns raw Signal objects (one per sub-channel).
as_ (str) –

Returns:

A list of matching items (possibly empty).

Raises:

ValueError – If as_='modality' is requested without a modality argument, or if as_ is unrecognized.

Return type:

list

Example

lf.find(modality="EMG")                  # list of EMG bundles
lf.find(modality="emg")                  # case-insensitive
lf.find(modality="EMGS")                 # variant matching
lf.find(modality="VO2")                  # link devices
lf.find(side="R")                        # right-side sensors
lf.find(location="Forearm")              # substring match
lf.find(sensor_number=5)                 # by number
lf.find(name="Avanti sensor 2")          # substring match
lf.find(modality="EMG", as_="signal")    # raw Signal objects
lf.find(modality="EMG", as_="sensor")    # whole Sensor objects
lf.find()                                # all sensors

clean_emg_ekg_artifact(*, config=None, motion='auto', in_place=True, generate_report=True, splice_source='combined')

Clean ECG and motion artifact from every EMG channel in this Log.

Stages:

Gather — emg (with pysampled.Data.shift_baseline() applied so the per-channel dB metrics in the report reflect the cleaning’s effect on the AC signal rather than a constant offset in the raw input) + ekg + (optional) per-EMG ACC predictors resolved from motion.
Harmonize — defensively resample EMG / EKG / ACC to the EMG bundle’s sampling rate. EMG passes through unchanged; the rest are tail-trimmed to a common length.
Pipeline — preprocess → ICA-based ECG suppression with auto component detection by lagged correlation → optional ACC-guided motion regression with safety gates. See delsys.cleaning.run_pipeline().
Splice-back — when in_place=True (default), replace the EMG samples in signals and rebuild every sensor that carries EMG so emg and sensors[*].emg reflect the cleaned data on next access. Note that the baseline-shift in step 1 is part of what gets spliced back, so any DC offset in the raw EMG is permanently removed once clean_emg_ekg_artifact() runs in place.

Parameters:

config (CleaningConfig | None) – Pipeline knobs. Defaults to CleaningConfig’s defaults.
motion (str | Dict[int, int | str] | None) –
ACC predictor source.
- "auto" (default) — pair each EMG sensor with its own ACC bundle when present (Trigno Avanti sensors carry both modalities). EMG sensors with no ACC pair pass through the ECG stage only.
- dict of {emg_sensor_number: target} — explicit mapping. target is either an integer sensor number whose ACC bundle to use, or a string matched against Sensor.location.
- None — skip the motion stage regardless of config.use_motion_stage.
in_place (bool) – If True (default), mutate signals and rebuild every EMG sensor; the returned CleaningResult is for diagnostics. If False, do not mutate; just return the result so the caller can splice manually.
generate_report (bool) – If True (default), write a multi-page PDF report to <source_csv_stem>_cleaning_report.pdf next to the input CSV after the run completes (and after the in-place splice-back, when applicable). Equivalent to calling CleaningResult.generate_report() on the returned result. Pass False to skip the PDF step.
splice_source (str) – Which cleaned variant to splice back into signals when in_place=True. One of "combined" (default — CleaningResult.cleaned_emg, preprocess + ECG + motion), "ekgonly" (CleaningResult.cleaned_emg_ekgonly, preprocess + ECG only), or "motiononly" (CleaningResult.cleaned_emg_motiononly, preprocess + motion only). Use splice_source="ekgonly" when the motion stage is doing more harm than good on a particular trial, or "motiononly" to keep just the motion regression. Ignored when in_place=False. The auto-report runs after the splice-back, so the per-channel pages reflect what lf.emg will look like.

Returns:

CleaningResult containing the cleaned EMG matrix, per-stage snapshots, and diagnostics.

Raises:

ValueError – If the Log has no EMG bundle, or if a motion dict references a sensor number that does not exist or has no ACC modality.

Return type:

CleaningResult

Example

import delsys
from delsys import CleaningConfig

lf = delsys.Log("trial.csv")

# Default: auto ECG component removal, auto ACC pairing
result = lf.clean_emg_ekg_artifact()

# Manual ECG component override
cfg = CleaningConfig(
    ecg_auto_remove_components=False,
    ecg_components_to_remove=[2, 5],
)
result = lf.clean_emg_ekg_artifact(config=cfg)

# Drive motion with explicit ACC pairing
result = lf.clean_emg_ekg_artifact(motion={4: 4, 14: 11})

# Dry-run, do not mutate
result = lf.clean_emg_ekg_artifact(in_place=False)

__getitem__(key)

Look up sensors or modality bundles by sensor number, side, modality, location, or name.

Deprecated since version 0.3.0: Use find() instead. __getitem__ overloads five key types (int sensor number, single-letter side, modality string, location substring, sensor-name substring) and collapses single-match results, which makes the return shape hard to predict at the call site. find takes named filters and always returns a list. The legacy method is retained indefinitely for backward compatibility — there is no removal plan — but new code should prefer find.

When key is a list, the result is the OR of the per-key matches.

Parameters:

key (int | str | list) –

One of:

int — sensor number.
str of length 1 matching L/R/C — side.
modality string ('EMG', 'ACC', 'analog', …).
location substring ('Foot' matches 'LFoot' and 'RFoot').
sensor-name substring.
list of any of the above.

Returns:

The single matching item if exactly one match was found, else a list of all matches (possibly empty).

Return type:

Any

add_sensor_group(name, sensor_list)

Register a named group of sensor numbers.

Parameters:

name (str) – Group label (e.g. 'LBFL').
sensor_list (Sequence[int]) – Sensor numbers to include. All numbers must exist in sensor_numbers.

Raises:

AssertionError – If sensor_list references an unknown sensor number.

Return type:

None

Example

lf.add_sensor_group("LBFL", (14, 7, 4))

is_resampled()

True if a non-unity clock_mul was applied at load time.

Return type:: bool

is_shifted()

True if a non-zero t0 was applied at load time.

Return type:: bool

is_adjusted()

True if either is_resampled() or is_shifted() is true.

Return type:: bool

export_to_csv(modality='emg', process_func=None, process_name='_processed', export_dir=None)

Write all signals of one modality to a CSV file.

Parameters:

modality (str) – Modality attribute name ('emg', 'acc', …).
process_func (Callable | None) – Optional per-signal processing callable applied before writing. Default is the identity.
process_name (str) – Suffix appended to the output filename when a processing function was supplied. Ignored when process_func is None.
export_dir (str | None) – Output directory. Defaults to <input-csv-parent>/export, created if missing.

Raises:

AssertionError – If signals selected by modality have heterogeneous sampling rates.

Return type:

None

Sensors

Sensor — one physical Delsys sensor with all its modality bundles.

A Sensor is built from a SensorInfo metadata record and the list of Signal objects produced by the loader for that sensor number. Its __init__() groups signals by modality, stacks per-channel arrays in canonical sub-channel order, and constructs the appropriate modality bundle (EMG, EKG, IMU, FSR, VO2Master, Analog, or HRStrap) for each.

delsys.sensor.MODALITY_REGISTRY: Dict[str, Type[Data]] = {'ACC': <class 'delsys.signals.IMU'>, 'Analog': <class 'pysampled.core.Data'>, 'EKG': <class 'delsys.ekg.EKG'>, 'EMGD': <class 'delsys.emg.EMG'>, 'EMGQ': <class 'delsys.emg.EMG'>, 'EMGS': <class 'delsys.emg.EMG'>, 'FSR': <class 'delsys.signals.FSR'>, 'GYRO': <class 'delsys.signals.IMU'>, 'HR': <class 'pysampled.core.Data'>, 'VO2': <class 'delsys.signals.VO2Master'>}: Modality tag → bundle class. Drives the dispatch in Sensor’s constructor: each parsed modality is wrapped in the corresponding pysampled.Data subclass (or plain pysampled.Data for Analog and HR, which don’t have a sensor-aware bundle class). Adding a new modality is a one-line edit here plus a delsys._constants.SUBCHANNEL_MAP entry.

class delsys.sensor.Sensor(sensor_info, signal_list)

All signals from one physical Delsys sensor, exposed as typed bundles.

The constructor walks the signal_list, groups signals by modality, and attaches each modality bundle as an instance attribute named per _mod_to_attr() (emg, ekg, acc, gyro, fsr, analog, vo2master, hrstrap). The metadata fields from sensor_info (name, number, type_sensorlog, lrc, location, modalities) are also copied onto the instance for direct attribute access.

Parameters:

sensor_info (SensorInfo) – Combined sensor metadata produced by the loader.
signal_list (List[Signal]) – All Signal objects belonging to this sensor. Every entry’s signal.sensor must equal sensor_info.

Variables:

name (str) – Human-readable sensor name (e.g. "EMG 01 04498").
number (int) – Delsys sensor number.
type_sensorlog (Optional[str]) – Sensor type from the channelmap (e.g. "EMG", "FSR"). None if no channelmap was used.
lrc (Optional[str]) – Sensor side — 'L', 'R', 'C' or None.
location (Optional[str]) – Body-location label from the channelmap.
modalities (set[str]) – All modality tags carried by this sensor’s signals.
emg (EMG) – Present iff this sensor has an EMG modality.
ekg (EKG) – Present iff this sensor has an EKG modality.
acc (IMU) – Present iff this sensor has ACC.
gyro (IMU) – Present iff this sensor has GYRO.
fsr (FSR) – Present iff this sensor has FSR.
analog (pysampled.Data) – Present iff this sensor has Analog. Plain pysampled.Data (not a sensor-aware bundle).
vo2master (VO2Master) – Present iff this sensor is a VO2 Master link device.
hrstrap (pysampled.Data) – Present iff this sensor is an HR Strap link device. Plain pysampled.Data (not a sensor-aware bundle).

Raises:

AssertionError – If sensor_info is not a SensorInfo, or if any element of signal_list is not a Signal, or if a signal’s .sensor does not match sensor_info, or if channels of one modality have inconsistent sample counts or t0.

Note

Construction via Log runs every signal through delsys._util._normalize_signal_lengths() first, so the same-modality length assert is a safety net rather than a tripwire on that path. Direct callers building a Sensor from un-normalized signals are responsible for length consistency.

property is_link: bool

True if this sensor is a Delsys link device.

A link device is any sensor whose modality appears in delsys._constants.LINK_DEVICE_REGISTRY — currently the VO2 Master and HR Strap. Use this in preference to comparing Sensor.number against magic constants: the underlying synthetic numbers are an internal detail of how the parser labels link channels, not a stable identifier.

get_signal()

Return the first non-IMU bundle attached to this sensor.

Useful when EMG/EKG and Analog/FSR get mis-typed and you don’t know which attribute to reach for. Lookup priority is: emg, ekg, analog, fsr, vo2master, hrstrap. Returns None if the sensor only has IMU data (or no data at all).

Return type:: EMG | EKG | Data | FSR | VO2Master | None

Signal classes

Signal classes.

Defines the per-channel Signal class and the modality-bundled classes IMU (3-axis accelerometer/gyroscope), FSR (4-channel force-sensitive resistor), and VO2Master (8-channel respiratory gas analyzer). Each is a thin extension of pysampled.Data that adds modality-specific property accessors plus two tiny shared conveniences (sensor and shape).

Sensor metadata is carried through pysampled.Data.meta under the key 'sensor' (and, for Signal, also 'modality' and 'subchannel'), so it survives every clone, slice, filter, or resample operation pysampled performs.

The metadata namedtuples (SensorLog and SensorInfo) live in delsys._metadata.

class delsys.signals.Signal(sig, sr, axis=None, history=None, t0=0.0, meta=None, signal_names=None, signal_coords=None)

One per-channel time series tagged with sensor + modality + sub-channel.

A Signal is the basic unit produced by the per-format dataframe parsers: one row in Log.signals corresponds to one column of sample data after resampling. Three identifying fields — sensor, modality, and subchannel — live in pysampled.Data.meta so they survive cloning operations (filtering, resampling, slicing) automatically.

Construct with meta populated:

Signal(sig, sr, t0=t0, meta={
    "sensor": sensor_info,
    "modality": "EMGS",
    "subchannel": "A",
})

Variables:

sensor (SensorInfo) – The source sensor’s metadata record.
modality (str) – Normalized modality tag.
subchannel (str) – Sub-channel label.
sensor_name (str) – The modality-as-attribute-name (e.g. 'emg', 'vo2master'); kept for backward compatibility.
shape (tuple) – Shape of the underlying sample array.

Parameters:

sig (ndarray) –
sr (float) –
axis (int | None) –
history (List[Tuple[str, Any | None]] | None) –
t0 (float) –
meta (dict | None) –
signal_names (List[str] | None) –
signal_coords (List[str] | None) –

property sensor: SensorInfo | None: The SensorInfo record, or None if not set.

property sensors: List[SensorInfo]

All SensorInfo records this signal carries.

See _bundle_sensors().

property modality: str: Normalized modality tag (one of 'EMGS', 'EMGD', 'EMGQ', 'EKG', 'ACC', 'GYRO', 'FSR', 'Analog', 'VO2', 'HR').

property subchannel: str: Sub-channel label — 'A' for single-channel modalities, 'X'/'Y'/'Z' for IMU axes, 'A'/'B'/'C'/'D' for multi-channel EMG and FSR.

property sensor_name: str

The modality-as-attribute-name (e.g. 'emg', 'vo2master').

Historical name; kept for backward compatibility. This is the modality translated into the attribute name used on the parent Sensor, not the sensor’s display name.

property shape: tuple: Shape of the underlying sample array (shortcut for self._sig.shape).

class delsys.signals.IMU(sig, sr, axis=None, history=None, t0=0.0, meta=None, signal_names=None, signal_coords=None)

Tri-axial accelerometer or gyroscope bundle.

Holds the three axes of a Delsys IMU as a single 2-D array of shape (n_samples, 3) with columns X, Y, Z in that order. Each axis is exposed as a property returning a single-axis IMU.

Example

import delsys

lf = delsys.Log("trial_01.csv")
for sensor in lf.find(modality="ACC", as_="sensor"):
    acc = sensor.acc
    x_axis = acc.x         # IMU containing only the X column
    magnitude = (acc.x() ** 2 + acc.y() ** 2 + acc.z() ** 2) ** 0.5

Parameters:

sig (ndarray) –
sr (float) –
axis (int | None) –
history (List[Tuple[str, Any | None]] | None) –
t0 (float) –
meta (dict | None) –
signal_names (List[str] | None) –
signal_coords (List[str] | None) –

property sensor: SensorInfo | None

The SensorInfo record, or None if not set.

For aggregate views (Log.acc / Log.gyro spanning multiple sensors) this returns None; use sensors (plural) to get the per-channel list.

property sensors: List[SensorInfo]

All SensorInfo records this bundle carries.

See _bundle_sensors().

property shape: tuple: Shape of the underlying sample array.

class delsys.signals.FSR(sig, sr, axis=None, history=None, t0=0.0, meta=None, signal_names=None, signal_coords=None)

Force-sensitive-resistor bundle (four channels).

Stores the four FSR channels as a single 2-D array of shape (n_samples, 4) with columns A, B, C, D in that order. Each channel is exposed as a property returning a single-channel FSR.

Example

import delsys

lf = delsys.Log("trial_01.csv")
for sensor in lf.find(modality="FSR", as_="sensor"):
    heel = sensor.fsr.a
    ball = sensor.fsr.b

Parameters:

sig (ndarray) –
sr (float) –
axis (int | None) –
history (List[Tuple[str, Any | None]] | None) –
t0 (float) –
meta (dict | None) –
signal_names (List[str] | None) –
signal_coords (List[str] | None) –

property sensor: SensorInfo | None

The SensorInfo record, or None if not set.

None on aggregate views — use sensors (plural) to get the per-channel list.

property sensors: List[SensorInfo]

All SensorInfo records this bundle carries.

See _bundle_sensors().

property shape: tuple: Shape of the underlying sample array.

class delsys.signals.VO2Master(sig, sr, axis=None, history=None, t0=0.0, meta=None, signal_names=None, signal_coords=None)

VO2 Master link-device bundle (eight channels in fixed order).

The BreathingCycle channel is dropped at parse time, so it does not appear here. The remaining eight columns are exposed under both short and verbose property names so notebook code can use whichever reads better:

Idx	Short name	Verbose alias
0	`rr`	`respiration_rate`
1	`td`	`tidal_vol`
2	`vent`	`ventilation`
3	`Feo2`	(none)
4	`vo2`	`VO2_absolute`
5	`ap`	`ambient_pressure`
6	`fl`	`flow_sensor`
7	`o2_hum`	`oxygen_sensor_humidity`

Example

import delsys

lf = delsys.Log("trial_01.csv")
vo2master, = lf.vo2master
mean_breathing_rate = vo2master.rr().mean()
tidal_volume_ml = vo2master.tidal_vol() * 1000

Parameters:

sig (ndarray) –
sr (float) –
axis (int | None) –
history (List[Tuple[str, Any | None]] | None) –
t0 (float) –
meta (dict | None) –
signal_names (List[str] | None) –
signal_coords (List[str] | None) –

property sensor: SensorInfo | None

The SensorInfo record, or None if not set.

None on aggregate views — use sensors (plural) to get the per-channel list.

property sensors: List[SensorInfo]

All SensorInfo records this bundle carries.

See _bundle_sensors().

property shape: tuple: Shape of the underlying sample array.

EMG

EMG-specific signal class and feature extraction.

Defines EMG, a pysampled.Data extension that adds the canonical EMG preprocessing pipeline (bandpass → rectify → amplitude → optional lowpass), the Teager–Kaiser energy operator, and two feature extractors (a NeuroKit2 wrapper and a hand-rolled temporal/frequency feature dict).

class delsys.emg.EMG(sig, sr, axis=None, history=None, t0=0.0, meta=None, signal_names=None, signal_coords=None)

EMG bundle: pysampled.Data plus EMG preprocessing and features.

Holds an EMG signal as either a 1-D (n_samples,) array (single channel) or a 2-D (n_samples, n_channels) array (Quattro / Duo). The SensorInfo record for the source sensor lives in self.meta['sensor'] and survives clone/filter/resample operations automatically; the convenience sensor property reads it back.

Example

import delsys

lf = delsys.Log("trial_01.csv")
for emg in lf.emg:
    envelope = emg.process(amp_kind="envelope2")
    features = emg.get_features(kind="temp", win_size=0.25, win_inc=0.1)

Parameters:

sig (ndarray) –
sr (float) –
axis (int | None) –
history (List[Tuple[str, Any | None]] | None) –
t0 (float) –
meta (dict | None) –
signal_names (List[str] | None) –
signal_coords (List[str] | None) –

property sensor: SensorInfo | None

The SensorInfo record, or None if not set.

None on aggregate views — use sensors (plural) to get the per-channel list.

property sensors: List[SensorInfo]

All SensorInfo records this bundle carries.

See delsys.signals._bundle_sensors().

property shape: tuple: Shape of the underlying sample array.

process_nk()

Process the EMG with NeuroKit2’s nk.emg_process().

Returns:: Dict of per-sample NeuroKit signal traces (clean signal, amplitude, activations, …). Wrap in pd.DataFrame(...) if you want a DataFrame.
Return type:: Dict[str, Any]

tkeo()

Apply the Teager–Kaiser Energy Operator.

TKEO improves onset detection by emphasizing transients in both amplitude and frequency. The output preserves shape, sampling rate, and meta (including sensor) via pysampled.Data._clone().

Returns:: A new EMG of the same shape and sampling rate.
Return type:: EMG

process(amp_kind='envelope2', lowpass=None, **kwargs)

Run the canonical EMG preprocessing pipeline.

Steps: shift_baseline → bandpass 20–500 Hz → abs → amplitude extraction (per amp_kind) → optional final lowpass.

Parameters:

amp_kind (str) – Amplitude extractor — one of 'rms', 'mean', 'envelope', 'envelope2', or 'nk'.
lowpass (float | int | None) – Optional cutoff (Hz) for a final lowpass after the amplitude step. None skips it.
**kwargs (Any) – Forwarded to the amplitude extractor. order (int) sets the bandpass/lowpass filter order (default 4). win_size and win_inc (seconds) configure the window for 'rms' and 'mean'.

Returns:

A new EMG holding the processed signal at the original sampling rate.

Raises:

ValueError – If amp_kind is not one of the supported values.

Return type:

EMG

rms(bandpass_low=20.0, bandpass_high=500.0, power_line_frequency=60.0, win_size=0.05, envelope_sr=240.0)

RMS amplitude envelope on a clean filter chain.

Pipeline: shift_baseline → highpass → lowpass → notch (power line) → running RMS over win_size. The window step is set to 1 / envelope_sr so the output sampling rate is exactly envelope_sr.

Notch-filtering the power line interference makes this preferable to process(amp_kind='rms') when working in line-frequency-noisy environments.

Returns a plain pysampled.Data rather than EMG, because the amplitude envelope isn’t an EMG signal anymore (different sampling rate, different units). The history of the filter chain and self.meta (including sensor) are propagated onto the result.

Parameters:

bandpass_low (float) – Highpass cutoff in Hz. Default 20.
bandpass_high (float) – Lowpass cutoff in Hz. Default 500 (drop to 450 if hardware bandwidth is the dominant constraint).
power_line_frequency (float) – Notch frequency in Hz. Default 60.
win_size (float) – RMS window length in seconds. Default 0.05.
envelope_sr (float) – Output sampling rate in Hz. Default 240.

Returns:

A pysampled.Data holding the RMS amplitude envelope, sampled at envelope_sr. meta carries the source sensor and _history reflects the full filter + RMS chain.

Return type:

Data

Example

import delsys

lf = delsys.Log("trial_01.csv")
envelope = lf.emg[0].rms(envelope_sr=240)
# Override defaults for hardware with narrower bandwidth:
envelope = lf.emg[0].rms(bandpass_high=450)

get_features_nk(method='interval')

Compute EMG features via NeuroKit2’s nk.emg_analyze().

Parameters:: method (str) – NeuroKit analysis method ('interval', 'event-related').
Returns:: Dict of per-feature values. Wrap in pd.DataFrame(...) if you want a DataFrame.
Return type:: Dict[str, Any]

get_features(kind='all', win_size=0.25, win_inc=0.1)

Hand-rolled temporal and/or frequency features over a sliding window.

Parameters:

kind (str) – Which feature family to compute — 'all' (temporal + frequency), 'temp', or 'freq'.
win_size (float) – Window length in seconds.
win_inc (float) – Window step in seconds.

Returns:

Dict keyed by feature name. The 'time' entry holds the per-window timestamps. Wrap in pd.DataFrame(...) if you want a DataFrame.

Raises:

ValueError – If kind is not one of 'all', 'temp', or 'freq'.

Return type:

Dict[str, Any]

EKG

EKG-specific signal class and analysis utilities.

Defines EKG, a pysampled.Data extension that adds:

R-peak detection (highpass + HeartPy + heuristic prune for spurious double peaks).
Aggregate heart-rate and respiration-rate properties.
Manual annotation slots in self.meta (added/removed/noisy peaks, flip flag, free-text tags).
Per-segment feature extraction via HeartPy.

class delsys.ekg.EKG(*args, **kwargs)

ECG bundle with R-peak detection, HRV-relevant metadata, and rate properties.

R-peak indices and manual annotations live in pysampled.Data.meta under fixed keys initialized by _initialize_meta_for_rpeaks():

meta key	Contents
`rpeaks_idx_default`	Auto-detected R-peak sample indices.
`rpeaks_idx_added`	User-added peaks merged into the result.
`rpeaks_idx_removed`	User-removed peaks subtracted from the result.
`noisy_segments_idx`	List of `(start_idx, end_idx)` pairs marking unusable spans; peaks inside them are dropped by `rpeak_times()`.
`is_flipped`	Whether to negate before peak detection.
`tags`	Free-text tags (`'reviewed'`, `'representative'`, `'interesting'`).

Note

Slicing currently does not reindex these cached R-peak entries; a sliced EKG still references peaks at the parent’s sample positions. See TODO.md for the planned __getitem__ fix.

Example

import delsys

lf = delsys.Log("trial_01.csv")
ekg = lf.ekg[0]
peaks = ekg.find_rpeaks()       # alias for find_rpeaks_pn
t_peaks = ekg.t[peaks]
t_window, bpm = ekg.ihr()       # instantaneous HR over time

Parameters:

args (Any) –
kwargs (Any) –

property sensor: SensorInfo | None

The SensorInfo record, or None if not set.

None on aggregate views — use sensors (plural) to get the per-channel list.

property sensors: List[SensorInfo]

All SensorInfo records this bundle carries.

See delsys.signals._bundle_sensors().

property shape: tuple: Shape of the underlying sample array.

property hr: float: Mean heart rate (bpm) over the full signal, via HeartPy.

property rr: float: Mean respiration rate (breaths per minute) over the full signal, via HeartPy.

process_nk(method='neurokit')

Process the ECG with NeuroKit2’s nk.ecg_process().

Parameters:: method (str) – Cleaning method passed to NeuroKit ('neurokit', 'biosppy', 'pantompkins1985', 'hamilton2002', 'elgendi2010', 'engzeemod2012').
Returns:: Dict of per-sample NeuroKit signal traces (ECG_Clean, ECG_R_Peaks, ECG_Rate, …). Wrap in pd.DataFrame(...) if you want a DataFrame.
Return type:: Dict[str, Any]

get_features_hp(win_size=5.0, win_inc=0.1, **kwargs)

Segment-wise ECG features via HeartPy’s hp.process_segmentwise().

Parameters:

win_size (float) – Segment width in seconds.
win_inc (float) – Segment step in seconds.
**kwargs (Any) – Forwarded to hp.process_segmentwise().

Returns:

Dict keyed by HeartPy metric name (e.g. 'bpm', 'sdnn', 'rmssd'). The added 'time' entry holds (t_start, t_end) pairs for each segment. Wrap in pd.DataFrame(...) if you want a DataFrame.

Return type:

Dict[str, Any]

find_rpeaks_pn(highpass=5.0, hr_max=200.0)

Detect R-peaks: highpass-prefilter, then HeartPy, then prune double peaks.

HeartPy occasionally reports two peaks for a single QRS complex when the signal is noisy. This method post-processes its output by comparing the inter-peak interval to hr_max and dropping the smaller-amplitude peak from each pair that implies an implausibly high heart rate.

Parameters:

highpass (float) – Highpass cutoff (Hz) applied before HeartPy. None to skip the prefilter. Default 5.
hr_max (float) – Maximum plausible heart rate (bpm). Pairs whose inferred inter-peak rate exceeds this are pruned. Default 200.

Returns:

Sorted list of R-peak sample indices, with manual additions merged in and manual removals subtracted.

Return type:

List[int]

Side effects:: Edits meta in place — sets rpeaks_idx_default and rpeaks_idx_removed. Manual rpeaks_idx_added entries are preserved.

Raises:

NotImplementedError – If the bundle holds more than one channel (i.e. an aggregate EKG). HeartPy expects a 1D series; use ekg.split_by_signal_name()[i] to pick one channel first.

Parameters:

highpass (float) –
hr_max (float) –

Return type:

List[int]

find_rpeaks(highpass=5.0, hr_max=200.0)

Canonical alias for find_rpeaks_pn().

Parameters:

highpass (float) –
hr_max (float) –

Return type:

List[int]

rpeak_times()

Return R-peak sample indices that fall outside noisy segments.

Calls find_rpeaks() first if the meta cache is empty.

Returns:: (idx, pk_idx) where idx is the array of clean R-peak sample indices, and pk_idx is the index of each surviving peak within the original (pre-filtered) ordered peak list. The second array is what ihr() uses to detect peaks straddling a noisy segment.
Return type:: Tuple[ndarray, ndarray]

ihr()

Instantaneous heart rate.

Returns:: (times, bpm) — the timestamps mid-way between consecutive clean R-peaks and the corresponding instantaneous HR. np.nan is inserted across noisy-segment gaps so that consumers can distinguish missing data from continuous low values.
Return type:: Tuple[List[float], List[float]]

flip_signal()

Toggle the is_flipped flag and re-detect R-peaks.

Useful when an ECG channel was recorded with reversed polarity: find_rpeaks_pn() will negate the signal before detection.

Return type:: None

Cleaning

Multi-stage EMG cleaning pipeline.

This module ports the staged cleaner from pn-projects/projects/emg_ica_cleaning.py into the package. Three stages, all optional via CleaningConfig:

Preprocess — pysampled.Data.interpnan() plus an optional high/low-pass.
ECG suppression — fit FastICA on the EMG matrix concatenated with the EKG reference, score components by lagged correlation against the EKG, drop the worst offenders, then ridge-regress the EKG residual out of every channel.
Motion suppression — per-channel ridge regression of lagged ACC predictors with safety gates that reject regressions which would shrink the signal below a variance / power threshold.

The high-level entry point is Log.clean_emg_ekg_artifact (in delsys.log); run_pipeline() is the lower-level numpy runner the method wraps. Everything else is exposed as a building block so power users can drive the pipeline component-by-component.

Realtime / overlap-add and matplotlib helpers from the source are intentionally not ported. The realtime variant was a chunked offline run rather than true streaming; ship it back if a real streaming use case appears.

class delsys.cleaning.ICAResult(model, sources, mixing, feature_names=None)

Container for fitted ICA state and derived arrays.

Variables:

model (sklearn.decomposition._fastica.FastICA) – The fitted sklearn.decomposition.FastICA instance.
sources (numpy.ndarray) – Source time courses, shape (n_samples, n_components).
mixing (numpy.ndarray) – Mixing matrix, shape (n_signals, n_components).
feature_names (List[str] | None) – Optional per-input-channel labels (length n_signals).

Parameters:

model (FastICA) –
sources (ndarray) –
mixing (ndarray) –
feature_names (List[str] | None) –

class delsys.cleaning.CleaningConfig(preprocess_highpass_hz=20.0, preprocess_lowpass_hz=None, preprocess_order=None, use_ecg_stage=True, ecg_n_components=None, ecg_components_to_remove=None, ecg_auto_remove_components=True, ecg_max_auto_components=1, ecg_corr_threshold=0.25, ecg_corr_max_lag_samples=10, ecg_use_regression=True, ecg_reg_max_lag_samples=10, ecg_reg_ridge_alpha=1e-06, ica_random_state=0, use_motion_stage=True, motion_max_lag_samples=10, motion_ridge_alpha=0.001, motion_include_magnitude=True, motion_include_derivative=False, min_variance_ratio=0.1, min_power_ratio=0.1)

All knobs for Log.clean_emg_ekg_artifact.

Defaults reproduce the source pipeline’s behavior. Stage gates (use_ecg_stage / use_motion_stage) skip the entire stage when False; per-stage knobs only apply when the stage runs.

Preprocess:

preprocess_highpass_hz: High-pass cutoff before ECG / motion: stages. None to skip. Default 20 Hz.
preprocess_lowpass_hz: Optional low-pass cutoff applied after: the high-pass. None to skip.

preprocess_order: Filter order; None lets pysampled pick.

ECG (ICA) stage:

use_ecg_stage: Master switch. If False, skip ICA / EKG: regression entirely.
ecg_n_components: n_components passed to FastICA. None: uses n_signals (one per input channel including EKG).
ecg_components_to_remove: Manual override — explicit list of IC: indices to zero. When None, components are picked by auto-detection (when ecg_auto_remove_components=True).
ecg_auto_remove_components: Pick ECG-like components by lagged: correlation against the EKG reference.
ecg_max_auto_components: Cap on how many ICs the auto-detector: removes. Default 1 (the strongest EKG-correlated IC).
ecg_corr_threshold: Minimum absolute lagged correlation for an: IC to be flagged.
ecg_corr_max_lag_samples: Lag window (± samples) searched when: scoring IC-vs-EKG correlation.
ecg_use_regression: Run a second pass that ridge-regresses the: lagged EKG out of every channel after IC removal. Cleans up residual artifact the IC step misses.

ecg_reg_max_lag_samples: Lag window for the regression pass. ecg_reg_ridge_alpha: Ridge regularization strength for the

regression pass.

ica_random_state: Seed passed to FastICA.

Motion (ACC) stage:

use_motion_stage: Master switch. If False, skip the motion: regression step.
motion_max_lag_samples: Lag window for the per-channel ACC: regression.

motion_ridge_alpha: Ridge regularization strength. motion_include_magnitude: Append the L2-norm of multi-axis ACC

to the predictor matrix.

motion_include_derivative: Append the per-feature first: difference to the predictor matrix.
min_variance_ratio: Reject the regression on a channel when the: variance of the cleaned residual drops below this fraction of the input variance. Guards against over-cleaning.

min_power_ratio: Same idea, on mean-square power.

Parameters:

preprocess_highpass_hz (float | None) –
preprocess_lowpass_hz (float | None) –
preprocess_order (int | None) –
use_ecg_stage (bool) –
ecg_n_components (int | None) –
ecg_components_to_remove (str | List[int] | None) –
ecg_auto_remove_components (bool) –
ecg_max_auto_components (int) –
ecg_corr_threshold (float) –
ecg_corr_max_lag_samples (int) –
ecg_use_regression (bool) –
ecg_reg_max_lag_samples (int) –
ecg_reg_ridge_alpha (float) –
ica_random_state (int) –
use_motion_stage (bool) –
motion_max_lag_samples (int) –
motion_ridge_alpha (float) –
motion_include_magnitude (bool) –
motion_include_derivative (bool) –
min_variance_ratio (float) –
min_power_ratio (float) –

class delsys.cleaning.CleaningResult(cleaned_emg, sr, time, stages=<factory>, diagnostics=<factory>, coefficients=<factory>, cleaned_emg_ekgonly=None, cleaned_emg_motiononly=None, feature_names=None, fname=None, ica=None, ica_input_feature_names=None)

Outputs and per-stage diagnostics from one pipeline run.

Variables:

cleaned_emg (numpy.ndarray) – Final cleaned EMG, shape (n_samples, n_emg_channels).
sr (float) – Pipeline sampling rate (Hz).
time (numpy.ndarray) – Time grid for cleaned_emg.
stages (Dict[str, numpy.ndarray]) – Per-stage intermediate arrays — keys 'raw', 'preprocessed', 'post_ecg', 'cleaned'.
diagnostics (Dict[str, Any]) – Per-stage diagnostic dicts — keys 'ecg', 'motion', plus 'harmonization' when produced by Log.clean_emg_ekg_artifact.
coefficients (Dict[str, Any]) – Fitted regression coefficients — keys 'ecg_regression_beta', 'motion_betas'.
cleaned_emg_ekgonly (numpy.ndarray | None) – Same as cleaned_emg but with the motion stage skipped (preprocess + ECG). None when the ECG stage didn’t run.
cleaned_emg_motiononly (numpy.ndarray | None) – Same as cleaned_emg but with the ECG stage skipped (preprocess + motion). None when the motion stage didn’t run.
feature_names (List[str] | None) – Per-EMG-channel labels (length n_emg_channels). Used by generate_report() and review() to title each channel.
fname (str | None) – Source CSV path. Stamped by Log.clean_emg_ekg_artifact so that generate_report() can default to a sibling of the input file.
ica (delsys.cleaning.ICAResult | None) – Full ICAResult from the ECG stage (model, sources, mixing matrix, feature names). None when the ECG stage didn’t run. Used by review_components().
ica_input_feature_names (List[str] | None) – Per-input-row labels for ica.mixing — the EMG channel names with "EKG" appended as the last entry. None when the ECG stage didn’t run. Distinct from feature_names (EMG only).

Parameters:

cleaned_emg (ndarray) –
sr (float) –
time (ndarray) –
stages (Dict[str, ndarray]) –
diagnostics (Dict[str, Any]) –
coefficients (Dict[str, Any]) –
cleaned_emg_ekgonly (ndarray | None) –
cleaned_emg_motiononly (ndarray | None) –
feature_names (List[str] | None) –
fname (str | None) –
ica (ICAResult | None) –
ica_input_feature_names (List[str] | None) –

generate_report(path=None)

Write a multi-page PDF report summarizing the cleaning result.

Page 1 is a ranked summary table (one row per EMG channel, sorted most-attenuated first). Subsequent pages plot raw vs each cleaning variant for one channel apiece, in the same ranked order.

Parameters:: path (str | Path | None) – Output PDF path. When None, defaults to <dir(self.fname)>/<stem(self.fname)>_cleaning_report.pdf; raises ValueError if self.fname is also unset.
Returns:: The pathlib.Path of the file that was written.
Return type:: Path

review(*, channels=None)

Open an interactive matplotlib viewer over the cleaned channels.

Three stacked time-domain panels per channel — raw vs ekg-only, raw vs motion-only, raw vs combined-cleaned — with arrow-key navigation and overlay toggles. See module docstring / tutorial for the full key map.

Parameters:: channels (List[int] | None) – Optional list of EMG column indices to cycle through. When None (default), every channel is shown in ranked-by-attenuation order.
Return type:: None

review_components(*, components=None)

Open an interactive viewer over the ICA components.

Four stacked panels per component — the IC time course on top and the three input signals it most contributes to (ranked by |A[i, c]|, the absolute mixing-matrix coefficient). Use to decide whether to manually add or drop a component from the auto-detected set in CleaningConfig.ecg_components_to_remove.

For unit-variance ICA sources the ranking by |A[i, c]| is proportional to corr(sources[:, c], input[:, i]) * std(input[:, i]), so it matches “input signals this IC shows up in most” up to per-input-variance scaling. Pure-correlation ranking (without the std factor) needs to be computed by hand.

Key bindings:

→ / n — next component (wrap)
← / p — previous component (wrap)
home / end — first / last
q — close

Parameters:: components (List[int] | None) – Optional list of IC indices to cycle through. When None (default), every component is shown in index order.
Raises:: ValueError – If the ECG stage did not run (no ICA result to review).
Return type:: None

delsys.cleaning.fit_ica(emg_2d, *, n_components=None, random_state=0, max_iter=1000, tol=0.0001, whiten='unit-variance', feature_names=None)

Fit FastICA on an EMG matrix shaped (time_points, signals).

Parameters:

emg_2d (ndarray) – 2-D float array, samples down rows.
n_components (int | None) – Number of components to extract. Defaults to emg_2d.shape[1].
random_state (int) – Seed for reproducibility.
max_iter (int) – FastICA iteration cap.
tol (float) – FastICA convergence tolerance.
whiten (str) – FastICA whiten parameter.
feature_names (List[str] | None) – Optional per-input-channel labels to attach to the returned ICAResult.

Returns:

ICAResult containing the fitted model and source/mix matrices.

Return type:

ICAResult

delsys.cleaning.score_components_against_ekg(sources, ekg_1d, *, max_lag_samples=10)

Score each IC by max absolute correlation against lagged EKG.

Returns:

(scores, best_lags) — both shape (n_components,). Scores are absolute correlations in [0, 1]; lags are signed sample offsets at which the maximum was found.

Parameters:

sources (ndarray) –
ekg_1d (ndarray) –
max_lag_samples (int) –

Return type:

tuple

delsys.cleaning.auto_select_ekg_components(corr_scores, *, min_corr=0.25, keep_at_least_one_when_strong=True, max_components=1)

Pick IC indices likely carrying ECG artifact.

Parameters:

corr_scores (ndarray) – Per-component scores from score_components_against_ekg().
min_corr (float) – Threshold above which a component is flagged.
keep_at_least_one_when_strong (bool) – If no IC clears min_corr but the strongest is at least 0.2, still keep that one.
max_components (int | None) – Cap on the number of components returned. None for no cap.

Returns:

Sorted-by-score (descending) list of IC indices to remove.

Return type:

List[int]

delsys.cleaning.reconstruct_without_components(ica_result, components_to_remove)

Reconstruct the input from ICA sources after zeroing selected components.

Returns:

(cleaned, src_clean) — the inverse-transformed signal (n_samples, n_signals) and the modified source matrix (n_samples, n_components) with the selected components zeroed.

Parameters:

ica_result (ICAResult) –
components_to_remove (str | List[int] | None) –

Return type:

tuple

delsys.cleaning.regress_out_ekg_from_emg(emg_2d, ekg_1d, *, max_lag_samples=10, ridge_alpha=1e-06)

Ridge-regress the lagged EKG basis from every EMG channel.

Returns:

(cleaned, beta) — the residual (n_samples, n_channels) and the fitted coefficient matrix (n_lags+1, n_channels).

Parameters:

emg_2d (ndarray) –
ekg_1d (ndarray) –
max_lag_samples (int) –
ridge_alpha (float) –

Return type:

tuple

delsys.cleaning.regress_out_motion_from_emg(emg_2d, acc_by_emg, *, max_lag_samples=10, ridge_alpha=0.001, include_magnitude=True, include_derivative=False, min_variance_ratio=0.1, min_power_ratio=0.1)

Per-channel ACC-guided ridge regression with safety gates.

Channels without an ACC predictor pass through unchanged with a diagnostic entry reason='missing_acc_predictor'. Channels where the regression would shrink the residual below min_variance_ratio or min_power_ratio of the input keep the input (reason='rejected_by_safety_gate').

Returns:

(cleaned, betas, diag) — cleaned EMG matrix, per-channel fitted coefficient vectors, and a {'per_channel': [...]} diagnostics dict.

Parameters:

emg_2d (ndarray) –
acc_by_emg (Dict[int, ndarray] | None) –
max_lag_samples (int) –
ridge_alpha (float) –
include_magnitude (bool) –
include_derivative (bool) –
min_variance_ratio (float) –
min_power_ratio (float) –

Return type:

tuple

delsys.cleaning.harmonize_multirate_inputs(emg_2d, emg_sr, *, ekg_1d=None, ekg_sr=None, acc_by_emg=None, acc_sr=None, target_sr=None)

Resample EMG / EKG / ACC streams to a single sampling rate.

All outputs are tail-trimmed to the same length.

Parameters:

emg_2d (ndarray) – EMG matrix, shape (time, channels).
emg_sr (float) – EMG sampling rate.
ekg_1d (ndarray | None) – Optional EKG reference, 1-D.
ekg_sr (float | None) – EKG sampling rate; required when ekg_1d is given.
acc_by_emg (Dict[int, ndarray] | None) – Map {emg_channel_index: acc_stream}. Each value is 1-D or 2-D.
acc_sr (float | Dict[int, float] | None) – Single sampling rate for every ACC stream, or {channel_index: sr} for per-channel rates.
target_sr (float | None) – Output sampling rate. Defaults to emg_sr (so EMG passes through, EKG / ACC are resampled to match).

Returns:

Dict with keys 'sr', 'emg', 'ekg', 'acc_by_emg', 'n_samples'.

Return type:

Dict[str, Any]

delsys.cleaning.run_pipeline(emg_2d, sr, *, ekg_1d=None, acc_by_emg=None, feature_names=None, time=None, config=None)

Run the staged cleaner over already-aligned numpy arrays.

Inputs must already share a sampling rate; use harmonize_multirate_inputs() first if they do not (or call Log.clean_emg_ekg_artifact, which harmonizes for you).

Stage order: preprocess → ECG (optional) → motion (optional). Each stage stores its output in result.stages.

Parameters:

emg_2d (ndarray) – EMG matrix, shape (time, channels).
sr (float) – Sampling rate (Hz) shared by every input array.
ekg_1d (ndarray | None) – EKG reference, 1-D. Required when config.use_ecg_stage=True.
acc_by_emg (Dict[int, ndarray] | None) – Map {emg_channel_index: acc_stream}. Used only when config.use_motion_stage=True.
feature_names (List[str] | None) – Optional per-EMG-channel labels carried into ECG-stage diagnostics.
time (ndarray | None) – Optional time grid for emg_2d; reused on the result when shapes line up. When None, a fresh grid is built from sr.
config (CleaningConfig | None) – CleaningConfig; defaults to CleaningConfig().

Returns:

CleaningResult with the cleaned EMG, per-stage snapshots, and diagnostics.

Return type:

CleaningResult

Metadata records

Foundational metadata records used throughout the package.

Three small namedtuples that travel through the loader pipeline:

SensorLog — one row of a manual channelmap text file (delsys_channelmap.txt).
SensorInfo — combined metadata from the channelmap and the CSV header. Held by every Signal and every Sensor.
SigInfoDelsys — parsed information for a single CSV column, produced by _parse_sig_name().

Living in their own module avoids circular imports between signals.py (which needs SensorInfo for Signal) and sensor.py (which uses Signal and also needs SensorInfo).

class delsys._metadata.SensorLog(number, type_sensorlog, lrc, location)

location: Alias for field number 3

lrc: Alias for field number 2

number: Alias for field number 0

type_sensorlog: Alias for field number 1

class delsys._metadata.SensorInfo(name, modalities, number, type_sensorlog, lrc, location)

location: Alias for field number 5

lrc: Alias for field number 4

modalities: Alias for field number 1

name: Alias for field number 0

number: Alias for field number 2

type_sensorlog: Alias for field number 3

class delsys._metadata.SigInfoDelsys(sensor_name, modality, sensor_number, subchannel)

modality: Alias for field number 1

sensor_name: Alias for field number 0

sensor_number: Alias for field number 2

subchannel: Alias for field number 3

Constants

Module-level constants for the delsys package.

The metadata namedtuples (SensorLog, SensorInfo, SigInfoDelsys) that used to live here moved to delsys._metadata to avoid a circular import between delsys.signals and delsys.sensor.

delsys._constants.TARGET_SR: Dict[str, float | None] = {'ACC': 120, 'Analog': 2400, 'EKG': 120, 'EMGD': 1920, 'EMGQ': 1920, 'EMGS': 1920, 'FSR': 120, 'GYRO': 120, 'HR': 1, 'SmO2': 5, 'Thb': 5, 'VO2': 1}

Default per-modality target sampling rate (Hz) used to normalize raw signals during loading. Override by passing target_sr= to delsys.Log.

Keys are normalized modality tags (see SUBCHANNEL_MAP). A value of None for a modality means “skip resampling” — currently honored by the Discover-basic and Discover-link Trigno-base parsers; not yet by the EMGworks parser or by link devices (see TODO.md).

delsys._constants.SUBCHANNEL_MAP: Dict[str, Tuple[str, ...]] = {'ACC': ('X', 'Y', 'Z'), 'Analog': ('A',), 'EKG': ('A',), 'EMGD': ('A', 'B'), 'EMGQ': ('A', 'B', 'C', 'D'), 'EMGS': ('A',), 'FSR': ('A', 'B', 'C', 'D'), 'GYRO': ('X', 'Y', 'Z'), 'HR': ('HeartRate',), 'VO2': ('BreathingCycle', 'Resp.Rate', 'TidalVol.', 'Ventilation(L/min)', 'FeO2(%)', 'VO2Absolute', 'AmbientPressure', 'FlowSensor', 'OxygenSensor')}: Canonical sub-channel ordering per modality. Used by delsys.Sensor to stack per-channel delsys.Signal objects into multi-channel modality bundles (delsys.IMU, delsys.FSR, delsys.VO2Master) in a stable order.

delsys._constants.APPLICATIONS: Tuple[str, ...] = ('EMGworks', 'Trigno Discover'): Recognized acquisition applications. The header parser uses this to gate which per-format reader runs.

delsys._constants.LINK_DEVICE_REGISTRY: Dict[str, Tuple[str, int]] = {'HR Strap': ('HR', 901), 'VO2 Master': ('VO2', 900)}

Link-device identification. Maps a substring of the column-header sensor_name to (modality, synthetic_sensor_number). The per-format parser walks this dict in registration order; first substring match wins. The synthetic numbers sit far from Trigno-Base sensor numbers (typically 1–16) so sensor_number lookups remain unambiguous.

Adding a new link device requires:

one entry here,
a SUBCHANNEL_MAP entry for its modality,
a TARGET_SR entry,
a MODALITY_REGISTRY entry in delsys.sensor.