BIDS Helpers

Helpers for the Brain Imaging Data Structure (BIDS): locate a sibling _channels.tsv / _events.tsv next to a recording, apply its per-channel types/units (apply_channels_tsv), and load its authoritative event table into rec.events (apply_events_tsv). apply_channels_tsv is applied automatically by Recording.from_file (unless bids_channels="off"); apply_events_tsv is called explicitly when the sidecar events should override the data file's own markers.

apply_channels_tsv_to_stream is the same thing for a recording that is never loaded into a Recording at all: the streaming Zarr exporter (stream_to_zarr, also bids_channels="auto" by default) has only a channel table and reads the samples window by window afterwards, so it takes a per-channel conversion factor to apply later instead of rescaling a column now. Both functions decide every per-channel question with the same internal table, so the two export paths cannot disagree about what a sidecar means.

Choosing the sidecar: bids_channels

Recording.from_file and stream_to_zarr take the same bids_channels argument with the same meanings, resolved by the same resolve_channels_tsv, so a caller can move a recording between the two paths without its units changing:

Value	Meaning
"auto" (default)	resolve the sibling _channels.tsv via find_channels_tsv; a recording with no sidecar is left as the importer read it
"off", or None	do not look for a sidecar at all
a path (str or PathLike)	use this sidecar, wherever it lives
a pandas.DataFrame	use this table, already loaded

The path and DataFrame forms exist for a recording whose sidecar is not next to it: a converter that filters or rewrites a recording into a scratch directory still has to apply the original recording's channels.tsv, and must pass it explicitly, because "auto" looks beside the file it is given.

A trailing slash on a directory-valued recording (a CTF .ds, a 4D/BTi directory) is stripped before the lookup, so .ds and .ds/ resolve the same sidecar on both paths.

apply_channels_tsv itself accepts either a path or a DataFrame, and treats them identically (missing cells in a supplied frame become the empty cell that means "declares nothing", matching how the TSV is read).

Units are converted, not relabelled

The units column describes the numbers, so adopting it rescales the samples (see Physical Units). An MNE-backed importer returns SI volts; a sidecar declaring uV therefore multiplies that channel by 10^6 and then sets the label, leaving values and unit in agreement. It is a no-op wherever the importer already reports the sidecar's unit (EDF via pyedflib).

What "idempotent" guarantees precisely: applying the same sidecar any number of times converts at most once and warns at most once. The second pass finds each channel's physical_dimension already equal to the declared unit (or its bids_unit already recorded) and does nothing. It is not a claim about applying two different sidecars in sequence -- those compose, so a channel read as V and then given a mV sidecar followed by a uV one ends up in uV, scaled by 10^6 overall.

Some channels are never rescaled, whatever the sidecar declares:

• discrete types (TRIG, SYSCLOCK, CTRL) hold codes rather than a measured quantity. MNE labels stim channels with the FIFF volts code while they carry integer event codes, so a sidecar declaring mV would turn codes 5/3/7 into 5000/3000/7000;
• channels with no samples: metadata without a data column has no numbers for a new label to agree with, so the label does not move either;
• units that are not convertible -- different quantities, or a spelling neither side can parse (n/a, a.u.).

In each case the importer's values and its label are kept and the sidecar's claim is recorded as channels[label]["bids_unit"] with a warning, so nothing is relabelled without being converted. Adopting a unit later clears any bids_unit left by an earlier disagreement.

A per-file summary lands in rec.metadata["channels_tsv_units"]:

{"converted": 2, "relabelled": 0, "kept_importer_unit": 1, "units_column_present": True}

units_column_present separates "the sidecar declared no units at all" from "the units were already correct", which the counters alone cannot.

Module Documentation

biosigio.bids

Brain Imaging Data Structure (BIDS) sidecar helpers.

When a data file follows the BIDS layout, the authoritative per-channel metadata lives in a sibling *_channels.tsv (the type and units columns), not in the data file's own headers. These helpers locate that sidecar and apply it to an :class:~biosigio.core.emg.Recording object so imported channels get their real BIDS types (e.g. SEEG) instead of header/label guesses.

The units column is a claim about the numbers, not just a label, so adopting it rescales the samples (see :func:_decide_unit). Applying a sidecar therefore leaves values and unit in agreement, which relabelling alone did not.

Two callers, one decision table. :func:apply_channels_tsv serves the in-memory path (a whole :class:~biosigio.core.emg.Recording in RAM, whose columns are rescaled in place), and :func:apply_channels_tsv_to_stream serves the bounded-memory streaming exporter (no Recording exists; the samples arrive window by window later). Both route every per-channel question through :func:_decide_unit, which decides and returns rather than mutating, so the two export paths cannot drift on what a sidecar means -- the failure issue #127 reports, where a dataset's small runs served microvolts and its large ones served volts.

DISCRETE_CHANNEL_TYPES = frozenset({'TRIG', 'SYSCLOCK', 'CTRL'}) module-attribute

_CONVERTED = 'converted' module-attribute

_KEPT = 'kept_importer_unit' module-attribute

_RELABELLED = 'relabelled' module-attribute

_UNCHANGED = 'unchanged' module-attribute

Recording

Core biosignal recording: signals + channels + events + metadata.

Modality-agnostic container for EEG / EMG / iEEG / MEG / stim / marker data imported from any supported format.

Attributes: signals (pd.DataFrame): Raw signal data with time as index. metadata (dict): Metadata dictionary containing recording information. channels (dict): Channel information including type, unit, sampling frequency. events (pd.DataFrame): Annotations or events associated with the signals, with columns 'onset', 'duration', 'description'.

Source code in biosigio/core/emg.py

class Recording:
    """
    Core biosignal recording: signals + channels + events + metadata.

    Modality-agnostic container for EEG / EMG / iEEG / MEG / stim / marker data
    imported from any supported format.

    Attributes:
        signals (pd.DataFrame): Raw signal data with time as index.
        metadata (dict): Metadata dictionary containing recording information.
        channels (dict): Channel information including type, unit, sampling frequency.
        events (pd.DataFrame): Annotations or events associated with the signals,
                               with columns 'onset', 'duration', 'description'.
    """

    def __init__(self):
        """Initialize an empty recording."""
        self.signals = None
        self.metadata = {}
        self.channels = {}
        # Initialize events as an empty DataFrame with specified columns
        self.events = pd.DataFrame(columns=["onset", "duration", "description"])

    def plot_signals(
        self,
        channels=None,
        time_range=None,
        offset_scale=0.8,
        uniform_scale=True,
        detrend=False,
        grid=True,
        title=None,
        show=True,
        plt_module=None,
    ):
        """
        Plot signals in a single plot with vertical offsets.

        Args:
            channels: List of channels to plot. If None, plot all channels.
            time_range: Tuple of (start_time, end_time) to plot. If None, plot all data.
            offset_scale: Portion of allocated space each signal can use (0.0 to 1.0).
            uniform_scale: Whether to use the same scale for all signals.
            detrend: Whether to remove mean from signals before plotting.
            grid: Whether to show grid lines.
            title: Optional title for the figure.
            show: Whether to display the plot.
            plt_module: Matplotlib pyplot module to use.
        """
        # Delegate to the static plotting function in visualization module
        static_plot_signals(
            rec_object=self,
            channels=channels,
            time_range=time_range,
            offset_scale=offset_scale,
            uniform_scale=uniform_scale,
            detrend=detrend,
            grid=grid,
            title=title,
            show=show,
            plt_module=plt_module,
        )

    @classmethod
    def _infer_importer(cls, filepath: str) -> ImporterName:
        """
        Infer the importer to use based on the file extension.
        """
        # rstrip path separators so a Zarr store passed as a directory with a
        # trailing slash (e.g. "rec.zarr/") still resolves by its ".zarr" suffix.
        extension = os.path.splitext(filepath.rstrip("/\\"))[1].lower()
        if extension in {".edf", ".bdf"}:
            return "edf"
        elif extension in {".set"}:
            return "eeglab"
        elif extension in {".otb", ".otb+"}:
            return "otb"
        elif extension in {".csv", ".txt"}:
            return "csv"
        elif extension in {".hea", ".dat", ".atr"}:
            return "wfdb"
        elif extension in {".xdf", ".xdfz"}:
            return "xdf"
        elif extension in {".fif", ".ds", ".con", ".sqd", ".kdf"}:
            # MEG via MNE: .fif (Neuromag/FIF), CTF .ds directory, KIT/Yokogawa
            # .con/.sqd/.kdf. The MEGImporter dispatches on extension internally.
            return "meg"
        elif extension in {".vhdr"}:
            return "brainvision"
        elif extension == ".mefd":
            # MEF3 (a directory, like CTF .ds) is a distinct iEEG modality with its
            # own MNE/pymef version floor -- see importers/mef3.py -- so it is its
            # own importer name rather than folding into "meg".
            return "mef3"
        elif extension == "" and os.path.isdir(filepath.rstrip("/\\")):
            # 4D/BTi ships as a directory with NO extension, so it cannot be
            # recognized above; fall back to content-based detection. Lazy
            # import: avoids importing the importers package (and MNE) just to
            # infer a name, and avoids a core<->importers import cycle.
            #
            # _resolve_bti_reader_kwargs is the ONE place that decides "is this
            # a valid BTi layout" and raises UnsupportedFormatError if not --
            # shared with MEGImporter._read_raw and the streaming path (see
            # importers/meg.py's module docstring) so there is exactly one
            # raise site to keep correct, not one per call site.
            from ..importers.meg import _resolve_bti_reader_kwargs

            _resolve_bti_reader_kwargs(filepath.rstrip("/\\"))
            return "meg"
        elif extension in {".parquet", ".feather", ".arrow"}:
            return "tabular"
        elif extension in {
            ".rhd",
            ".rhs",
            ".ns1",
            ".ns2",
            ".ns3",
            ".ns4",
            ".ns5",
            ".ns6",
            ".smr",
            ".smrx",
            ".plx",
            ".pl2",
            ".trc",
            ".ncs",
        }:
            return "neo"
        elif extension == ".zarr":
            return "zarr"
        else:
            raise UnsupportedFormatError(f"Unsupported file extension: {extension!r}")

    @classmethod
    def from_file(
        cls,
        filepath: str,
        importer: ImporterName | None = None,
        force_csv: bool = False,
        bids_channels: "str | os.PathLike | pd.DataFrame | None" = "auto",
        mixed_rate: str = "error",
        **kwargs,
    ) -> "Recording":
        """
        The method to create a Recording object from file.

        Args:
            filepath: Path to the input file
            importer: Name of the importer to use. Can be one of the following:
                - 'trigno': Delsys Trigno EMG system (CSV)
                - 'otb': OTB/OTB+ EMG system (OTB, OTB+)
                - 'eeglab': EEGLAB .set files (SET)
                - 'edf': EDF/EDF+/BDF/BDF+ format (EDF, BDF)
                - 'csv': Generic CSV (or TXT) files with columnar data
                - 'wfdb': Waveform Database (WFDB)
                - 'xdf': XDF format (multi-stream Lab Streaming Layer files)
                - 'meg': MEG via MNE (.fif, CTF .ds, KIT .con/.sqd/.kdf, 4D/BTi
                  directory (no extension, detected by content); requires the
                  'meg' extra)
                - 'brainvision': BrainVision .vhdr via MNE (requires the 'meg' extra)
                - 'mef3': MEF3 iEEG via MNE (.mefd directory; requires the 'mef3'
                  extra -- mne>=1.12 plus pymef, stricter than the 'meg' extra)
                - 'tabular': biosigIO Parquet/Arrow/Feather (requires the 'arrow' extra)
                - 'neo': proprietary electrophysiology formats via python-neo
                  (Intan, Blackrock, Spike2, Plexon, Micromed, Neuralynx, ...;
                  requires the 'neo' extra)
                - 'zarr': biosigIO Zarr serving store (requires the 'zarr' extra)
                If None, the importer will be inferred from the file extension.
                Automatic import is supported for CSV/TXT files.
            force_csv: If True and importer is 'csv', forces using the generic CSV
                      importer even if the file appears to match a specialized format.
            bids_channels: Which BIDS ``_channels.tsv`` to apply over the
                      importer's inferred per-channel type/units. 'auto'
                      (default) looks for the sibling sidecar next to the file;
                      'off' (or None) disables the lookup; a path or a pandas
                      DataFrame is used as given, for a recording whose sidecar
                      is not adjacent to it (a filtered copy on scratch, say).
                      Adopting a declared unit converts the samples into it, not
                      just the label. ``stream_to_zarr`` takes the same argument
                      with the same meanings, so the in-memory and streaming
                      export paths agree on a recording (issue #127).
            mixed_rate: Policy for an EDF/BDF file whose signals carry differing
                      per-channel sampling rates (ignored for every other format,
                      which is single-rate). 'error' (default) raises -- biosigIO
                      stores one uniform grid and will not fabricate a common one
                      silently. 'resample' upsamples the slower channels to the
                      fastest rate (a lossy derived view; each channel keeps its
                      native rate as ``original_sample_frequency``).
            **kwargs: Additional arguments passed to the importer.
                For XDF files, useful kwargs include:
                - stream_names: List of stream names to import
                - stream_types: List of stream types to import (e.g., ["EMG", "EXG"])
                - stream_ids: List of stream IDs to import

        Returns:
            Recording: New Recording object with loaded data
        """
        if importer is None:
            importer = cls._infer_importer(filepath)

        importers = {
            "trigno": "TrignoImporter",  # CSV with Delsys Trigno Headers
            "otb": "OTBImporter",  # OTB/OTB+ EMG system data
            "edf": "EDFImporter",  # EDF/EDF+/BDF format
            "eeglab": "EEGLABImporter",  # EEGLAB .set files
            "csv": "CSVImporter",  # Generic CSV/Text files
            "wfdb": "WFDBImporter",  # Waveform Database format
            "xdf": "XDFImporter",  # XDF multi-stream format
            "meg": "MEGImporter",  # MEG via MNE (.fif, CTF .ds, KIT .con/.sqd/.kdf, 4D/BTi)
            "brainvision": "BrainVisionImporter",  # BrainVision via MNE (.vhdr)
            "mef3": "MEF3Importer",  # MEF3 iEEG via MNE (.mefd; requires the 'mef3' extra)
            "tabular": "TabularImporter",  # biosigIO Parquet / Arrow / Feather
            "neo": "NeoImporter",  # proprietary ephys via python-neo
            "zarr": "ZarrImporter",  # biosigIO Zarr serving store
        }

        if importer not in importers:
            raise ValueError(
                f"Unsupported importer: {importer}. "
                f"Available importers: {list(importers.keys())}\n"
                "- trigno: Delsys Trigno EMG system\n"
                "- otb: OTB/OTB+ EMG system\n"
                "- edf: EDF/EDF+/BDF format\n"
                "- eeglab: EEGLAB .set files\n"
                "- csv: Generic CSV/Text files\n"
                "- wfdb: Waveform Database\n"
                "- xdf: XDF multi-stream format\n"
                "- meg: MEG via MNE (.fif, CTF .ds, KIT .con/.sqd/.kdf, 4D/BTi directory)\n"
                "- brainvision: BrainVision via MNE (.vhdr)\n"
                "- mef3: MEF3 iEEG via MNE (.mefd directory; requires the 'mef3' extra)\n"
                "- tabular: biosigIO Parquet/Arrow/Feather (.parquet, .feather, .arrow)\n"
                "- neo: proprietary electrophysiology formats via python-neo "
                "(Intan, Blackrock, Spike2, Plexon, Micromed, Neuralynx, ...)\n"
                "- zarr: biosigIO Zarr serving store (.zarr)"
            )

        # If using CSV importer and force_csv is set, pass it as force_generic
        if importer == "csv":
            kwargs["force_generic"] = force_csv

        # mixed_rate is only meaningful for EDF/BDF, the one format whose signals may
        # carry differing per-channel sampling rates; forward it there and nowhere
        # else (so the default "error" never reaches an importer that can't use it).
        if importer == "edf":
            kwargs["mixed_rate"] = mixed_rate

        # Import the appropriate importer class
        importer_module = __import__(
            f"biosigio.importers.{importer}", globals(), locals(), [importers[importer]]
        )
        importer_class = getattr(importer_module, importers[importer])

        # Create importer instance and load data
        rec = importer_class().load(filepath, **kwargs)

        # Record provenance: which format this recording came from. setdefault so a
        # re-imported serialization file (tabular/zarr) keeps the ORIGINAL
        # source_format restored from its metadata rather than being relabeled.
        rec.metadata.setdefault("source_format", importer)

        # In a BIDS layout, the sibling _channels.tsv is the authoritative source
        # of per-channel type/units; apply it over the importer's header/label
        # guesses unless explicitly disabled with bids_channels="off". The
        # argument is interpreted by bids.resolve_channels_tsv, shared with
        # stream_to_zarr so the two export paths cannot read it differently.
        from ..bids import apply_channels_tsv, resolve_channels_tsv

        channels_tsv = resolve_channels_tsv(filepath, bids_channels)
        if channels_tsv is not None:
            apply_channels_tsv(rec, channels_tsv)

        return rec

    def select_channels(
        self,
        channels: str | list[str] | None = None,
        channel_type: str | None = None,
        inplace: bool = False,
        *,
        modality: str | None = None,
    ) -> "Recording":
        """
        Select specific channels from the data and return a new Recording object.

        Args:
            channels: Channel name or list of channel names to select. If None and
                    channel_type is specified, selects all channels of that type.
            channel_type: Type of channels to select ('EMG', 'ACC', 'GYRO', etc.).
                        If specified with channels, filters the selection to only
                        channels of this type.

        Returns:
            Recording: A new Recording object containing only the selected channels

        Examples:
            # Select specific channels
            new_rec = rec.select_channels(['EMG1', 'ACC1'])

            # Select all EMG channels
            emg_only = rec.select_channels(channel_type='EMG')

            # Select specific EMG channels only, this example does not select ACC channels
            emg_subset = rec.select_channels(['EMG1', 'ACC1'], channel_type='EMG')
        """
        if self.signals is None:
            raise ValueError("No signals loaded")

        if channels is None and channel_type is None and modality is None:
            raise ValueError("Specify at least one of: channels, channel_type, or modality.")

        # If type/modality specified but no channels, select all matching channels
        if channels is None and channel_type is not None:
            channels = self.get_channels_by_type(channel_type)
            if not channels:
                raise ValueError(f"No channels found of type: {channel_type}")
        elif channels is None and modality is not None:
            channels = self.get_channels_by_modality(modality)
            if not channels:
                raise ValueError(f"No channels found of modality: {modality}")
        elif isinstance(channels, str):
            channels = [channels]

        if channels is None:
            raise ValueError("Specify at least one of: channels, channel_type, or modality.")

        # Validate channels exist
        if not all(ch in self.signals.columns for ch in channels):
            missing = [ch for ch in channels if ch not in self.signals.columns]
            raise ValueError(f"Channels not found: {missing}")

        # Filter by type if specified
        if channel_type is not None:
            channels = [ch for ch in channels if self.channels[ch]["channel_type"] == channel_type]
            if not channels:
                raise ValueError(f"None of the selected channels are of type: {channel_type}")

        # Filter by modality if specified
        if modality is not None:
            canonical_modality = validate_modality(modality)
            channels = [
                ch for ch in channels if self.channels[ch].get("modality") == canonical_modality
            ]
            if not channels:
                raise ValueError(f"None of the selected channels are of modality: {modality}")

        # Create new Recording object
        new_rec = Recording()

        # Copy selected signals and channels
        new_rec.signals = self.signals[channels].copy()
        new_rec.channels = {ch: self.channels[ch].copy() for ch in channels}

        # Copy metadata
        new_rec.metadata = self.metadata.copy()

        if not inplace:
            return new_rec
        else:
            self.signals = new_rec.signals
            self.channels = new_rec.channels
            self.metadata = new_rec.metadata
            return self

    def resample(self, target_rate: float) -> "Recording":
        """Return a NEW, anti-aliased down-sampled copy of this recording.

        Low-resolution demos need a smaller, lighter recording; this rebuilds the
        uniform signal grid at ``target_rate`` using a polyphase resampler
        (``scipy.signal.resample_poly``), which applies a Kaiser-windowed sinc
        anti-alias FIR before decimation. A naive stride-decimation would fold
        energy above the new Nyquist back into the band (aliasing); resample_poly
        removes that energy first, so no aliasing occurs.

        Non-destructive: ``self`` is left untouched and a new Recording is returned,
        mirroring ``select_channels``'s copy semantics.

        Resampling factors come from the integer source/target rates:
        ``g = gcd(int(src), int(target)); up = int(target)//g; down = int(src)//g``
        and ``resample_poly(x, up, down)`` runs once, vectorized over all channels
        along ``axis=0``.

        Args:
            target_rate: Desired sampling rate in Hz. Must be <= the source rate
                (this is a DOWN-sampling helper). A target equal to the source
                returns an unchanged copy; a target above it raises ``ValueError``
                rather than silently up-sampling (up-sampling cannot recover
                detail and is out of scope for the low-res pipeline).

        Returns:
            Recording: A new Recording with the resampled signals, each channel's
                ``sample_frequency`` set to the achieved rate (source * up / down,
                which equals ``target_rate`` for integer rates), and channel/recording
                metadata and events preserved. Events are unchanged because their
                onsets/durations are in SECONDS, which stay valid under any rate
                change (only the per-sample grid shrinks, not wall-clock time).

        Raises:
            ValueError: If no signals are loaded, if channels do not share a single
                ``sample_frequency`` (biosigio stores one uniform grid; mixed-rate
                resampling is out of scope), or if ``target_rate`` exceeds the
                source rate.
        """
        from math import gcd

        from scipy.signal import resample_poly

        if self.signals is None:
            raise ValueError("No signals loaded")

        if target_rate <= 0:
            raise ValueError(f"target_rate must be positive, got {target_rate}")

        # biosigio stores all channels on one uniform-length grid; a per-channel rate
        # mix is out of scope here, matching the exporter's single-rate guard.
        distinct_rates = {info["sample_frequency"] for info in self.channels.values()}
        if len(distinct_rates) > 1:
            raise ValueError(
                "Resampling requires a single sampling rate across all channels, but "
                f"multiple were found: {sorted(distinct_rates)} Hz. biosigio stores one "
                "uniform grid; resample each rate group separately."
            )

        source_rate = float(next(iter(distinct_rates)))

        # Down-sampling only: refuse to up-sample/alias; an equal rate is a no-op
        # copy so callers can resample unconditionally without special-casing.
        if target_rate > source_rate:
            raise ValueError(
                f"target_rate {target_rate} Hz exceeds source rate {source_rate} Hz; "
                "resample() only down-samples (low-res). Up-sampling is out of scope."
            )

        new_rec = Recording()
        new_rec.channels = {ch: info.copy() for ch, info in self.channels.items()}
        new_rec.metadata = self.metadata.copy()
        # Onsets/durations are in SECONDS, so they remain valid after the grid
        # changes; copy them through unchanged.
        new_rec.events = self.events.copy() if self.events is not None else self.events

        if target_rate == source_rate:
            # No grid change: copy signals through untouched (fresh RangeIndex for
            # consistency with the resampled path).
            new_rec.signals = self.signals.copy().reset_index(drop=True)
            return new_rec

        # Rational resampling factors from the integer rates.
        src_i = int(round(source_rate))
        tgt_i = int(round(target_rate))
        g = gcd(src_i, tgt_i)
        up = tgt_i // g
        down = src_i // g

        # The achieved rate is exactly source * up / down. Store THAT, not the
        # requested float, so the metadata can never disagree with the data (a
        # non-integer or odd target snaps to the nearest achievable rational rate;
        # warn so the caller knows). This avoids silently writing e.g. 99.5 Hz
        # onto a grid that resample_poly actually produced at 100 Hz.
        actual_rate = source_rate * up / down
        if abs(actual_rate - target_rate) > 1e-9:
            logging.warning(
                "Requested resample to %g Hz; nearest achievable rational rate is "
                "%g Hz, which is what is stored on the channels.",
                target_rate,
                actual_rate,
            )

        columns = list(self.signals.columns)
        data = self.signals.to_numpy(dtype=float)
        # resample_poly over axis=0 resamples every channel column at once with the
        # shared anti-alias FIR.
        resampled = resample_poly(data, up, down, axis=0)

        new_rec.signals = pd.DataFrame(resampled, columns=columns)
        new_rec.signals.index = pd.RangeIndex(len(new_rec.signals))
        for info in new_rec.channels.values():
            info["sample_frequency"] = actual_rate

        return new_rec

    def get_channel_types(self) -> list[str]:
        """
        Get list of unique channel types in the data.

        Returns:
            List of channel types (e.g., ['EMG', 'ACC', 'GYRO'])
        """
        return list({info["channel_type"] for info in self.channels.values()})

    def get_channels_by_type(self, channel_type: str) -> list[str]:
        """
        Get list of channels of a specific type.

        Args:
            channel_type: Type of channels to get ('EMG', 'ACC', 'GYRO', etc.)

        Returns:
            List of channel names of the specified type
        """
        return [ch for ch, info in self.channels.items() if info["channel_type"] == channel_type]

    def get_modalities(self) -> list[str]:
        """
        Get the list of unique modalities present in the data.

        Returns:
            List of modalities (e.g., ['EEG', 'EMG', 'MISC']).
        """
        return list(
            {info.get("modality") for info in self.channels.values() if info.get("modality")}
        )

    def get_channels_by_modality(self, modality: str) -> list[str]:
        """
        Get the channels belonging to a given modality.

        Args:
            modality: Modality to filter by ('EEG', 'EMG', 'IEEG', 'MEG', 'BEH', 'MISC').

        Returns:
            List of channel names of the specified modality.
        """
        canonical_modality = validate_modality(modality)
        return [
            ch for ch, info in self.channels.items() if info.get("modality") == canonical_modality
        ]

    def to_edf(
        self,
        filepath: str,
        method: str = "both",
        fft_noise_range: tuple | None = None,
        svd_rank: int | None = None,
        precision_threshold: float = 0.01,
        format: Literal["auto", "edf", "bdf"] = "auto",
        bypass_analysis: bool | None = None,
        verify: bool = False,
        verify_tolerance: float = 1e-6,
        verify_channel_map: dict[str, str] | None = None,
        verify_plot: bool = False,
        events_df: pd.DataFrame | None = None,
        create_channels_tsv: bool = True,
        clip_outliers: bool | str = "auto",
        **kwargs,
    ) -> dict | None:
        """
        Export the recording to EDF/BDF format, optionally including events.

        Args:
            filepath: Path to save the EDF/BDF file
            method: Method for signal analysis ('svd', 'fft', or 'both')
                'svd': Uses Singular Value Decomposition for noise floor estimation
                'fft': Uses Fast Fourier Transform for noise floor estimation
                'both': Uses both methods and takes the minimum noise floor (default)
            fft_noise_range: Optional tuple (min_freq, max_freq) specifying frequency range for noise in FFT method
            svd_rank: Optional manual rank cutoff for signal/noise separation in SVD method
            precision_threshold: Maximum acceptable precision loss percentage (default: 0.01%)
            format: Format to use ('auto', 'edf', or 'bdf'). Default is 'auto'.
                    If 'edf' or 'bdf' is specified, that format will be used directly.
                    If 'auto', the format (EDF/16-bit or BDF/24-bit) is chosen based
                    on signal analysis to minimize precision loss while preferring EDF
                    if sufficient.
            bypass_analysis: If True, skip signal analysis step when format is explicitly
                             set to 'edf' or 'bdf'. If None (default), analysis is skipped
                             automatically when format is forced. Set to False to force
                             analysis even with a specified format. Ignored if format='auto'.
            verify: If True, reload the exported file and compare signals with the original
                    to check for data integrity loss. Results are printed. (default: False)
            verify_tolerance: Absolute tolerance used when comparing signals during verification. (default: 1e-6)
            verify_channel_map: Optional dictionary mapping original channel names (keys)
                                to reloaded channel names (values) for verification.
                                Used if `verify` is True and channel names might differ.
            verify_plot: If True and verify is True, plots a comparison of original vs reloaded signals.
            events_df: Optional DataFrame with events ('onset', 'duration', 'description').
                      If None, uses self.events. (This provides flexibility)
            create_channels_tsv: If True, create a BIDS-compliant channels.tsv file (default: True)
            clip_outliers: Singularity handling for the per-channel physical window.
                'auto' (default) keeps the full range losslessly but clips rare extreme
                outliers to a robust window only when keeping them would crater the bulk
                signal's resolution at the chosen format (with a warning); True always
                clips to the robust window; False never clips. See EDFExporter.export for
                the advanced ``outlier_sigmas`` / ``min_effective_bits`` knobs.
            **kwargs: Additional arguments for the EDF exporter

        Returns:
            Union[str, None]: If verify is True, returns a string with verification results.
                             Otherwise, returns None.

        Raises:
            ValueError: If no signals are loaded
        """
        from ..exporters.edf import EDFExporter  # Local import

        if self.signals is None:
            raise ValueError("No signals loaded")

        # --- Determine if analysis should be bypassed ---
        final_bypass_analysis = False
        if format.lower() == "auto":
            if bypass_analysis is True:
                logging.warning(
                    "bypass_analysis=True ignored because format='auto'. Analysis is required."
                )
            # Analysis is always needed for 'auto' format
            final_bypass_analysis = False
        elif format.lower() in ["edf", "bdf"]:
            if bypass_analysis is None:
                # Default behaviour: skip analysis if format is forced
                final_bypass_analysis = True
                msg = (
                    f"Format forced to '{format}'. Skipping signal analysis for faster export. "
                    "Set bypass_analysis=False to force analysis."
                )
                logging.log(logging.CRITICAL, msg)
            elif bypass_analysis is True:
                final_bypass_analysis = True
                logging.log(logging.CRITICAL, "bypass_analysis=True set. Skipping signal analysis.")
            else:  # bypass_analysis is False
                final_bypass_analysis = False
                logging.info(
                    f"Format forced to '{format}' but bypass_analysis=False. Performing signal analysis."
                )
        else:
            # Should not happen if Literal type hint works, but good practice
            logging.warning(
                f"Unknown format '{format}'. Defaulting to 'auto' behavior (analysis enabled)."
            )
            format = "auto"
            final_bypass_analysis = False

        # Determine which events DataFrame to use
        if events_df is None:
            events_to_export = self.events
        else:
            events_to_export = events_df

        # Combine parameters
        all_params: dict[str, Any] = {
            "precision_threshold": precision_threshold,
            "method": method,
            "fft_noise_range": fft_noise_range,
            "svd_rank": svd_rank,
            "format": format,
            "bypass_analysis": final_bypass_analysis,
            "events_df": events_to_export,  # Pass the events dataframe
            "create_channels_tsv": create_channels_tsv,
            "clip_outliers": clip_outliers,
            **kwargs,
        }

        EDFExporter.export(self, filepath, **all_params)

        verification_report_dict = None
        if verify:
            logging.info(f"Verification requested. Reloading exported file: {filepath}")
            try:
                # Reload the exported file
                reloaded_rec = Recording.from_file(filepath, importer="edf")

                logging.info("Comparing original signals with reloaded signals...")
                # Compare signals using the imported function
                verification_results = compare_signals(
                    self, reloaded_rec, tolerance=verify_tolerance, channel_map=verify_channel_map
                )

                # Generate and log report using the imported function
                report_verification_results(verification_results, verify_tolerance)
                verification_report_dict = verification_results

                # Plot comparison using imported function if requested
                summary = verification_results.get("channel_summary", {})
                comparison_mode = summary.get("comparison_mode", "unknown")
                compared_count = sum(1 for k in verification_results if k != "channel_summary")

                if verify_plot and compared_count > 0 and comparison_mode != "failed":
                    plot_comparison(self, reloaded_rec, channel_map=verify_channel_map)
                elif verify_plot:
                    logging.warning(
                        "Skipping verification plot: No channels were successfully compared."
                    )

            except Exception as e:
                logging.error(f"Verification failed during reload or comparison: {e}")
                verification_report_dict = {
                    "error": str(e),
                    "channel_summary": {"comparison_mode": "failed"},
                }

        return verification_report_dict

    def to_parquet(self, filepath: str) -> str:
        """Export to a self-describing biosigIO Parquet file.

        Signals are stored as a columnar table (channels = columns, time index
        preserved); channels/events/metadata travel in the file's schema metadata,
        so ``Recording.from_file`` round-trips it losslessly. Great for analytics
        (DuckDB/Polars/pandas/Spark). Requires the ``arrow`` extra (pyarrow).

        Args:
            filepath: Output ``.parquet`` path.

        Returns:
            str: The written file path.
        """
        from ..exporters.tabular import TabularExporter

        return TabularExporter.to_parquet(self, filepath)

    def to_arrow(self, filepath: str) -> str:
        """Export to a biosigIO Arrow/Feather file (fast zero-copy IPC).

        Same self-describing schema as :meth:`to_parquet`; round-trips via
        ``Recording.from_file``. Requires the ``arrow`` extra (pyarrow).

        Args:
            filepath: Output ``.feather`` / ``.arrow`` path.

        Returns:
            str: The written file path.
        """
        from ..exporters.tabular import TabularExporter

        return TabularExporter.to_arrow(self, filepath)

    def to_zarr(self, filepath: str, **kwargs) -> str:
        """Export to a sharded Zarr v3 serving store with a min/max view pyramid.

        Writes one cloud-native store that serves viewing, inference, and training
        from a single conversion: ``level 0`` of each ``(modality, rate)`` group is
        the anti-aliased, per-modality-resampled inference signal, with a min/max
        render pyramid above it (flagged not-for-inference). A derived serving copy,
        not the archival source (BIDS/EDF stay authoritative). Requires the ``zarr``
        extra (zarr v3). See :class:`~biosigio.exporters.zarr.ZarrExporter` for the
        tuning knobs (``modality_rates``, ``dtype``, level-0 chunk/shard sizing,
        ``view_chunk_columns`` for the view pyramid, ...).

        Args:
            filepath: Output store path (``.zarr`` appended if missing).
            **kwargs: Forwarded to :meth:`ZarrExporter.export`.

        Returns:
            str: The written store path.
        """
        from ..exporters.zarr import ZarrExporter

        # The empty-signal guard lives once, in ZarrExporter.export ("No signals
        # loaded"), matching the tabular path; no duplicate guard here.
        return ZarrExporter.export(self, filepath, **kwargs)

    def set_metadata(self, key: str, value: Any) -> None:
        """
        Set metadata value.

        Args:
            key: Metadata key
            value: Metadata value
        """
        self.metadata[key] = value

    def get_metadata(self, key: str) -> Any:
        """
        Get metadata value.

        Args:
            key: Metadata key

        Returns:
            Value associated with the key
        """
        return self.metadata.get(key)

    def has_metadata(self, key: str) -> bool:
        """Return True if ``key`` is present in the recording metadata."""
        return key in self.metadata

    def get_n_channels(self) -> int:
        """Number of channels in the recording."""
        return len(self.channels)

    def get_n_samples(self) -> int:
        """Number of time samples per channel (0 if no signals are loaded)."""
        return 0 if self.signals is None else len(self.signals)

    def get_sampling_frequency(self) -> float:
        """Sampling frequency in Hz, when all channels share a single rate.

        Raises:
            ValueError: if no channels are loaded, or channels have differing
                sampling frequencies; for a mixed-rate recording read
                ``channels[ch]["sample_frequency"]`` per channel instead.
        """
        if not self.channels:
            raise ValueError("No channels loaded")
        rates = {info["sample_frequency"] for info in self.channels.values()}
        if len(rates) > 1:
            raise ValueError(
                "Channels have differing sampling frequencies; read "
                "channels[ch]['sample_frequency'] per channel instead."
            )
        return float(next(iter(rates)))

    def get_duration(self) -> float:
        """Total recording duration in seconds (n_samples / sampling_frequency).

        Computed from the time index spacing, so it is the full window length
        (one sample period longer than the last sample's timestamp). Returns 0.0
        when fewer than two samples are loaded (a single sample has no inferable
        sample period).
        """
        if self.signals is None or len(self.signals) < 2:
            return 0.0
        index = self.signals.index
        sample_period = float(index[1] - index[0])
        return len(index) * sample_period

    def add_channel(
        self,
        label: str,
        data: np.ndarray,
        sample_frequency: float,
        physical_dimension: str,
        channel_type: str,
        *,
        modality: str | None = None,
        prefilter: str = "n/a",
    ) -> None:
        """
        Add a new channel to the recording.

        Args:
            label: Channel label or name (as per EDF specification)
            data: Channel data
            sample_frequency: Sampling frequency in Hz (as per EDF specification)
            physical_dimension: Physical dimension/unit of measurement (as per EDF specification)
            channel_type: BIDS channel type ('EEG', 'EMG', 'ECG', 'ACC', 'SEEG', ...).
                Required; validated against the modality vocabulary. There is no
                default (a missing type must be explicit, e.g. 'OTHER'/'MISC').
            modality: Coarse modality ('EEG', 'EMG', 'IEEG', 'MEG', 'BEH', 'MISC').
                If None, it is inferred from ``channel_type``.
            prefilter: Pre-filtering applied to the channel (keyword-only).
        """
        canonical_type = validate_channel_type(channel_type)
        canonical_modality = (
            infer_modality_from_channel_type(canonical_type)
            if modality is None
            else validate_modality(modality)
        )

        if self.signals is None:
            # Create DataFrame with time index
            time = np.arange(len(data)) / sample_frequency
            self.signals = pd.DataFrame(index=time)

        self.signals[label] = data
        self.channels[label] = {
            "sample_frequency": sample_frequency,
            "physical_dimension": physical_dimension,
            "prefilter": prefilter,
            "channel_type": canonical_type,
            "modality": canonical_modality,
        }

    def set_channel(
        self,
        label: str,
        *,
        channel_type: str | None = None,
        modality: str | None = None,
        physical_dimension: str | None = None,
        prefilter: str | None = None,
    ) -> None:
        """
        Update metadata of an existing channel (the supported relabel path).

        Args:
            label: Existing channel label.
            channel_type: New BIDS channel type (validated). When given without an
                explicit ``modality``, the modality is re-derived from it.
            modality: New coarse modality (validated).
            physical_dimension: New physical unit.
            prefilter: New prefilter string.

        Raises:
            KeyError: If ``label`` is not an existing channel.
            ValueError: If ``channel_type`` or ``modality`` is not in the
                modality vocabulary.
        """
        if label not in self.channels:
            raise KeyError(f"Channel not found: {label}")
        info = self.channels[label]
        if channel_type is not None:
            info["channel_type"] = validate_channel_type(channel_type)
            if modality is None:
                info["modality"] = infer_modality_from_channel_type(info["channel_type"])
        if modality is not None:
            info["modality"] = validate_modality(modality)
        if physical_dimension is not None:
            info["physical_dimension"] = physical_dimension
        if prefilter is not None:
            info["prefilter"] = prefilter

    def add_event(self, onset: float, duration: float, description: str) -> None:
        """
        Add an event/annotation to the recording.

        Args:
            onset: Event onset time in seconds.
            duration: Event duration in seconds.
            description: Event description string.
        """
        new_event = pd.DataFrame(
            [{"onset": float(onset), "duration": float(duration), "description": description}]
        )
        # Avoid concatenating onto the empty, object-dtype events frame, which
        # would coerce the numeric columns to object. Start from the typed
        # new_event when there are no existing events.
        if self.events is None or self.events.empty:
            self.events = new_event
        else:
            self.events = pd.concat([self.events, new_event], ignore_index=True)
        # Sort events by onset time for consistency
        self.events = self.events.sort_values(by="onset").reset_index(drop=True)

__init__()

Initialize an empty recording.

Source code in biosigio/core/emg.py

def __init__(self):
    """Initialize an empty recording."""
    self.signals = None
    self.metadata = {}
    self.channels = {}
    # Initialize events as an empty DataFrame with specified columns
    self.events = pd.DataFrame(columns=["onset", "duration", "description"])

add_channel(label, data, sample_frequency, physical_dimension, channel_type, *, modality=None, prefilter='n/a')

Add a new channel to the recording.

Args: label: Channel label or name (as per EDF specification) data: Channel data sample_frequency: Sampling frequency in Hz (as per EDF specification) physical_dimension: Physical dimension/unit of measurement (as per EDF specification) channel_type: BIDS channel type ('EEG', 'EMG', 'ECG', 'ACC', 'SEEG', ...). Required; validated against the modality vocabulary. There is no default (a missing type must be explicit, e.g. 'OTHER'/'MISC'). modality: Coarse modality ('EEG', 'EMG', 'IEEG', 'MEG', 'BEH', 'MISC'). If None, it is inferred from channel_type. prefilter: Pre-filtering applied to the channel (keyword-only).

Source code in biosigio/core/emg.py

def add_channel(
    self,
    label: str,
    data: np.ndarray,
    sample_frequency: float,
    physical_dimension: str,
    channel_type: str,
    *,
    modality: str | None = None,
    prefilter: str = "n/a",
) -> None:
    """
    Add a new channel to the recording.

    Args:
        label: Channel label or name (as per EDF specification)
        data: Channel data
        sample_frequency: Sampling frequency in Hz (as per EDF specification)
        physical_dimension: Physical dimension/unit of measurement (as per EDF specification)
        channel_type: BIDS channel type ('EEG', 'EMG', 'ECG', 'ACC', 'SEEG', ...).
            Required; validated against the modality vocabulary. There is no
            default (a missing type must be explicit, e.g. 'OTHER'/'MISC').
        modality: Coarse modality ('EEG', 'EMG', 'IEEG', 'MEG', 'BEH', 'MISC').
            If None, it is inferred from ``channel_type``.
        prefilter: Pre-filtering applied to the channel (keyword-only).
    """
    canonical_type = validate_channel_type(channel_type)
    canonical_modality = (
        infer_modality_from_channel_type(canonical_type)
        if modality is None
        else validate_modality(modality)
    )

    if self.signals is None:
        # Create DataFrame with time index
        time = np.arange(len(data)) / sample_frequency
        self.signals = pd.DataFrame(index=time)

    self.signals[label] = data
    self.channels[label] = {
        "sample_frequency": sample_frequency,
        "physical_dimension": physical_dimension,
        "prefilter": prefilter,
        "channel_type": canonical_type,
        "modality": canonical_modality,
    }

add_event(onset, duration, description)

Add an event/annotation to the recording.

Args: onset: Event onset time in seconds. duration: Event duration in seconds. description: Event description string.

Source code in biosigio/core/emg.py

def add_event(self, onset: float, duration: float, description: str) -> None:
    """
    Add an event/annotation to the recording.

    Args:
        onset: Event onset time in seconds.
        duration: Event duration in seconds.
        description: Event description string.
    """
    new_event = pd.DataFrame(
        [{"onset": float(onset), "duration": float(duration), "description": description}]
    )
    # Avoid concatenating onto the empty, object-dtype events frame, which
    # would coerce the numeric columns to object. Start from the typed
    # new_event when there are no existing events.
    if self.events is None or self.events.empty:
        self.events = new_event
    else:
        self.events = pd.concat([self.events, new_event], ignore_index=True)
    # Sort events by onset time for consistency
    self.events = self.events.sort_values(by="onset").reset_index(drop=True)

from_file(filepath, importer=None, force_csv=False, bids_channels='auto', mixed_rate='error', **kwargs) classmethod

The method to create a Recording object from file.

Args: filepath: Path to the input file importer: Name of the importer to use. Can be one of the following: - 'trigno': Delsys Trigno EMG system (CSV) - 'otb': OTB/OTB+ EMG system (OTB, OTB+) - 'eeglab': EEGLAB .set files (SET) - 'edf': EDF/EDF+/BDF/BDF+ format (EDF, BDF) - 'csv': Generic CSV (or TXT) files with columnar data - 'wfdb': Waveform Database (WFDB) - 'xdf': XDF format (multi-stream Lab Streaming Layer files) - 'meg': MEG via MNE (.fif, CTF .ds, KIT .con/.sqd/.kdf, 4D/BTi directory (no extension, detected by content); requires the 'meg' extra) - 'brainvision': BrainVision .vhdr via MNE (requires the 'meg' extra) - 'mef3': MEF3 iEEG via MNE (.mefd directory; requires the 'mef3' extra -- mne>=1.12 plus pymef, stricter than the 'meg' extra) - 'tabular': biosigIO Parquet/Arrow/Feather (requires the 'arrow' extra) - 'neo': proprietary electrophysiology formats via python-neo (Intan, Blackrock, Spike2, Plexon, Micromed, Neuralynx, ...; requires the 'neo' extra) - 'zarr': biosigIO Zarr serving store (requires the 'zarr' extra) If None, the importer will be inferred from the file extension. Automatic import is supported for CSV/TXT files. force_csv: If True and importer is 'csv', forces using the generic CSV importer even if the file appears to match a specialized format. bids_channels: Which BIDS _channels.tsv to apply over the importer's inferred per-channel type/units. 'auto' (default) looks for the sibling sidecar next to the file; 'off' (or None) disables the lookup; a path or a pandas DataFrame is used as given, for a recording whose sidecar is not adjacent to it (a filtered copy on scratch, say). Adopting a declared unit converts the samples into it, not just the label. stream_to_zarr takes the same argument with the same meanings, so the in-memory and streaming export paths agree on a recording (issue #127). mixed_rate: Policy for an EDF/BDF file whose signals carry differing per-channel sampling rates (ignored for every other format, which is single-rate). 'error' (default) raises -- biosigIO stores one uniform grid and will not fabricate a common one silently. 'resample' upsamples the slower channels to the fastest rate (a lossy derived view; each channel keeps its native rate as original_sample_frequency). **kwargs: Additional arguments passed to the importer. For XDF files, useful kwargs include: - stream_names: List of stream names to import - stream_types: List of stream types to import (e.g., ["EMG", "EXG"]) - stream_ids: List of stream IDs to import

Returns: Recording: New Recording object with loaded data

Source code in biosigio/core/emg.py

@classmethod
def from_file(
    cls,
    filepath: str,
    importer: ImporterName | None = None,
    force_csv: bool = False,
    bids_channels: "str | os.PathLike | pd.DataFrame | None" = "auto",
    mixed_rate: str = "error",
    **kwargs,
) -> "Recording":
    """
    The method to create a Recording object from file.

    Args:
        filepath: Path to the input file
        importer: Name of the importer to use. Can be one of the following:
            - 'trigno': Delsys Trigno EMG system (CSV)
            - 'otb': OTB/OTB+ EMG system (OTB, OTB+)
            - 'eeglab': EEGLAB .set files (SET)
            - 'edf': EDF/EDF+/BDF/BDF+ format (EDF, BDF)
            - 'csv': Generic CSV (or TXT) files with columnar data
            - 'wfdb': Waveform Database (WFDB)
            - 'xdf': XDF format (multi-stream Lab Streaming Layer files)
            - 'meg': MEG via MNE (.fif, CTF .ds, KIT .con/.sqd/.kdf, 4D/BTi
              directory (no extension, detected by content); requires the
              'meg' extra)
            - 'brainvision': BrainVision .vhdr via MNE (requires the 'meg' extra)
            - 'mef3': MEF3 iEEG via MNE (.mefd directory; requires the 'mef3'
              extra -- mne>=1.12 plus pymef, stricter than the 'meg' extra)
            - 'tabular': biosigIO Parquet/Arrow/Feather (requires the 'arrow' extra)
            - 'neo': proprietary electrophysiology formats via python-neo
              (Intan, Blackrock, Spike2, Plexon, Micromed, Neuralynx, ...;
              requires the 'neo' extra)
            - 'zarr': biosigIO Zarr serving store (requires the 'zarr' extra)
            If None, the importer will be inferred from the file extension.
            Automatic import is supported for CSV/TXT files.
        force_csv: If True and importer is 'csv', forces using the generic CSV
                  importer even if the file appears to match a specialized format.
        bids_channels: Which BIDS ``_channels.tsv`` to apply over the
                  importer's inferred per-channel type/units. 'auto'
                  (default) looks for the sibling sidecar next to the file;
                  'off' (or None) disables the lookup; a path or a pandas
                  DataFrame is used as given, for a recording whose sidecar
                  is not adjacent to it (a filtered copy on scratch, say).
                  Adopting a declared unit converts the samples into it, not
                  just the label. ``stream_to_zarr`` takes the same argument
                  with the same meanings, so the in-memory and streaming
                  export paths agree on a recording (issue #127).
        mixed_rate: Policy for an EDF/BDF file whose signals carry differing
                  per-channel sampling rates (ignored for every other format,
                  which is single-rate). 'error' (default) raises -- biosigIO
                  stores one uniform grid and will not fabricate a common one
                  silently. 'resample' upsamples the slower channels to the
                  fastest rate (a lossy derived view; each channel keeps its
                  native rate as ``original_sample_frequency``).
        **kwargs: Additional arguments passed to the importer.
            For XDF files, useful kwargs include:
            - stream_names: List of stream names to import
            - stream_types: List of stream types to import (e.g., ["EMG", "EXG"])
            - stream_ids: List of stream IDs to import

    Returns:
        Recording: New Recording object with loaded data
    """
    if importer is None:
        importer = cls._infer_importer(filepath)

    importers = {
        "trigno": "TrignoImporter",  # CSV with Delsys Trigno Headers
        "otb": "OTBImporter",  # OTB/OTB+ EMG system data
        "edf": "EDFImporter",  # EDF/EDF+/BDF format
        "eeglab": "EEGLABImporter",  # EEGLAB .set files
        "csv": "CSVImporter",  # Generic CSV/Text files
        "wfdb": "WFDBImporter",  # Waveform Database format
        "xdf": "XDFImporter",  # XDF multi-stream format
        "meg": "MEGImporter",  # MEG via MNE (.fif, CTF .ds, KIT .con/.sqd/.kdf, 4D/BTi)
        "brainvision": "BrainVisionImporter",  # BrainVision via MNE (.vhdr)
        "mef3": "MEF3Importer",  # MEF3 iEEG via MNE (.mefd; requires the 'mef3' extra)
        "tabular": "TabularImporter",  # biosigIO Parquet / Arrow / Feather
        "neo": "NeoImporter",  # proprietary ephys via python-neo
        "zarr": "ZarrImporter",  # biosigIO Zarr serving store
    }

    if importer not in importers:
        raise ValueError(
            f"Unsupported importer: {importer}. "
            f"Available importers: {list(importers.keys())}\n"
            "- trigno: Delsys Trigno EMG system\n"
            "- otb: OTB/OTB+ EMG system\n"
            "- edf: EDF/EDF+/BDF format\n"
            "- eeglab: EEGLAB .set files\n"
            "- csv: Generic CSV/Text files\n"
            "- wfdb: Waveform Database\n"
            "- xdf: XDF multi-stream format\n"
            "- meg: MEG via MNE (.fif, CTF .ds, KIT .con/.sqd/.kdf, 4D/BTi directory)\n"
            "- brainvision: BrainVision via MNE (.vhdr)\n"
            "- mef3: MEF3 iEEG via MNE (.mefd directory; requires the 'mef3' extra)\n"
            "- tabular: biosigIO Parquet/Arrow/Feather (.parquet, .feather, .arrow)\n"
            "- neo: proprietary electrophysiology formats via python-neo "
            "(Intan, Blackrock, Spike2, Plexon, Micromed, Neuralynx, ...)\n"
            "- zarr: biosigIO Zarr serving store (.zarr)"
        )

    # If using CSV importer and force_csv is set, pass it as force_generic
    if importer == "csv":
        kwargs["force_generic"] = force_csv

    # mixed_rate is only meaningful for EDF/BDF, the one format whose signals may
    # carry differing per-channel sampling rates; forward it there and nowhere
    # else (so the default "error" never reaches an importer that can't use it).
    if importer == "edf":
        kwargs["mixed_rate"] = mixed_rate

    # Import the appropriate importer class
    importer_module = __import__(
        f"biosigio.importers.{importer}", globals(), locals(), [importers[importer]]
    )
    importer_class = getattr(importer_module, importers[importer])

    # Create importer instance and load data
    rec = importer_class().load(filepath, **kwargs)

    # Record provenance: which format this recording came from. setdefault so a
    # re-imported serialization file (tabular/zarr) keeps the ORIGINAL
    # source_format restored from its metadata rather than being relabeled.
    rec.metadata.setdefault("source_format", importer)

    # In a BIDS layout, the sibling _channels.tsv is the authoritative source
    # of per-channel type/units; apply it over the importer's header/label
    # guesses unless explicitly disabled with bids_channels="off". The
    # argument is interpreted by bids.resolve_channels_tsv, shared with
    # stream_to_zarr so the two export paths cannot read it differently.
    from ..bids import apply_channels_tsv, resolve_channels_tsv

    channels_tsv = resolve_channels_tsv(filepath, bids_channels)
    if channels_tsv is not None:
        apply_channels_tsv(rec, channels_tsv)

    return rec

get_channel_types()

Get list of unique channel types in the data.

Returns: List of channel types (e.g., ['EMG', 'ACC', 'GYRO'])

Source code in biosigio/core/emg.py

def get_channel_types(self) -> list[str]:
    """
    Get list of unique channel types in the data.

    Returns:
        List of channel types (e.g., ['EMG', 'ACC', 'GYRO'])
    """
    return list({info["channel_type"] for info in self.channels.values()})

get_channels_by_modality(modality)

Get the channels belonging to a given modality.

Args: modality: Modality to filter by ('EEG', 'EMG', 'IEEG', 'MEG', 'BEH', 'MISC').

Returns: List of channel names of the specified modality.

Source code in biosigio/core/emg.py

def get_channels_by_modality(self, modality: str) -> list[str]:
    """
    Get the channels belonging to a given modality.

    Args:
        modality: Modality to filter by ('EEG', 'EMG', 'IEEG', 'MEG', 'BEH', 'MISC').

    Returns:
        List of channel names of the specified modality.
    """
    canonical_modality = validate_modality(modality)
    return [
        ch for ch, info in self.channels.items() if info.get("modality") == canonical_modality
    ]

get_channels_by_type(channel_type)

Get list of channels of a specific type.

Args: channel_type: Type of channels to get ('EMG', 'ACC', 'GYRO', etc.)

Returns: List of channel names of the specified type

Source code in biosigio/core/emg.py

def get_channels_by_type(self, channel_type: str) -> list[str]:
    """
    Get list of channels of a specific type.

    Args:
        channel_type: Type of channels to get ('EMG', 'ACC', 'GYRO', etc.)

    Returns:
        List of channel names of the specified type
    """
    return [ch for ch, info in self.channels.items() if info["channel_type"] == channel_type]

get_duration()

Total recording duration in seconds (n_samples / sampling_frequency).

Computed from the time index spacing, so it is the full window length (one sample period longer than the last sample's timestamp). Returns 0.0 when fewer than two samples are loaded (a single sample has no inferable sample period).

Source code in biosigio/core/emg.py

def get_duration(self) -> float:
    """Total recording duration in seconds (n_samples / sampling_frequency).

    Computed from the time index spacing, so it is the full window length
    (one sample period longer than the last sample's timestamp). Returns 0.0
    when fewer than two samples are loaded (a single sample has no inferable
    sample period).
    """
    if self.signals is None or len(self.signals) < 2:
        return 0.0
    index = self.signals.index
    sample_period = float(index[1] - index[0])
    return len(index) * sample_period

get_metadata(key)

Get metadata value.

Args: key: Metadata key

Returns: Value associated with the key

Source code in biosigio/core/emg.py

def get_metadata(self, key: str) -> Any:
    """
    Get metadata value.

    Args:
        key: Metadata key

    Returns:
        Value associated with the key
    """
    return self.metadata.get(key)

get_modalities()

Get the list of unique modalities present in the data.

Returns: List of modalities (e.g., ['EEG', 'EMG', 'MISC']).

Source code in biosigio/core/emg.py

def get_modalities(self) -> list[str]:
    """
    Get the list of unique modalities present in the data.

    Returns:
        List of modalities (e.g., ['EEG', 'EMG', 'MISC']).
    """
    return list(
        {info.get("modality") for info in self.channels.values() if info.get("modality")}
    )

get_n_channels()

Number of channels in the recording.

Source code in biosigio/core/emg.py

def get_n_channels(self) -> int:
    """Number of channels in the recording."""
    return len(self.channels)

get_n_samples()

Number of time samples per channel (0 if no signals are loaded).

Source code in biosigio/core/emg.py

def get_n_samples(self) -> int:
    """Number of time samples per channel (0 if no signals are loaded)."""
    return 0 if self.signals is None else len(self.signals)

get_sampling_frequency()

Sampling frequency in Hz, when all channels share a single rate.

Raises: ValueError: if no channels are loaded, or channels have differing sampling frequencies; for a mixed-rate recording read channels[ch]["sample_frequency"] per channel instead.

Source code in biosigio/core/emg.py

def get_sampling_frequency(self) -> float:
    """Sampling frequency in Hz, when all channels share a single rate.

    Raises:
        ValueError: if no channels are loaded, or channels have differing
            sampling frequencies; for a mixed-rate recording read
            ``channels[ch]["sample_frequency"]`` per channel instead.
    """
    if not self.channels:
        raise ValueError("No channels loaded")
    rates = {info["sample_frequency"] for info in self.channels.values()}
    if len(rates) > 1:
        raise ValueError(
            "Channels have differing sampling frequencies; read "
            "channels[ch]['sample_frequency'] per channel instead."
        )
    return float(next(iter(rates)))

has_metadata(key)

Return True if key is present in the recording metadata.

Source code in biosigio/core/emg.py

def has_metadata(self, key: str) -> bool:
    """Return True if ``key`` is present in the recording metadata."""
    return key in self.metadata

plot_signals(channels=None, time_range=None, offset_scale=0.8, uniform_scale=True, detrend=False, grid=True, title=None, show=True, plt_module=None)

Plot signals in a single plot with vertical offsets.

Args: channels: List of channels to plot. If None, plot all channels. time_range: Tuple of (start_time, end_time) to plot. If None, plot all data. offset_scale: Portion of allocated space each signal can use (0.0 to 1.0). uniform_scale: Whether to use the same scale for all signals. detrend: Whether to remove mean from signals before plotting. grid: Whether to show grid lines. title: Optional title for the figure. show: Whether to display the plot. plt_module: Matplotlib pyplot module to use.

Source code in biosigio/core/emg.py

def plot_signals(
    self,
    channels=None,
    time_range=None,
    offset_scale=0.8,
    uniform_scale=True,
    detrend=False,
    grid=True,
    title=None,
    show=True,
    plt_module=None,
):
    """
    Plot signals in a single plot with vertical offsets.

    Args:
        channels: List of channels to plot. If None, plot all channels.
        time_range: Tuple of (start_time, end_time) to plot. If None, plot all data.
        offset_scale: Portion of allocated space each signal can use (0.0 to 1.0).
        uniform_scale: Whether to use the same scale for all signals.
        detrend: Whether to remove mean from signals before plotting.
        grid: Whether to show grid lines.
        title: Optional title for the figure.
        show: Whether to display the plot.
        plt_module: Matplotlib pyplot module to use.
    """
    # Delegate to the static plotting function in visualization module
    static_plot_signals(
        rec_object=self,
        channels=channels,
        time_range=time_range,
        offset_scale=offset_scale,
        uniform_scale=uniform_scale,
        detrend=detrend,
        grid=grid,
        title=title,
        show=show,
        plt_module=plt_module,
    )

resample(target_rate)

Return a NEW, anti-aliased down-sampled copy of this recording.

Low-resolution demos need a smaller, lighter recording; this rebuilds the uniform signal grid at target_rate using a polyphase resampler (scipy.signal.resample_poly), which applies a Kaiser-windowed sinc anti-alias FIR before decimation. A naive stride-decimation would fold energy above the new Nyquist back into the band (aliasing); resample_poly removes that energy first, so no aliasing occurs.

Non-destructive: self is left untouched and a new Recording is returned, mirroring select_channels's copy semantics.

Resampling factors come from the integer source/target rates: g = gcd(int(src), int(target)); up = int(target)//g; down = int(src)//g and resample_poly(x, up, down) runs once, vectorized over all channels along axis=0.

Args: target_rate: Desired sampling rate in Hz. Must be <= the source rate (this is a DOWN-sampling helper). A target equal to the source returns an unchanged copy; a target above it raises ValueError rather than silently up-sampling (up-sampling cannot recover detail and is out of scope for the low-res pipeline).

Returns: Recording: A new Recording with the resampled signals, each channel's sample_frequency set to the achieved rate (source * up / down, which equals target_rate for integer rates), and channel/recording metadata and events preserved. Events are unchanged because their onsets/durations are in SECONDS, which stay valid under any rate change (only the per-sample grid shrinks, not wall-clock time).

Raises: ValueError: If no signals are loaded, if channels do not share a single sample_frequency (biosigio stores one uniform grid; mixed-rate resampling is out of scope), or if target_rate exceeds the source rate.

Source code in biosigio/core/emg.py

def resample(self, target_rate: float) -> "Recording":
    """Return a NEW, anti-aliased down-sampled copy of this recording.

    Low-resolution demos need a smaller, lighter recording; this rebuilds the
    uniform signal grid at ``target_rate`` using a polyphase resampler
    (``scipy.signal.resample_poly``), which applies a Kaiser-windowed sinc
    anti-alias FIR before decimation. A naive stride-decimation would fold
    energy above the new Nyquist back into the band (aliasing); resample_poly
    removes that energy first, so no aliasing occurs.

    Non-destructive: ``self`` is left untouched and a new Recording is returned,
    mirroring ``select_channels``'s copy semantics.

    Resampling factors come from the integer source/target rates:
    ``g = gcd(int(src), int(target)); up = int(target)//g; down = int(src)//g``
    and ``resample_poly(x, up, down)`` runs once, vectorized over all channels
    along ``axis=0``.

    Args:
        target_rate: Desired sampling rate in Hz. Must be <= the source rate
            (this is a DOWN-sampling helper). A target equal to the source
            returns an unchanged copy; a target above it raises ``ValueError``
            rather than silently up-sampling (up-sampling cannot recover
            detail and is out of scope for the low-res pipeline).

    Returns:
        Recording: A new Recording with the resampled signals, each channel's
            ``sample_frequency`` set to the achieved rate (source * up / down,
            which equals ``target_rate`` for integer rates), and channel/recording
            metadata and events preserved. Events are unchanged because their
            onsets/durations are in SECONDS, which stay valid under any rate
            change (only the per-sample grid shrinks, not wall-clock time).

    Raises:
        ValueError: If no signals are loaded, if channels do not share a single
            ``sample_frequency`` (biosigio stores one uniform grid; mixed-rate
            resampling is out of scope), or if ``target_rate`` exceeds the
            source rate.
    """
    from math import gcd

    from scipy.signal import resample_poly

    if self.signals is None:
        raise ValueError("No signals loaded")

    if target_rate <= 0:
        raise ValueError(f"target_rate must be positive, got {target_rate}")

    # biosigio stores all channels on one uniform-length grid; a per-channel rate
    # mix is out of scope here, matching the exporter's single-rate guard.
    distinct_rates = {info["sample_frequency"] for info in self.channels.values()}
    if len(distinct_rates) > 1:
        raise ValueError(
            "Resampling requires a single sampling rate across all channels, but "
            f"multiple were found: {sorted(distinct_rates)} Hz. biosigio stores one "
            "uniform grid; resample each rate group separately."
        )

    source_rate = float(next(iter(distinct_rates)))

    # Down-sampling only: refuse to up-sample/alias; an equal rate is a no-op
    # copy so callers can resample unconditionally without special-casing.
    if target_rate > source_rate:
        raise ValueError(
            f"target_rate {target_rate} Hz exceeds source rate {source_rate} Hz; "
            "resample() only down-samples (low-res). Up-sampling is out of scope."
        )

    new_rec = Recording()
    new_rec.channels = {ch: info.copy() for ch, info in self.channels.items()}
    new_rec.metadata = self.metadata.copy()
    # Onsets/durations are in SECONDS, so they remain valid after the grid
    # changes; copy them through unchanged.
    new_rec.events = self.events.copy() if self.events is not None else self.events

    if target_rate == source_rate:
        # No grid change: copy signals through untouched (fresh RangeIndex for
        # consistency with the resampled path).
        new_rec.signals = self.signals.copy().reset_index(drop=True)
        return new_rec

    # Rational resampling factors from the integer rates.
    src_i = int(round(source_rate))
    tgt_i = int(round(target_rate))
    g = gcd(src_i, tgt_i)
    up = tgt_i // g
    down = src_i // g

    # The achieved rate is exactly source * up / down. Store THAT, not the
    # requested float, so the metadata can never disagree with the data (a
    # non-integer or odd target snaps to the nearest achievable rational rate;
    # warn so the caller knows). This avoids silently writing e.g. 99.5 Hz
    # onto a grid that resample_poly actually produced at 100 Hz.
    actual_rate = source_rate * up / down
    if abs(actual_rate - target_rate) > 1e-9:
        logging.warning(
            "Requested resample to %g Hz; nearest achievable rational rate is "
            "%g Hz, which is what is stored on the channels.",
            target_rate,
            actual_rate,
        )

    columns = list(self.signals.columns)
    data = self.signals.to_numpy(dtype=float)
    # resample_poly over axis=0 resamples every channel column at once with the
    # shared anti-alias FIR.
    resampled = resample_poly(data, up, down, axis=0)

    new_rec.signals = pd.DataFrame(resampled, columns=columns)
    new_rec.signals.index = pd.RangeIndex(len(new_rec.signals))
    for info in new_rec.channels.values():
        info["sample_frequency"] = actual_rate

    return new_rec

select_channels(channels=None, channel_type=None, inplace=False, *, modality=None)

Select specific channels from the data and return a new Recording object.

Args: channels: Channel name or list of channel names to select. If None and channel_type is specified, selects all channels of that type. channel_type: Type of channels to select ('EMG', 'ACC', 'GYRO', etc.). If specified with channels, filters the selection to only channels of this type.

Returns: Recording: A new Recording object containing only the selected channels

Examples: # Select specific channels new_rec = rec.select_channels(['EMG1', 'ACC1'])

# Select all EMG channels
emg_only = rec.select_channels(channel_type='EMG')

# Select specific EMG channels only, this example does not select ACC channels
emg_subset = rec.select_channels(['EMG1', 'ACC1'], channel_type='EMG')

Source code in biosigio/core/emg.py

def select_channels(
    self,
    channels: str | list[str] | None = None,
    channel_type: str | None = None,
    inplace: bool = False,
    *,
    modality: str | None = None,
) -> "Recording":
    """
    Select specific channels from the data and return a new Recording object.

    Args:
        channels: Channel name or list of channel names to select. If None and
                channel_type is specified, selects all channels of that type.
        channel_type: Type of channels to select ('EMG', 'ACC', 'GYRO', etc.).
                    If specified with channels, filters the selection to only
                    channels of this type.

    Returns:
        Recording: A new Recording object containing only the selected channels

    Examples:
        # Select specific channels
        new_rec = rec.select_channels(['EMG1', 'ACC1'])

        # Select all EMG channels
        emg_only = rec.select_channels(channel_type='EMG')

        # Select specific EMG channels only, this example does not select ACC channels
        emg_subset = rec.select_channels(['EMG1', 'ACC1'], channel_type='EMG')
    """
    if self.signals is None:
        raise ValueError("No signals loaded")

    if channels is None and channel_type is None and modality is None:
        raise ValueError("Specify at least one of: channels, channel_type, or modality.")

    # If type/modality specified but no channels, select all matching channels
    if channels is None and channel_type is not None:
        channels = self.get_channels_by_type(channel_type)
        if not channels:
            raise ValueError(f"No channels found of type: {channel_type}")
    elif channels is None and modality is not None:
        channels = self.get_channels_by_modality(modality)
        if not channels:
            raise ValueError(f"No channels found of modality: {modality}")
    elif isinstance(channels, str):
        channels = [channels]

    if channels is None:
        raise ValueError("Specify at least one of: channels, channel_type, or modality.")

    # Validate channels exist
    if not all(ch in self.signals.columns for ch in channels):
        missing = [ch for ch in channels if ch not in self.signals.columns]
        raise ValueError(f"Channels not found: {missing}")

    # Filter by type if specified
    if channel_type is not None:
        channels = [ch for ch in channels if self.channels[ch]["channel_type"] == channel_type]
        if not channels:
            raise ValueError(f"None of the selected channels are of type: {channel_type}")

    # Filter by modality if specified
    if modality is not None:
        canonical_modality = validate_modality(modality)
        channels = [
            ch for ch in channels if self.channels[ch].get("modality") == canonical_modality
        ]
        if not channels:
            raise ValueError(f"None of the selected channels are of modality: {modality}")

    # Create new Recording object
    new_rec = Recording()

    # Copy selected signals and channels
    new_rec.signals = self.signals[channels].copy()
    new_rec.channels = {ch: self.channels[ch].copy() for ch in channels}

    # Copy metadata
    new_rec.metadata = self.metadata.copy()

    if not inplace:
        return new_rec
    else:
        self.signals = new_rec.signals
        self.channels = new_rec.channels
        self.metadata = new_rec.metadata
        return self

set_channel(label, *, channel_type=None, modality=None, physical_dimension=None, prefilter=None)

Update metadata of an existing channel (the supported relabel path).

Args: label: Existing channel label. channel_type: New BIDS channel type (validated). When given without an explicit modality, the modality is re-derived from it. modality: New coarse modality (validated). physical_dimension: New physical unit. prefilter: New prefilter string.

Raises: KeyError: If label is not an existing channel. ValueError: If channel_type or modality is not in the modality vocabulary.

Source code in biosigio/core/emg.py

def set_channel(
    self,
    label: str,
    *,
    channel_type: str | None = None,
    modality: str | None = None,
    physical_dimension: str | None = None,
    prefilter: str | None = None,
) -> None:
    """
    Update metadata of an existing channel (the supported relabel path).

    Args:
        label: Existing channel label.
        channel_type: New BIDS channel type (validated). When given without an
            explicit ``modality``, the modality is re-derived from it.
        modality: New coarse modality (validated).
        physical_dimension: New physical unit.
        prefilter: New prefilter string.

    Raises:
        KeyError: If ``label`` is not an existing channel.
        ValueError: If ``channel_type`` or ``modality`` is not in the
            modality vocabulary.
    """
    if label not in self.channels:
        raise KeyError(f"Channel not found: {label}")
    info = self.channels[label]
    if channel_type is not None:
        info["channel_type"] = validate_channel_type(channel_type)
        if modality is None:
            info["modality"] = infer_modality_from_channel_type(info["channel_type"])
    if modality is not None:
        info["modality"] = validate_modality(modality)
    if physical_dimension is not None:
        info["physical_dimension"] = physical_dimension
    if prefilter is not None:
        info["prefilter"] = prefilter

set_metadata(key, value)

Set metadata value.

Args: key: Metadata key value: Metadata value

Source code in biosigio/core/emg.py

def set_metadata(self, key: str, value: Any) -> None:
    """
    Set metadata value.

    Args:
        key: Metadata key
        value: Metadata value
    """
    self.metadata[key] = value

to_arrow(filepath)

Export to a biosigIO Arrow/Feather file (fast zero-copy IPC).

Same self-describing schema as :meth:to_parquet; round-trips via Recording.from_file. Requires the arrow extra (pyarrow).

Args: filepath: Output .feather / .arrow path.

Returns: str: The written file path.

Source code in biosigio/core/emg.py

def to_arrow(self, filepath: str) -> str:
    """Export to a biosigIO Arrow/Feather file (fast zero-copy IPC).

    Same self-describing schema as :meth:`to_parquet`; round-trips via
    ``Recording.from_file``. Requires the ``arrow`` extra (pyarrow).

    Args:
        filepath: Output ``.feather`` / ``.arrow`` path.

    Returns:
        str: The written file path.
    """
    from ..exporters.tabular import TabularExporter

    return TabularExporter.to_arrow(self, filepath)

to_edf(filepath, method='both', fft_noise_range=None, svd_rank=None, precision_threshold=0.01, format='auto', bypass_analysis=None, verify=False, verify_tolerance=1e-06, verify_channel_map=None, verify_plot=False, events_df=None, create_channels_tsv=True, clip_outliers='auto', **kwargs)

Export the recording to EDF/BDF format, optionally including events.

Args: filepath: Path to save the EDF/BDF file method: Method for signal analysis ('svd', 'fft', or 'both') 'svd': Uses Singular Value Decomposition for noise floor estimation 'fft': Uses Fast Fourier Transform for noise floor estimation 'both': Uses both methods and takes the minimum noise floor (default) fft_noise_range: Optional tuple (min_freq, max_freq) specifying frequency range for noise in FFT method svd_rank: Optional manual rank cutoff for signal/noise separation in SVD method precision_threshold: Maximum acceptable precision loss percentage (default: 0.01%) format: Format to use ('auto', 'edf', or 'bdf'). Default is 'auto'. If 'edf' or 'bdf' is specified, that format will be used directly. If 'auto', the format (EDF/16-bit or BDF/24-bit) is chosen based on signal analysis to minimize precision loss while preferring EDF if sufficient. bypass_analysis: If True, skip signal analysis step when format is explicitly set to 'edf' or 'bdf'. If None (default), analysis is skipped automatically when format is forced. Set to False to force analysis even with a specified format. Ignored if format='auto'. verify: If True, reload the exported file and compare signals with the original to check for data integrity loss. Results are printed. (default: False) verify_tolerance: Absolute tolerance used when comparing signals during verification. (default: 1e-6) verify_channel_map: Optional dictionary mapping original channel names (keys) to reloaded channel names (values) for verification. Used if verify is True and channel names might differ. verify_plot: If True and verify is True, plots a comparison of original vs reloaded signals. events_df: Optional DataFrame with events ('onset', 'duration', 'description'). If None, uses self.events. (This provides flexibility) create_channels_tsv: If True, create a BIDS-compliant channels.tsv file (default: True) clip_outliers: Singularity handling for the per-channel physical window. 'auto' (default) keeps the full range losslessly but clips rare extreme outliers to a robust window only when keeping them would crater the bulk signal's resolution at the chosen format (with a warning); True always clips to the robust window; False never clips. See EDFExporter.export for the advanced outlier_sigmas / min_effective_bits knobs. **kwargs: Additional arguments for the EDF exporter

Returns: Union[str, None]: If verify is True, returns a string with verification results. Otherwise, returns None.

Raises: ValueError: If no signals are loaded

Source code in biosigio/core/emg.py

def to_edf(
    self,
    filepath: str,
    method: str = "both",
    fft_noise_range: tuple | None = None,
    svd_rank: int | None = None,
    precision_threshold: float = 0.01,
    format: Literal["auto", "edf", "bdf"] = "auto",
    bypass_analysis: bool | None = None,
    verify: bool = False,
    verify_tolerance: float = 1e-6,
    verify_channel_map: dict[str, str] | None = None,
    verify_plot: bool = False,
    events_df: pd.DataFrame | None = None,
    create_channels_tsv: bool = True,
    clip_outliers: bool | str = "auto",
    **kwargs,
) -> dict | None:
    """
    Export the recording to EDF/BDF format, optionally including events.

    Args:
        filepath: Path to save the EDF/BDF file
        method: Method for signal analysis ('svd', 'fft', or 'both')
            'svd': Uses Singular Value Decomposition for noise floor estimation
            'fft': Uses Fast Fourier Transform for noise floor estimation
            'both': Uses both methods and takes the minimum noise floor (default)
        fft_noise_range: Optional tuple (min_freq, max_freq) specifying frequency range for noise in FFT method
        svd_rank: Optional manual rank cutoff for signal/noise separation in SVD method
        precision_threshold: Maximum acceptable precision loss percentage (default: 0.01%)
        format: Format to use ('auto', 'edf', or 'bdf'). Default is 'auto'.
                If 'edf' or 'bdf' is specified, that format will be used directly.
                If 'auto', the format (EDF/16-bit or BDF/24-bit) is chosen based
                on signal analysis to minimize precision loss while preferring EDF
                if sufficient.
        bypass_analysis: If True, skip signal analysis step when format is explicitly
                         set to 'edf' or 'bdf'. If None (default), analysis is skipped
                         automatically when format is forced. Set to False to force
                         analysis even with a specified format. Ignored if format='auto'.
        verify: If True, reload the exported file and compare signals with the original
                to check for data integrity loss. Results are printed. (default: False)
        verify_tolerance: Absolute tolerance used when comparing signals during verification. (default: 1e-6)
        verify_channel_map: Optional dictionary mapping original channel names (keys)
                            to reloaded channel names (values) for verification.
                            Used if `verify` is True and channel names might differ.
        verify_plot: If True and verify is True, plots a comparison of original vs reloaded signals.
        events_df: Optional DataFrame with events ('onset', 'duration', 'description').
                  If None, uses self.events. (This provides flexibility)
        create_channels_tsv: If True, create a BIDS-compliant channels.tsv file (default: True)
        clip_outliers: Singularity handling for the per-channel physical window.
            'auto' (default) keeps the full range losslessly but clips rare extreme
            outliers to a robust window only when keeping them would crater the bulk
            signal's resolution at the chosen format (with a warning); True always
            clips to the robust window; False never clips. See EDFExporter.export for
            the advanced ``outlier_sigmas`` / ``min_effective_bits`` knobs.
        **kwargs: Additional arguments for the EDF exporter

    Returns:
        Union[str, None]: If verify is True, returns a string with verification results.
                         Otherwise, returns None.

    Raises:
        ValueError: If no signals are loaded
    """
    from ..exporters.edf import EDFExporter  # Local import

    if self.signals is None:
        raise ValueError("No signals loaded")

    # --- Determine if analysis should be bypassed ---
    final_bypass_analysis = False
    if format.lower() == "auto":
        if bypass_analysis is True:
            logging.warning(
                "bypass_analysis=True ignored because format='auto'. Analysis is required."
            )
        # Analysis is always needed for 'auto' format
        final_bypass_analysis = False
    elif format.lower() in ["edf", "bdf"]:
        if bypass_analysis is None:
            # Default behaviour: skip analysis if format is forced
            final_bypass_analysis = True
            msg = (
                f"Format forced to '{format}'. Skipping signal analysis for faster export. "
                "Set bypass_analysis=False to force analysis."
            )
            logging.log(logging.CRITICAL, msg)
        elif bypass_analysis is True:
            final_bypass_analysis = True
            logging.log(logging.CRITICAL, "bypass_analysis=True set. Skipping signal analysis.")
        else:  # bypass_analysis is False
            final_bypass_analysis = False
            logging.info(
                f"Format forced to '{format}' but bypass_analysis=False. Performing signal analysis."
            )
    else:
        # Should not happen if Literal type hint works, but good practice
        logging.warning(
            f"Unknown format '{format}'. Defaulting to 'auto' behavior (analysis enabled)."
        )
        format = "auto"
        final_bypass_analysis = False

    # Determine which events DataFrame to use
    if events_df is None:
        events_to_export = self.events
    else:
        events_to_export = events_df

    # Combine parameters
    all_params: dict[str, Any] = {
        "precision_threshold": precision_threshold,
        "method": method,
        "fft_noise_range": fft_noise_range,
        "svd_rank": svd_rank,
        "format": format,
        "bypass_analysis": final_bypass_analysis,
        "events_df": events_to_export,  # Pass the events dataframe
        "create_channels_tsv": create_channels_tsv,
        "clip_outliers": clip_outliers,
        **kwargs,
    }

    EDFExporter.export(self, filepath, **all_params)

    verification_report_dict = None
    if verify:
        logging.info(f"Verification requested. Reloading exported file: {filepath}")
        try:
            # Reload the exported file
            reloaded_rec = Recording.from_file(filepath, importer="edf")

            logging.info("Comparing original signals with reloaded signals...")
            # Compare signals using the imported function
            verification_results = compare_signals(
                self, reloaded_rec, tolerance=verify_tolerance, channel_map=verify_channel_map
            )

            # Generate and log report using the imported function
            report_verification_results(verification_results, verify_tolerance)
            verification_report_dict = verification_results

            # Plot comparison using imported function if requested
            summary = verification_results.get("channel_summary", {})
            comparison_mode = summary.get("comparison_mode", "unknown")
            compared_count = sum(1 for k in verification_results if k != "channel_summary")

            if verify_plot and compared_count > 0 and comparison_mode != "failed":
                plot_comparison(self, reloaded_rec, channel_map=verify_channel_map)
            elif verify_plot:
                logging.warning(
                    "Skipping verification plot: No channels were successfully compared."
                )

        except Exception as e:
            logging.error(f"Verification failed during reload or comparison: {e}")
            verification_report_dict = {
                "error": str(e),
                "channel_summary": {"comparison_mode": "failed"},
            }

    return verification_report_dict

to_parquet(filepath)

Export to a self-describing biosigIO Parquet file.

Signals are stored as a columnar table (channels = columns, time index preserved); channels/events/metadata travel in the file's schema metadata, so Recording.from_file round-trips it losslessly. Great for analytics (DuckDB/Polars/pandas/Spark). Requires the arrow extra (pyarrow).

Args: filepath: Output .parquet path.

Returns: str: The written file path.

Source code in biosigio/core/emg.py

def to_parquet(self, filepath: str) -> str:
    """Export to a self-describing biosigIO Parquet file.

    Signals are stored as a columnar table (channels = columns, time index
    preserved); channels/events/metadata travel in the file's schema metadata,
    so ``Recording.from_file`` round-trips it losslessly. Great for analytics
    (DuckDB/Polars/pandas/Spark). Requires the ``arrow`` extra (pyarrow).

    Args:
        filepath: Output ``.parquet`` path.

    Returns:
        str: The written file path.
    """
    from ..exporters.tabular import TabularExporter

    return TabularExporter.to_parquet(self, filepath)

to_zarr(filepath, **kwargs)

Export to a sharded Zarr v3 serving store with a min/max view pyramid.

Writes one cloud-native store that serves viewing, inference, and training from a single conversion: level 0 of each (modality, rate) group is the anti-aliased, per-modality-resampled inference signal, with a min/max render pyramid above it (flagged not-for-inference). A derived serving copy, not the archival source (BIDS/EDF stay authoritative). Requires the zarr extra (zarr v3). See :class:~biosigio.exporters.zarr.ZarrExporter for the tuning knobs (modality_rates, dtype, level-0 chunk/shard sizing, view_chunk_columns for the view pyramid, ...).

Args: filepath: Output store path (.zarr appended if missing). **kwargs: Forwarded to :meth:ZarrExporter.export.

Returns: str: The written store path.

Source code in biosigio/core/emg.py

def to_zarr(self, filepath: str, **kwargs) -> str:
    """Export to a sharded Zarr v3 serving store with a min/max view pyramid.

    Writes one cloud-native store that serves viewing, inference, and training
    from a single conversion: ``level 0`` of each ``(modality, rate)`` group is
    the anti-aliased, per-modality-resampled inference signal, with a min/max
    render pyramid above it (flagged not-for-inference). A derived serving copy,
    not the archival source (BIDS/EDF stay authoritative). Requires the ``zarr``
    extra (zarr v3). See :class:`~biosigio.exporters.zarr.ZarrExporter` for the
    tuning knobs (``modality_rates``, ``dtype``, level-0 chunk/shard sizing,
    ``view_chunk_columns`` for the view pyramid, ...).

    Args:
        filepath: Output store path (``.zarr`` appended if missing).
        **kwargs: Forwarded to :meth:`ZarrExporter.export`.

    Returns:
        str: The written store path.
    """
    from ..exporters.zarr import ZarrExporter

    # The empty-signal guard lives once, in ZarrExporter.export ("No signals
    # loaded"), matching the tabular path; no duplicate guard here.
    return ZarrExporter.export(self, filepath, **kwargs)

_UnitDecision

Bases: NamedTuple

What one sidecar units cell does to one channel.

A decision is data, not an edit: it says what the channel's unit label becomes, what its samples must be multiplied by to still mean that label, and what its recorded bids_unit should be afterwards. That is what lets the in-memory path (which has the samples in a DataFrame and rescales them now) and the streaming path (which will see them window by window, later) share one decision table instead of two implementations of the same rules.

Attributes: outcome: _UNCHANGED, _CONVERTED, _RELABELLED or _KEPT; the key this channel contributes to the channels_tsv_units report. unit: The unit label the channel ends up with. Only meaningful to write when the outcome adopted the sidecar's unit (_CONVERTED / _RELABELLED); otherwise it is the unit the channel already had. factor: The multiplier the samples need, always exactly 1.0 unless the outcome is _CONVERTED. bids_unit: What the channel's bids_unit key should hold afterwards -- the sidecar's declared unit when it was recorded rather than adopted, or None meaning "there is no unresolved conflict, drop it".

Source code in biosigio/bids.py

class _UnitDecision(NamedTuple):
    """What one sidecar ``units`` cell does to one channel.

    A decision is data, not an edit: it says what the channel's unit label
    becomes, what its samples must be multiplied by to still mean that label,
    and what its recorded ``bids_unit`` should be afterwards. That is what lets
    the in-memory path (which has the samples in a DataFrame and rescales them
    now) and the streaming path (which will see them window by window, later)
    share one decision table instead of two implementations of the same rules.

    Attributes:
        outcome: ``_UNCHANGED``, ``_CONVERTED``, ``_RELABELLED`` or ``_KEPT``;
            the key this channel contributes to the ``channels_tsv_units`` report.
        unit: The unit label the channel ends up with. Only meaningful to write
            when the outcome adopted the sidecar's unit (``_CONVERTED`` /
            ``_RELABELLED``); otherwise it is the unit the channel already had.
        factor: The multiplier the samples need, always exactly ``1.0`` unless
            the outcome is ``_CONVERTED``.
        bids_unit: What the channel's ``bids_unit`` key should hold afterwards --
            the sidecar's declared unit when it was recorded rather than adopted,
            or **None** meaning "there is no unresolved conflict, drop it".
    """

    outcome: str
    unit: str
    factor: float
    bids_unit: str | None

_adopt_units(rec, label, declared, origin)

Apply :func:_decide_unit to one in-memory channel, rescaling its column.

Args: rec: The Recording object to update in place. label: Channel name, already known to exist in rec.channels. declared: The sidecar's units value, already known to be non-empty and not n/a. origin: The sidecar's path, or <DataFrame>, for warnings.

Returns: One of _UNCHANGED, _CONVERTED, _RELABELLED or _KEPT.

Source code in biosigio/bids.py

def _adopt_units(rec: Recording, label: str, declared: str, origin: str) -> str:
    """Apply :func:`_decide_unit` to one in-memory channel, rescaling its column.

    Args:
        rec: The Recording object to update in place.
        label: Channel name, already known to exist in ``rec.channels``.
        declared: The sidecar's ``units`` value, already known to be non-empty
            and not ``n/a``.
        origin: The sidecar's path, or ``<DataFrame>``, for warnings.

    Returns:
        One of ``_UNCHANGED``, ``_CONVERTED``, ``_RELABELLED`` or ``_KEPT``.
    """
    info = rec.channels[label]
    signals = rec.signals
    decision = _decide_unit(
        label=label,
        current=str(info.get("physical_dimension") or "").strip(),
        declared=declared,
        channel_type=str(info.get("channel_type") or ""),
        has_samples=signals is not None and label in signals,
        origin=origin,
        recorded_bids_unit=info.get("bids_unit"),
    )
    if decision.outcome in (_CONVERTED, _RELABELLED):
        if decision.factor != 1.0:
            assert signals is not None  # a conversion decision implies has_samples
            signals[label] = _rescale(signals[label], decision.factor)
        info["physical_dimension"] = decision.unit
    if decision.bids_unit is None:
        info.pop("bids_unit", None)
    else:
        info["bids_unit"] = decision.bids_unit
    return decision.outcome

_decide_unit(*, label, current, declared, channel_type, has_samples, origin, recorded_bids_unit=None)

Decide what moving one channel onto the sidecar's unit means.

The single decision table for both export paths. It answers the question and returns it (see :class:_UnitDecision); the caller performs whatever edit that implies -- rescaling a DataFrame column now (:func:apply_channels_tsv) or carrying a factor into a later windowed read (:func:apply_channels_tsv_to_stream). Splitting the answer from the edit is the whole point: a store built by streaming and a store built in memory must disagree about nothing (issue #127).

A unit label is a claim about the numbers next to it, so the two move together or not at all. The sidecar's units describes the values as the data file stores them, while the importer's unit describes the values biosigIO currently holds -- and those differ whenever the importer rescaled on the way in (every MNE-backed importer returns SI volts regardless of the file's own µV). Adopting the label without the conversion is what issue #122 reports: volts relabelled as microvolts, wrong by 10^6.

The checks run in this order, and the order is load-bearing:

1. Already there (labels equal): nothing at all. This is what makes repeated application idempotent, and it comes first so a discrete or sample-less channel whose unit already matches is not flagged as a conflict with itself.
2. Discrete channel type (TRIG and friends, see :data:~biosigio.core.channel_types.DISCRETE_CHANNEL_TYPES): never rescaled. MNE labels stim channels with the FIFF volts code while they hold integer event codes, so a sidecar declaring mV would turn codes 5/3/7 into 5000/3000/7000. The declared unit is recorded, not applied.
3. No samples: a channel with metadata but no column cannot be rescaled, so it is not relabelled either -- checked before the conversion so even a same-magnitude spelling change cannot slip through on a channel whose numbers are not there to agree with it. (Always false for a streamed channel: every channel the source lists has a row in the transpose memmap.)
4. Convertible (same quantity, e.g. V -> uV): multiply the samples by the ratio and set the label. A ratio of exactly 1 (uV -> µV, a spelling difference) sets the label alone, which is not a semantic relabel.
5. Not convertible (unparsable on either side, or different quantities): keep the importer's values and its label, and record the sidecar's claim under bids_unit so the BIDS metadata is preserved without being asserted over numbers that would contradict it.

Whenever the sidecar's unit is adopted, any bids_unit left by an earlier disagreement is dropped, so the two never both describe the channel.

Args: label: Channel name, for warnings. current: The unit the channel's values are in now, already stripped. declared: The sidecar's units value, already known to be non-empty and not n/a. channel_type: The channel's type, for the discrete-code exemption. has_samples: Whether there are samples this decision can apply to. origin: The sidecar's path, or <DataFrame>, for warnings. recorded_bids_unit: The channel's existing bids_unit, if any.

Returns: The :class:_UnitDecision for this channel.

Source code in biosigio/bids.py

def _decide_unit(
    *,
    label: str,
    current: str,
    declared: str,
    channel_type: str,
    has_samples: bool,
    origin: str,
    recorded_bids_unit: str | None = None,
) -> _UnitDecision:
    """Decide what moving one channel onto the sidecar's unit means.

    **The single decision table for both export paths.** It answers the question
    and returns it (see :class:`_UnitDecision`); the caller performs whatever
    edit that implies -- rescaling a DataFrame column now
    (:func:`apply_channels_tsv`) or carrying a factor into a later windowed read
    (:func:`apply_channels_tsv_to_stream`). Splitting the answer from the edit is
    the whole point: a store built by streaming and a store built in memory must
    disagree about nothing (issue #127).

    A unit label is a claim about the numbers next to it, so the two move
    together or not at all. The sidecar's ``units`` describes the values *as the
    data file stores them*, while the importer's unit describes the values
    biosigIO currently holds -- and those differ whenever the importer rescaled
    on the way in (every MNE-backed importer returns SI volts regardless of the
    file's own µV). Adopting the label without the conversion is what issue #122
    reports: volts relabelled as microvolts, wrong by 10^6.

    The checks run in this order, and the order is load-bearing:

    1. **Already there** (labels equal): nothing at all. This is what makes
       repeated application idempotent, and it comes first so a discrete or
       sample-less channel whose unit already matches is not flagged as a
       conflict with itself.
    2. **Discrete channel type** (``TRIG`` and friends, see
       :data:`~biosigio.core.channel_types.DISCRETE_CHANNEL_TYPES`): never
       rescaled. MNE labels stim channels with the FIFF volts code while they
       hold integer event codes, so a sidecar declaring ``mV`` would turn codes
       5/3/7 into 5000/3000/7000. The declared unit is recorded, not applied.
    3. **No samples**: a channel with metadata but no column cannot be rescaled,
       so it is not relabelled either -- checked before the conversion so even a
       same-magnitude spelling change cannot slip through on a channel whose
       numbers are not there to agree with it. (Always false for a streamed
       channel: every channel the source lists has a row in the transpose memmap.)
    4. **Convertible** (same quantity, e.g. ``V`` -> ``uV``): multiply the samples
       by the ratio and set the label. A ratio of exactly 1 (``uV`` -> ``µV``, a
       spelling difference) sets the label alone, which is not a semantic relabel.
    5. **Not convertible** (unparsable on either side, or different quantities):
       keep the importer's values *and* its label, and record the sidecar's claim
       under ``bids_unit`` so the BIDS metadata is preserved without being
       asserted over numbers that would contradict it.

    Whenever the sidecar's unit *is* adopted, any ``bids_unit`` left by an
    earlier disagreement is dropped, so the two never both describe the channel.

    Args:
        label: Channel name, for warnings.
        current: The unit the channel's values are in now, already stripped.
        declared: The sidecar's ``units`` value, already known to be non-empty
            and not ``n/a``.
        channel_type: The channel's type, for the discrete-code exemption.
        has_samples: Whether there are samples this decision can apply to.
        origin: The sidecar's path, or ``<DataFrame>``, for warnings.
        recorded_bids_unit: The channel's existing ``bids_unit``, if any.

    Returns:
        The :class:`_UnitDecision` for this channel.
    """
    if current == declared:
        return _UnitDecision(_UNCHANGED, current, 1.0, None)

    if str(channel_type or "").upper() in DISCRETE_CHANNEL_TYPES:
        return _keep_importer_unit(
            label,
            current,
            declared,
            "holds discrete codes rather than a measured quantity",
            origin,
            recorded_bids_unit,
        )

    if not has_samples:
        return _keep_importer_unit(
            label,
            current,
            declared,
            "has no samples to rescale",
            origin,
            recorded_bids_unit,
        )

    factor = conversion_factor(current, declared)
    if factor is None:
        return _keep_importer_unit(
            label,
            current,
            declared,
            f"is not convertible to {declared!r}",
            origin,
            recorded_bids_unit,
        )

    return _UnitDecision(_CONVERTED if factor != 1.0 else _RELABELLED, declared, factor, None)

_find_sidecar(data_filepath, kind)

Return the sibling BIDS _<kind>.tsv path for a data file, or None.

BIDS names a sidecar with the data file's entities but a different suffix, e.g. sub-01_task-rest_ieeg.edf -> sub-01_task-rest_channels.tsv.

Args: data_filepath: Path to a (BIDS) data file. kind: Sidecar kind, e.g. "channels" or "events".

Returns: The sidecar path if it exists next to the data file, else None.

Source code in biosigio/bids.py

def _find_sidecar(data_filepath: str, kind: str) -> str | None:
    """Return the sibling BIDS ``_<kind>.tsv`` path for a data file, or None.

    BIDS names a sidecar with the data file's entities but a different suffix,
    e.g. ``sub-01_task-rest_ieeg.edf`` -> ``sub-01_task-rest_channels.tsv``.

    Args:
        data_filepath: Path to a (BIDS) data file.
        kind: Sidecar kind, e.g. ``"channels"`` or ``"events"``.

    Returns:
        The sidecar path if it exists next to the data file, else ``None``.
    """
    directory, filename = os.path.split(data_filepath)
    stem = os.path.splitext(filename)[0]
    candidates = []
    # BIDS-correct: drop the trailing _<suffix> entity (e.g. _ieeg, _eeg, _meg).
    if "_" in stem:
        candidates.append(f"{stem.rsplit('_', 1)[0]}_{kind}.tsv")
    # Fallback: the full-stem form biosigio's own EDF exporter currently writes,
    # so a to_edf -> from_file round-trip also finds the sidecar.
    candidates.append(f"{stem}_{kind}.tsv")
    for name in candidates:
        path = os.path.join(directory, name)
        if os.path.isfile(path):
            return path
    return None

_keep_importer_unit(label, current, declared, reason, origin, recorded_bids_unit)

Record the sidecar's unit without asserting it over contradicting values.

The BIDS metadata is real and worth keeping, so it lands under bids_unit rather than being dropped; what it must not do is overwrite a physical_dimension that correctly describes the samples.

Args: label: Channel name, for the warning. current: The unit the samples are actually in. declared: The sidecar's units value. reason: Why the sidecar's unit was not adopted, for the warning. origin: The sidecar's path, or <DataFrame>, for the warning. recorded_bids_unit: The channel's existing bids_unit, if any.

Returns: A _KEPT decision, or an _UNCHANGED one when this exact bids_unit was already recorded -- so re-applying a sidecar neither re-warns nor re-counts.

Source code in biosigio/bids.py

def _keep_importer_unit(
    label: str,
    current: str,
    declared: str,
    reason: str,
    origin: str,
    recorded_bids_unit: str | None,
) -> _UnitDecision:
    """Record the sidecar's unit without asserting it over contradicting values.

    The BIDS metadata is real and worth keeping, so it lands under ``bids_unit``
    rather than being dropped; what it must not do is overwrite a
    ``physical_dimension`` that correctly describes the samples.

    Args:
        label: Channel name, for the warning.
        current: The unit the samples are actually in.
        declared: The sidecar's ``units`` value.
        reason: Why the sidecar's unit was not adopted, for the warning.
        origin: The sidecar's path, or ``<DataFrame>``, for the warning.
        recorded_bids_unit: The channel's existing ``bids_unit``, if any.

    Returns:
        A ``_KEPT`` decision, or an ``_UNCHANGED`` one when this exact
        ``bids_unit`` was already recorded -- so re-applying a sidecar neither
        re-warns nor re-counts.
    """
    if recorded_bids_unit == declared:
        return _UnitDecision(_UNCHANGED, current, 1.0, declared)
    logging.warning(
        "channels.tsv declares units %r for channel %r, whose values are in %r and %s; "
        "keeping the importer's values and unit, and recording the declared unit "
        "as 'bids_unit': %s",
        declared,
        label,
        current,
        reason,
        origin,
    )
    return _UnitDecision(_KEPT, current, 1.0, declared)

_read_channels_tsv(channels_tsv)

Return a channels.tsv as an all-string frame, from a path or a frame.

Reading with dtype=str, keep_default_na=False is what keeps a literal n/a cell (BIDS's own spelling for "not applicable") a string the callers can test for, rather than a float NaN. A caller-supplied DataFrame gets the equivalent treatment -- stringified, with genuine missing values flattened to the empty cell that means "declares nothing" -- so passing a frame and passing the file it was read from behave identically.

Source code in biosigio/bids.py

def _read_channels_tsv(channels_tsv: str | os.PathLike | pd.DataFrame) -> pd.DataFrame:
    """Return a ``channels.tsv`` as an all-string frame, from a path or a frame.

    Reading with ``dtype=str, keep_default_na=False`` is what keeps a literal
    ``n/a`` cell (BIDS's own spelling for "not applicable") a string the callers
    can test for, rather than a float NaN. A caller-supplied DataFrame gets the
    equivalent treatment -- stringified, with genuine missing values flattened to
    the empty cell that means "declares nothing" -- so passing a frame and passing
    the file it was read from behave identically.
    """
    if isinstance(channels_tsv, pd.DataFrame):
        return channels_tsv.astype(str).mask(channels_tsv.isna(), "")
    return pd.read_csv(channels_tsv, sep="\t", dtype=str, keep_default_na=False)

_rescale(column, factor)

Multiply a signal column, preserving a float column's own precision.

A float column keeps its own dtype: an EEGLAB recording loads at float32 deliberately, to halve memory, and applying a sidecar must not double it back. The cast is defensive rather than currently load-bearing -- pandas treats a scalar operand as weak under NEP 50, so float32 * factor is already float32 today -- but raw numpy promotes the same expression to float64, so the guarantee is stated in the code rather than inherited from a promotion rule that may change.

An integer column has no precision to preserve and becomes float64, which is the only correct result for a non-integral factor.

Source code in biosigio/bids.py

def _rescale(column: pd.Series, factor: float) -> pd.Series:
    """Multiply a signal column, preserving a float column's own precision.

    A float column keeps its own dtype: an EEGLAB recording loads at float32
    deliberately, to halve memory, and applying a sidecar must not double it
    back. The cast is defensive rather than currently load-bearing -- pandas
    treats a scalar operand as weak under NEP 50, so ``float32 * factor`` is
    already float32 today -- but raw numpy promotes the same expression to
    float64, so the guarantee is stated in the code rather than inherited from a
    promotion rule that may change.

    An integer column has no precision to preserve and becomes float64, which is
    the only correct result for a non-integral factor.
    """
    scaled = column * factor
    if pd.api.types.is_float_dtype(column.dtype):
        return scaled.astype(column.dtype)
    return scaled

_sidecar_origin(channels_tsv)

What to call the sidecar in a log message: its path, or that it was a frame.

A DataFrame has no path, and interpolating one into a warning would dump the whole table into the log.

Source code in biosigio/bids.py

def _sidecar_origin(channels_tsv: str | os.PathLike | pd.DataFrame) -> str:
    """What to call the sidecar in a log message: its path, or that it was a frame.

    A DataFrame has no path, and interpolating one into a warning would dump the
    whole table into the log.
    """
    return "<DataFrame>" if isinstance(channels_tsv, pd.DataFrame) else str(channels_tsv)

apply_channels_tsv(rec, channels_tsv)

Override per-channel type/units in rec from a _channels.tsv.

Rows are matched to channels by the name column; n/a and empty values are skipped (the importer-inferred value is kept). An unrecognized type is warned about and skipped rather than raising.

Adopting the sidecar's units converts the channel's samples into that unit rather than merely relabelling them (issue #122); see :func:_decide_unit for the per-channel rule, including the channel types and situations that are exempt. Type and units are applied independently, so an unrecognized type no longer costs the row its unit correction.

A summary of what the units column did lands in rec.metadata["channels_tsv_units"] as {"converted", "relabelled", "kept_importer_unit", "units_column_present"}, so a caller can tell "the sidecar declared no units" from "the units were already correct" without re-scanning every channel.

Args: rec: The Recording object to update in place. channels_tsv: Path to the BIDS _channels.tsv, or an already-loaded DataFrame of it (the two are equivalent; see :func:_read_channels_tsv).

Returns: The number of distinct channels whose record changed -- type adopted, unit adopted, or bids_unit recorded. A channel named by two rows, or by a sidecar applied twice, counts once; a row naming a channel the recording does not have counts not at all.

Source code in biosigio/bids.py

def apply_channels_tsv(rec: Recording, channels_tsv: str | os.PathLike | pd.DataFrame) -> int:
    """Override per-channel ``type``/``units`` in ``rec`` from a ``_channels.tsv``.

    Rows are matched to channels by the ``name`` column; ``n/a`` and empty
    values are skipped (the importer-inferred value is kept). An unrecognized
    ``type`` is warned about and skipped rather than raising.

    Adopting the sidecar's ``units`` **converts the channel's samples** into that
    unit rather than merely relabelling them (issue #122); see
    :func:`_decide_unit` for the per-channel rule, including the channel types
    and situations that are exempt. Type and units are applied independently, so
    an unrecognized ``type`` no longer costs the row its unit correction.

    A summary of what the ``units`` column did lands in
    ``rec.metadata["channels_tsv_units"]`` as
    ``{"converted", "relabelled", "kept_importer_unit", "units_column_present"}``,
    so a caller can tell "the sidecar declared no units" from "the units were
    already correct" without re-scanning every channel.

    Args:
        rec: The Recording object to update in place.
        channels_tsv: Path to the BIDS ``_channels.tsv``, or an already-loaded
            DataFrame of it (the two are equivalent; see
            :func:`_read_channels_tsv`).

    Returns:
        The number of **distinct channels** whose record changed -- type adopted,
        unit adopted, or ``bids_unit`` recorded. A channel named by two rows, or
        by a sidecar applied twice, counts once; a row naming a channel the
        recording does not have counts not at all.
    """
    df = _read_channels_tsv(channels_tsv)
    origin = _sidecar_origin(channels_tsv)
    if "name" not in df.columns:
        logging.warning("channels.tsv has no 'name' column: %s", origin)
        return 0

    units_present = "units" in df.columns
    if not units_present:
        logging.warning("channels.tsv has no 'units' column: %s", origin)

    report = {_CONVERTED: 0, _RELABELLED: 0, _KEPT: 0, "units_column_present": units_present}
    changed: set[str] = set()
    matched: set[str] = set()
    for _, row in df.iterrows():
        name = str(row["name"]).strip()
        if name not in rec.channels:
            continue
        matched.add(name)
        ctype = str(row.get("type", "")).strip()
        if ctype and ctype.lower() != "n/a":
            try:
                rec.set_channel(name, channel_type=ctype)
                changed.add(name)
            except ValueError:
                logging.warning(
                    "channels.tsv type %r for channel %r is not a known channel type; "
                    "keeping the importer-inferred type: %s",
                    ctype,
                    name,
                    origin,
                )
        units = str(row.get("units", "")).strip()
        if units and units.lower() != "n/a":
            outcome = _adopt_units(rec, name, units, origin)
            if outcome != _UNCHANGED:
                report[outcome] += 1
                changed.add(name)
        elif units_present:
            logging.debug(
                "channels.tsv declares no unit (%r) for channel %r: %s", units, name, origin
            )

    uncovered = [label for label in rec.channels if label not in matched]
    if uncovered:
        logging.debug(
            "channels.tsv names no row for %d of the recording's channels (e.g. %s): %s",
            len(uncovered),
            ", ".join(repr(label) for label in uncovered[:5]),
            origin,
        )

    rec.set_metadata("channels_tsv_units", report)
    return len(changed)

apply_channels_tsv_to_stream(channels, channels_tsv, *, force_modality=None)

Apply a _channels.tsv to a streaming source's channel table.

The streaming Zarr exporter has no :class:~biosigio.core.emg.Recording and no samples in memory: it knows each channel's label, type, modality and unit, and will read the values window by window afterwards. So this settles every per-channel question through the same :func:_decide_unit table the in-memory path uses and leaves the arithmetic for later, as a per-channel unit_factor the exporter multiplies into each channel exactly once. Without it, a dataset whose recordings straddle a streaming size threshold serves its small runs in the sidecar's unit and its large ones in the importer's native unit -- a 10^6 disagreement inside one dataset (issue #127).

Each entry in channels is mutated in place: channel_type (and, unless force_modality pins it, modality) from the sidecar's type; unit and unit_factor from its units; bids_unit when a declared unit was recorded rather than adopted. Entries the sidecar does not name are left exactly as the importer built them.

Rows are matched by the name column, and several rows naming one channel compose in file order, the same as :func:apply_channels_tsv. Unlike a Recording -- whose channels are a dict and so unique by label -- a streaming source may list the same label twice (EDF permits it); every entry with that label gets the same treatment, so duplicate labels cannot end up in different units inside one store.

Args: channels: The source's per-channel dicts (label, channel_type, modality, unit), mutated in place. channels_tsv: Path to the sidecar, or an already-loaded DataFrame. force_modality: When set, the caller has pinned every channel to one modality (the BIDS datatype suffix, say) and the sidecar's type must not move it. The type is still adopted per channel.

Returns: The channels_tsv_units report, in the same shape :func:apply_channels_tsv leaves in rec.metadata, or None when the sidecar has no name column and nothing could be applied -- so the exporter records an attr exactly when the in-memory path would.

Source code in biosigio/bids.py

def apply_channels_tsv_to_stream(
    channels: list[dict],
    channels_tsv: str | os.PathLike | pd.DataFrame,
    *,
    force_modality: str | None = None,
) -> dict | None:
    """Apply a ``_channels.tsv`` to a **streaming** source's channel table.

    The streaming Zarr exporter has no :class:`~biosigio.core.emg.Recording` and
    no samples in memory: it knows each channel's label, type, modality and unit,
    and will read the values window by window afterwards. So this settles every
    per-channel question through the same :func:`_decide_unit` table the
    in-memory path uses and leaves the *arithmetic* for later, as a per-channel
    ``unit_factor`` the exporter multiplies into each channel exactly once. Without
    it, a dataset whose recordings straddle a streaming size threshold serves its
    small runs in the sidecar's unit and its large ones in the importer's native
    unit -- a 10^6 disagreement inside one dataset (issue #127).

    Each entry in ``channels`` is mutated in place: ``channel_type`` (and, unless
    ``force_modality`` pins it, ``modality``) from the sidecar's ``type``;
    ``unit`` and ``unit_factor`` from its ``units``; ``bids_unit`` when a declared
    unit was recorded rather than adopted. Entries the sidecar does not name are
    left exactly as the importer built them.

    Rows are matched by the ``name`` column, and several rows naming one channel
    compose in file order, the same as :func:`apply_channels_tsv`. Unlike a
    Recording -- whose channels are a dict and so unique by label -- a streaming
    source may list the same label twice (EDF permits it); every entry with that
    label gets the same treatment, so duplicate labels cannot end up in different
    units inside one store.

    Args:
        channels: The source's per-channel dicts (``label``, ``channel_type``,
            ``modality``, ``unit``), mutated in place.
        channels_tsv: Path to the sidecar, or an already-loaded DataFrame.
        force_modality: When set, the caller has pinned every channel to one
            modality (the BIDS datatype suffix, say) and the sidecar's ``type``
            must not move it. The ``type`` is still adopted per channel.

    Returns:
        The ``channels_tsv_units`` report, in the same shape
        :func:`apply_channels_tsv` leaves in ``rec.metadata``, or **None** when
        the sidecar has no ``name`` column and nothing could be applied -- so the
        exporter records an attr exactly when the in-memory path would.
    """
    df = _read_channels_tsv(channels_tsv)
    origin = _sidecar_origin(channels_tsv)
    if "name" not in df.columns:
        logging.warning("channels.tsv has no 'name' column: %s", origin)
        return None

    units_present = "units" in df.columns
    if not units_present:
        logging.warning("channels.tsv has no 'units' column: %s", origin)

    report = {_CONVERTED: 0, _RELABELLED: 0, _KEPT: 0, "units_column_present": units_present}
    rows_by_name: dict[str, list[pd.Series]] = {}
    for _, row in df.iterrows():
        rows_by_name.setdefault(str(row["name"]).strip(), []).append(row)

    uncovered: list[str] = []
    for entry in channels:
        # The label is NOT stripped, and must not be: apply_channels_tsv matches a
        # stripped row name against the Recording's channel keys as they are, so
        # stripping here would match rows on this path that the in-memory path
        # leaves unmatched -- a difference between the two exports, which is the
        # one thing this function exists to prevent.
        label = str(entry["label"])
        rows = rows_by_name.get(label)
        if not rows:
            uncovered.append(label)
            continue
        for row in rows:
            ctype = str(row.get("type", "")).strip()
            if ctype and ctype.lower() != "n/a":
                try:
                    entry["channel_type"] = validate_channel_type(ctype)
                    if force_modality is None:
                        entry["modality"] = infer_modality_from_channel_type(entry["channel_type"])
                except ValueError:
                    logging.warning(
                        "channels.tsv type %r for channel %r is not a known channel type; "
                        "keeping the importer-inferred type: %s",
                        ctype,
                        label,
                        origin,
                    )
            units = str(row.get("units", "")).strip()
            if units and units.lower() != "n/a":
                decision = _decide_unit(
                    label=label,
                    current=str(entry.get("unit") or "").strip(),
                    declared=units,
                    channel_type=str(entry.get("channel_type") or ""),
                    # Every channel the source lists has a row in the exporter's
                    # transpose memmap, so the "no samples to rescale" branch --
                    # which exists for a Recording carrying channel metadata
                    # without a column -- cannot apply here.
                    has_samples=True,
                    origin=origin,
                    recorded_bids_unit=entry.get("bids_unit"),
                )
                if decision.outcome != _UNCHANGED:
                    report[decision.outcome] += 1
                if decision.outcome in (_CONVERTED, _RELABELLED):
                    entry["unit"] = decision.unit
                    entry["unit_factor"] = float(entry.get("unit_factor", 1.0)) * decision.factor
                if decision.bids_unit is None:
                    entry.pop("bids_unit", None)
                else:
                    entry["bids_unit"] = decision.bids_unit
            elif units_present:
                logging.debug(
                    "channels.tsv declares no unit (%r) for channel %r: %s", units, label, origin
                )

    if uncovered:
        logging.debug(
            "channels.tsv names no row for %d of the recording's channels (e.g. %s): %s",
            len(uncovered),
            ", ".join(repr(label) for label in uncovered[:5]),
            origin,
        )
    return report

apply_events_tsv(rec, events_tsv_path, *, description_column=None)

Replace rec.events with the authoritative BIDS _events.tsv list.

A BIDS _events.tsv is the curated event table for a recording; it is richer and more reliable than the data file's own markers (e.g. a BrainVision .vmrk or an EEGLAB event struct), so when it is present it is the source of truth. This loads it into rec.events as the standard onset/duration/description frame, overwriting any importer-loaded events.

Columns: onset (required, seconds) and duration (seconds; n/a or missing -> 0.0) follow the BIDS spec. The description is taken from description_column if given, else per row from the first of trial_type then value that is present and not n/a (BIDS names the categorical label trial_type and the raw marker value; datasets populate one or both). Rows whose onset is missing or non-numeric are skipped.

Args: rec: The Recording object to update in place. events_tsv_path: Path to the BIDS _events.tsv. description_column: Force the description to come from this column.

Returns: The number of events loaded into rec.events. An unparsable sidecar (no onset column / missing forced column) loads nothing and leaves any importer-loaded events intact, returning 0.

Source code in biosigio/bids.py

def apply_events_tsv(
    rec: Recording,
    events_tsv_path: str,
    *,
    description_column: str | None = None,
) -> int:
    """Replace ``rec.events`` with the authoritative BIDS ``_events.tsv`` list.

    A BIDS ``_events.tsv`` is the curated event table for a recording; it is
    richer and more reliable than the data file's own markers (e.g. a
    BrainVision ``.vmrk`` or an EEGLAB event struct), so when it is present it
    is the source of truth. This loads it into ``rec.events`` as the standard
    ``onset``/``duration``/``description`` frame, overwriting any
    importer-loaded events.

    Columns:
        ``onset`` (required, seconds) and ``duration`` (seconds; ``n/a`` or
        missing -> ``0.0``) follow the BIDS spec. The ``description`` is taken
        from ``description_column`` if given, else per row from the first of
        ``trial_type`` then ``value`` that is present and not ``n/a`` (BIDS
        names the categorical label ``trial_type`` and the raw marker
        ``value``; datasets populate one or both). Rows whose ``onset`` is
        missing or non-numeric are skipped.

    Args:
        rec: The Recording object to update in place.
        events_tsv_path: Path to the BIDS ``_events.tsv``.
        description_column: Force the description to come from this column.

    Returns:
        The number of events loaded into ``rec.events``. An unparsable sidecar
        (no ``onset`` column / missing forced column) loads nothing and leaves any
        importer-loaded events intact, returning 0.
    """
    events = read_events_tsv(events_tsv_path, description_column=description_column)
    if events is None:
        return 0
    rec.events = events
    return len(rec.events)

conversion_factor(from_unit, to_unit)

The multiplier that re-expresses a value from from_unit in to_unit.

Args: from_unit: The unit the values are currently in. to_unit: The unit the values should be expressed in.

Returns: A float k such that value_in_to_unit == value_in_from_unit * k, or None when either unit is unparsable or the two measure different quantities. None means "do not touch the values"; it is never a conversion of 1.0 in disguise.

Examples: >>> conversion_factor("V", "uV") 1000000.0 >>> conversion_factor("uV", "uV") 1.0 >>> conversion_factor("V", "T") is None True

Source code in biosigio/units.py

def conversion_factor(from_unit: str | None, to_unit: str | None) -> float | None:
    """The multiplier that re-expresses a value from ``from_unit`` in ``to_unit``.

    Args:
        from_unit: The unit the values are currently in.
        to_unit: The unit the values should be expressed in.

    Returns:
        A float ``k`` such that ``value_in_to_unit == value_in_from_unit * k``,
        or **None** when either unit is unparsable or the two measure different
        quantities. None means "do not touch the values"; it is never a
        conversion of 1.0 in disguise.

    Examples:
        >>> conversion_factor("V", "uV")
        1000000.0
        >>> conversion_factor("uV", "uV")
        1.0
        >>> conversion_factor("V", "T") is None
        True
    """
    source = parse_unit(from_unit)
    target = parse_unit(to_unit)
    if source is None or target is None or source[0] != target[0]:
        return None
    return 10.0 ** (source[1] - target[1])

find_channels_tsv(data_filepath)

Return the sibling BIDS _channels.tsv path for a data file, or None.

Source code in biosigio/bids.py

def find_channels_tsv(data_filepath: str) -> str | None:
    """Return the sibling BIDS ``_channels.tsv`` path for a data file, or None."""
    return _find_sidecar(data_filepath, "channels")

find_events_tsv(data_filepath)

Return the sibling BIDS _events.tsv path for a data file, or None.

Source code in biosigio/bids.py

def find_events_tsv(data_filepath: str) -> str | None:
    """Return the sibling BIDS ``_events.tsv`` path for a data file, or None."""
    return _find_sidecar(data_filepath, "events")

infer_modality_from_channel_type(channel_type)

Derive the coarse modality for a channel type.

The mapping is deterministic: neural types map to EEG/IEEG/MEG, EMG maps to EMG, and every other valid type (ECG, EOG, ACC, TRIG, ...) maps to MISC.

Args: channel_type: A channel type string (case-insensitive); validated first.

Returns: One of :data:VALID_MODALITIES.

Raises: ValueError: If channel_type is not a known type.

Source code in biosigio/core/modality.py

def infer_modality_from_channel_type(channel_type: str) -> str:
    """Derive the coarse modality for a channel type.

    The mapping is deterministic: neural types map to EEG/IEEG/MEG, ``EMG`` maps
    to EMG, and every other valid type (ECG, EOG, ACC, TRIG, ...) maps to MISC.

    Args:
        channel_type: A channel type string (case-insensitive); validated first.

    Returns:
        One of :data:`VALID_MODALITIES`.

    Raises:
        ValueError: If ``channel_type`` is not a known type.
    """
    ct = validate_channel_type(channel_type)
    return _CHANNEL_TYPE_TO_MODALITY.get(ct, "MISC")

read_events_tsv(events_tsv_path, *, description_column=None)

Parse a BIDS _events.tsv into the standard events frame.

Returns a DataFrame with onset/duration/description columns (sorted by onset), the same shape :attr:Recording.events uses, or None when the sidecar is unparsable (no onset column, or a forced description_column is absent) -- distinct from a valid-but-empty table (an empty DataFrame). Callers use None to leave any existing events untouched rather than wiping them. Used by :func:apply_events_tsv and by the streaming Zarr exporter (which has no Recording to mutate). See :func:apply_events_tsv for the column rules.

Source code in biosigio/bids.py

def read_events_tsv(
    events_tsv_path: str,
    *,
    description_column: str | None = None,
) -> pd.DataFrame | None:
    """Parse a BIDS ``_events.tsv`` into the standard events frame.

    Returns a DataFrame with ``onset``/``duration``/``description`` columns
    (sorted by onset), the same shape :attr:`Recording.events` uses, or **None**
    when the sidecar is unparsable (no ``onset`` column, or a forced
    ``description_column`` is absent) -- distinct from a valid-but-empty table (an
    empty DataFrame). Callers use None to leave any existing events untouched
    rather than wiping them. Used by :func:`apply_events_tsv` and by the streaming
    Zarr exporter (which has no Recording to mutate). See :func:`apply_events_tsv`
    for the column rules.
    """
    df = pd.read_csv(events_tsv_path, sep="\t", dtype=str, keep_default_na=False)
    if "onset" not in df.columns:
        logging.warning("events.tsv has no 'onset' column: %s", events_tsv_path)
        return None

    def _is_na(value: str) -> bool:
        v = value.strip()
        return v == "" or v.lower() == "n/a"

    if description_column is not None:
        if description_column not in df.columns:
            logging.warning("events.tsv has no %r column: %s", description_column, events_tsv_path)
            return None
        desc_columns = [description_column]
    else:
        desc_columns = [c for c in ("trial_type", "value") if c in df.columns]

    onsets: list[float] = []
    durations: list[float] = []
    descriptions: list[str] = []
    skipped = 0
    for _, row in df.iterrows():
        raw_onset = str(row["onset"]).strip()
        try:
            onset = float(raw_onset)
        except ValueError:
            skipped += 1
            continue
        if not math.isfinite(onset):  # NaN or +/-inf
            skipped += 1
            continue
        raw_duration = str(row.get("duration", "")).strip()
        try:
            duration = 0.0 if _is_na(raw_duration) else float(raw_duration)
        except ValueError:
            duration = 0.0
        description = "n/a"
        for col in desc_columns:
            candidate = str(row.get(col, "")).strip()
            if not _is_na(candidate):
                description = candidate
                break
        onsets.append(onset)
        durations.append(duration)
        descriptions.append(description)

    if skipped:
        logging.warning(
            "events.tsv: skipped %d row(s) with missing/non-numeric onset: %s",
            skipped,
            events_tsv_path,
        )

    return (
        pd.DataFrame({"onset": onsets, "duration": durations, "description": descriptions})
        .sort_values(by="onset")
        .reset_index(drop=True)
    )

resolve_channels_tsv(filepath, bids_channels)

Turn a bids_channels argument into the sidecar to apply, or None.

The one place the bids_channels vocabulary is interpreted, shared by :meth:~biosigio.core.emg.Recording.from_file and :func:~biosigio.exporters.zarr_stream.stream_to_zarr so the two export paths cannot come to differ about what an argument means -- which is the same reason :func:_decide_unit is shared (issue #127).

• "auto" (the default both callers use) resolves the sibling _channels.tsv through :func:find_channels_tsv, and yields None when the recording has none.
• "off", and None as its synonym, disable the lookup.
• Anything else is a path or a DataFrame the caller chose explicitly. NEMAR's MaxShield path needs this: it converts a filtered copy written to scratch, where the recording's real sidecar is not adjacent.

The string cases are tested first because a DataFrame compared against "auto" compares elementwise and has no truth value.

Args: filepath: The recording, used only to resolve "auto". bids_channels: "auto", "off", None, a path, or a DataFrame.

Returns: A path, a DataFrame, or None when no sidecar should be applied.

Source code in biosigio/bids.py

def resolve_channels_tsv(
    filepath: str, bids_channels: str | os.PathLike | pd.DataFrame | None
) -> str | os.PathLike | pd.DataFrame | None:
    """Turn a ``bids_channels`` argument into the sidecar to apply, or None.

    The one place the ``bids_channels`` vocabulary is interpreted, shared by
    :meth:`~biosigio.core.emg.Recording.from_file` and
    :func:`~biosigio.exporters.zarr_stream.stream_to_zarr` so the two export
    paths cannot come to differ about what an argument means -- which is the
    same reason :func:`_decide_unit` is shared (issue #127).

    - ``"auto"`` (the default both callers use) resolves the sibling
      ``_channels.tsv`` through :func:`find_channels_tsv`, and yields None when
      the recording has none.
    - ``"off"``, and None as its synonym, disable the lookup.
    - Anything else is a path or a DataFrame the caller chose explicitly. NEMAR's
      MaxShield path needs this: it converts a filtered copy written to scratch,
      where the recording's real sidecar is not adjacent.

    The string cases are tested first because a DataFrame compared against
    ``"auto"`` compares elementwise and has no truth value.

    Args:
        filepath: The recording, used only to resolve ``"auto"``.
        bids_channels: ``"auto"``, ``"off"``, None, a path, or a DataFrame.

    Returns:
        A path, a DataFrame, or **None** when no sidecar should be applied.
    """
    if isinstance(bids_channels, str):
        if bids_channels == "auto":
            # rstrip for the directory-valued recordings (CTF .ds, 4D/BTi), so a
            # trailing slash does not leave os.path.split with an empty filename.
            return find_channels_tsv(filepath.rstrip("/\\"))
        return None if bids_channels == "off" else bids_channels
    return bids_channels

validate_channel_type(channel_type)

Normalize and validate a channel type against :data:VALID_CHANNEL_TYPES.

Args: channel_type: A channel type string (case-insensitive).

Returns: The canonical uppercase channel type.

Raises: ValueError: If channel_type is empty, n/a, or not a known type.

Source code in biosigio/core/modality.py

def validate_channel_type(channel_type: str) -> str:
    """Normalize and validate a channel type against :data:`VALID_CHANNEL_TYPES`.

    Args:
        channel_type: A channel type string (case-insensitive).

    Returns:
        The canonical uppercase channel type.

    Raises:
        ValueError: If ``channel_type`` is empty, ``n/a``, or not a known type.
    """
    ct = channel_type.strip().upper()
    if ct in ("", "N/A", "NA"):
        raise ValueError(
            "channel_type 'n/a' is not allowed; use 'OTHER' or 'MISC' for an unknown type."
        )
    if ct not in VALID_CHANNEL_TYPES:
        raise ValueError(
            f"Unknown channel_type {channel_type!r}. Valid channel types: "
            f"{sorted(VALID_CHANNEL_TYPES)}"
        )
    return ct