CSV Importer

The CSVImporter class is responsible for importing EMG and other physiological data from generic CSV files with flexible format detection and configuration options.

Class Documentation

biosigio.importers.csv

BaseImporter

Bases: ABC

Base class for EMG data importers.

Source code in biosigio/importers/base.py

class BaseImporter(ABC):
    """Base class for EMG data importers."""

    @abstractmethod
    def load(self, filepath: str) -> Recording:
        """
        Load EMG data from file.

        Args:
            filepath: Path to the input file

        Returns:
            Recording: Recording object containing the loaded data
        """
        pass

load(filepath) abstractmethod

Load EMG data from file.

Args: filepath: Path to the input file

Returns: Recording: Recording object containing the loaded data

Source code in biosigio/importers/base.py

@abstractmethod
def load(self, filepath: str) -> Recording:
    """
    Load EMG data from file.

    Args:
        filepath: Path to the input file

    Returns:
        Recording: Recording object containing the loaded data
    """
    pass

CSVImporter

Bases: BaseImporter

General purpose CSV importer for EMG data.

This importer can handle various CSV formats with columnar data, auto-detect headers, time columns, and allow for specific column selection.

Source code in biosigio/importers/csv.py

class CSVImporter(BaseImporter):
    """
    General purpose CSV importer for EMG data.

    This importer can handle various CSV formats with columnar data, auto-detect
    headers, time columns, and allow for specific column selection.
    """

    def _detect_specialized_format(self, filepath: str) -> str | None:
        """
        Detect if the file matches a known specialized format.

        Args:
            filepath: Path to the CSV file

        Returns:
            Name of the detected specialized format, or None if no specific format is detected
        """
        # Try to read the first few lines to check for format signatures
        try:
            with open(filepath) as f:
                header_lines = [f.readline().strip() for _ in range(20)]
                header_text = "\n".join(header_lines)

                # Check for Trigno format signatures
                if any(marker in header_text for marker in ["Trigno", "Delsys", "Label:", "X[s]"]):
                    return "trigno"

                # Additional format checks can be added here for other importers
                # For example:
                # if 'OTB' in header_text or 'Sessantaquattro' in header_text:
                #     return 'otb'

        except Exception:
            # If we can't read the file or encounter an error,
            # don't try to guess the format
            pass

        return None

    def load(self, filepath: str, force_generic: bool = False, **kwargs) -> Recording:
        """
        Load EMG data from a CSV file.

        Args:
            filepath: Path to the CSV file
            force_generic: If True, forces using the generic CSV importer even if a
                          specialized format is detected
            **kwargs: Additional options including:
                - columns: List of column names or indices to include
                - time_column: Name or index of column to use as time index (default: auto-detect)
                - has_header: Whether file has a header row (default: auto-detect)
                - skiprows: Number of rows to skip at the beginning (default: auto-detect)
                - delimiter: Column delimiter (default: auto-detect)
                - sample_frequency: Sampling frequency in Hz (required if no time column)
                - channel_types: Dict mapping column names to channel types ('EMG', 'ACC', etc.)
                - physical_dimensions: Dict mapping column names to physical dimensions
                - metadata: Dict of additional metadata to include

        Returns:
            Recording: Recording object containing the loaded data

        Raises:
            ValueError: If a specialized format is detected and force_generic is False
            FileNotFoundError: If the file does not exist
        """
        # Check if this file matches a specialized format
        if not force_generic:
            format_name = self._detect_specialized_format(filepath)
            if format_name:
                importer_messages = {
                    "trigno": (
                        "This file appears to be a Delsys Trigno CSV export. "
                        "For better metadata extraction and channel detection, use:\n\n"
                        "recording = Recording.from_file(filepath, importer='trigno')\n\n"
                        "If you still want to use the generic CSV importer, set force_generic=True:\n"
                        "importer = CSVImporter()\n"
                        "recording = importer.load(filepath, force_generic=True, **params)"
                    )
                    # Add more format-specific messages here as new importers are developed
                }

                if format_name in importer_messages:
                    raise ValueError(importer_messages[format_name])

        # Extract kwargs with defaults
        columns = kwargs.get("columns", None)
        time_column = kwargs.get("time_column", None)
        has_header = kwargs.get("has_header", None)
        skiprows = kwargs.get("skiprows", None)
        delimiter = kwargs.get("delimiter", None)
        sample_frequency = kwargs.get("sample_frequency", None)
        channel_names = kwargs.get("channel_names", [])
        channel_types = kwargs.get("channel_types", {})
        physical_dimensions = kwargs.get("physical_dimensions", {})
        metadata = kwargs.get("metadata", {})

        # Analyze file structure if parameters not explicitly provided
        try:
            if any(param is None for param in [has_header, skiprows, delimiter]):
                analyzed_params = self._analyze_csv_structure(filepath)

                # Use analyzed parameters for any not explicitly provided
                has_header = has_header if has_header is not None else analyzed_params["has_header"]
                skiprows = skiprows if skiprows is not None else analyzed_params["skiprows"]
                delimiter = delimiter if delimiter is not None else analyzed_params["delimiter"]
        except FileNotFoundError:
            # Pass through file not found errors
            raise

        # Read the CSV file
        try:
            df = pd.read_csv(
                filepath,
                header=0 if has_header else None,
                skiprows=skiprows,
                delimiter=delimiter,
                index_col=None,
            )
        except FileNotFoundError:
            # Pass through file not found errors
            raise
        except Exception as e:
            raise ValueError(f"Failed to read CSV file: {str(e)}") from e

        # If no header, generate column names
        if not has_header:
            df.columns = [f"Channel_{i}" for i in range(len(df.columns))]

        # If channel names are provided, use them.
        # Also, handle the case where the length of channel_names is less than the number of columns.
        if channel_names:
            if len(channel_names) < len(df.columns):
                raise ValueError(
                    "Number of channel names provided is less than the number of columns in the CSV file."
                )
            df.columns = channel_names

        # Filter columns if specified
        if columns is not None:
            if all(isinstance(col, int) for col in columns):
                # Convert numerical indices to column names
                col_names = [df.columns[i] for i in columns]
                # Save original columns for potential renumbering
                df = df[col_names]

                # If using default channel names, renumber them sequentially
                if not has_header and not channel_names:
                    # Check if these are auto-generated channel names
                    if all(col.startswith("Channel_") for col in col_names):
                        # Rename columns to be sequential
                        new_names = [f"Channel_{i}" for i in range(len(col_names))]
                        rename_map = dict(zip(col_names, new_names, strict=True))
                        df = df.rename(columns=rename_map)
            else:
                # Filter by column names
                df = df[columns]

        # Handle time column
        if time_column is not None:
            # If time_column is an index, convert to column name
            if isinstance(time_column, int):
                time_column = df.columns[time_column]

            # Set time column as index
            if time_column in df.columns:
                df.set_index(time_column, inplace=True)
            else:
                raise ValueError(f"Time column '{time_column}' not found in data")
        elif has_header:
            # When header exists, try to auto-detect time column only if has_header is True
            time_col = self._detect_time_column(df)
            if time_col:
                df.set_index(time_col, inplace=True)
            elif sample_frequency:
                # Create time index based on provided sampling frequency
                time_index = np.arange(len(df)) / sample_frequency
                df.index = time_index
            else:
                # No time column and no sample frequency provided
                raise ValueError(
                    "No time column detected and no sample_frequency provided. "
                    "Please specify either time_column or sample_frequency."
                )
        else:
            # For headerless data, don't attempt to auto-detect time column
            # to avoid treating the first column as time
            if sample_frequency:
                # Create time index based on provided sampling frequency
                time_index = np.arange(len(df)) / sample_frequency
                df.index = time_index
            else:
                # No sample frequency provided
                raise ValueError(
                    "No sample_frequency provided for headerless data. "
                    "Please specify sample_frequency for proper time indexing."
                )

        # Create recording object
        rec = Recording()

        # Add metadata
        rec.set_metadata("source_file", filepath)
        rec.set_metadata("file_format", "CSV")

        # Add any user-provided metadata
        for key, value in metadata.items():
            rec.set_metadata(key, value)

        # Default sampling frequency if not specified
        default_sample_frequency = 1000.0  # 1 kHz is a common default for EMG
        if hasattr(df.index, "to_series"):
            # Calculate sampling frequency from time index if possible
            try:
                time_diffs = df.index.to_series().diff().dropna()
                if len(time_diffs) > 0:
                    avg_diff = time_diffs.mean()
                    if avg_diff > 0:
                        calculated_freq = 1.0 / avg_diff
                        default_sample_frequency = calculated_freq
            except Exception:
                # If calculation fails, keep default
                pass

        # Add each column as a channel
        for column in df.columns:
            # Determine channel type
            if column in channel_types:
                ch_type = channel_types[column]
            else:
                # Try to infer channel type from name
                ch_type = self._infer_channel_type(column)

            # Determine physical dimension
            if column in physical_dimensions:
                phys_dim = physical_dimensions[column]
            else:
                # Default based on channel type
                phys_dim = self._default_physical_dimension(ch_type)

            # Add the channel to the recording
            rec.add_channel(
                label=column,
                data=df[column].values,
                sample_frequency=sample_frequency or default_sample_frequency,
                physical_dimension=phys_dim,
                channel_type=ch_type,
            )

        # Encourage user to add metadata if missing essential information
        self._print_metadata_reminder(rec)

        return rec

    def _analyze_csv_structure(self, filepath: str) -> dict:
        """
        Analyze the CSV file structure to detect delimiter, headers, and rows to skip.

        Args:
            filepath: Path to the CSV file

        Returns:
            Dict with detected parameters:
                - delimiter: Detected delimiter character
                - has_header: Whether the file has a header row
                - skiprows: Number of rows to skip
        """
        # Default results
        results = {"delimiter": ",", "has_header": True, "skiprows": 0}

        try:
            # Read the first few lines to analyze structure
            with open(filepath) as f:
                lines = [
                    f.readline().strip() for _ in range(30)
                ]  # Read first 30 lines or until EOF
                lines = [line for line in lines if line]  # Remove empty lines

                # Special case for Trigno CSV format
                data_start = 0
                for i, line in enumerate(lines):
                    if "X[s]" in line:
                        data_start = i
                        results["skiprows"] = data_start
                        results["has_header"] = True
                        break

                if data_start > 0:
                    # Found a header line with X[s], use the line after it as data
                    return results

                # If not a special format, continue with regular analysis
                # Count occurrences of each delimiter and choose the most common one
                delimiters = {",": 0, "\t": 0, ";": 0, "|": 0}

                for line in lines[:5]:  # Check first 5 lines
                    if not line or line.startswith("#"):
                        continue

                    for delim in delimiters:
                        if delim in line:
                            # Count occurrences but also consider how many fields it creates
                            fields = line.split(delim)
                            if len(fields) > 1:  # Must create at least 2 fields to be valid
                                delimiters[delim] += len(fields)

                # Choose the delimiter that creates the most fields
                if any(delimiters.values()):
                    most_common = max(delimiters.items(), key=lambda x: x[1])
                    results["delimiter"] = most_common[0]

                # Infer if file has a header by checking if first row looks different from data rows
                if len(lines) >= 2:
                    possible_header = lines[0]
                    possible_data = lines[1]

                    # If first row contains alphabetic characters and data rows are numeric
                    header_values = possible_header.split(results["delimiter"])
                    data_values = possible_data.split(results["delimiter"])

                    # Check for alpha chars in header
                    has_alpha = any(
                        any(c.isalpha() for c in val) for val in header_values if val.strip()
                    )
                    # Check if data rows are numeric
                    numeric_data = all(self._is_numeric(val) for val in data_values if val.strip())

                    if has_alpha and numeric_data:
                        results["has_header"] = True
                    else:
                        # If no clear distinction, assume no header if all fields look numeric
                        results["has_header"] = not all(
                            self._is_numeric(val) for val in header_values if val.strip()
                        )

        except Exception:
            # If analysis fails, return defaults
            pass

        return results

    def _is_numeric(self, value: str) -> bool:
        """Check if a string value is numeric."""
        try:
            float(value)
            return True
        except ValueError:
            return False

    def _detect_time_column(self, df: pd.DataFrame) -> str | None:
        """
        Try to detect which column represents time.

        Args:
            df: DataFrame with loaded data

        Returns:
            Name of detected time column or None if not found
        """
        time_keywords = ["time", "second", "seconds", "s"]

        # Check column names for time keywords
        for col in df.columns:
            col_lower = col.lower()
            if any(keyword in col_lower for keyword in time_keywords):
                return col

        # Check if first column is monotonically increasing (typical for time)
        first_col = df.columns[0]
        if len(df) > 1 and pd.Series(df[first_col]).is_monotonic_increasing:
            # Check if the values are plausible time values (e.g., not all integers if diff is small)
            if df[first_col].dtype in [np.float64, np.float32]:
                return first_col
            elif (
                df[first_col].diff().dropna().mean() > 1e-9
            ):  # Avoid treating integer indices as time
                return first_col

        return None

    def _infer_channel_type(self, column_name: str) -> str:
        """
        Infer channel type from column name.

        Args:
            column_name: Name of the column

        Returns:
            Inferred channel type
        """
        name_lower = column_name.lower()

        if any(keyword in name_lower for keyword in ["emg", "muscle"]):
            return "EMG"
        elif any(keyword in name_lower for keyword in ["acc", "accel"]):
            return "ACC"
        elif any(keyword in name_lower for keyword in ["gyro"]):
            return "GYRO"
        elif any(keyword in name_lower for keyword in ["time", "second"]):
            # A time column is normally consumed as the index. If one still
            # reaches here it is not a signal channel; classify it as MISC
            # ("TIME" is not a valid BIDS/modality channel type).
            return "MISC"
        else:
            return "OTHER"

    def _default_physical_dimension(self, channel_type: str) -> str:
        """
        Return default physical dimension for a channel type.

        Args:
            channel_type: Type of channel

        Returns:
            Default physical dimension
        """
        dimensions = {"EMG": "µV", "ACC": "g", "GYRO": "deg/s", "TIME": "s", "OTHER": "a.u."}
        return dimensions.get(channel_type, "a.u.")

    def _print_metadata_reminder(self, rec: Recording) -> None:
        """
        Print a reminder to add metadata if essential information is missing.

        Args:
            rec: Recording object to check
        """
        essential_metadata = ["subject", "device", "recording_date"]
        missing = [meta for meta in essential_metadata if meta not in rec.metadata]

        if missing:
            print("[INFO] Reminder: Consider adding essential metadata for better context:")
            for meta in missing:
                print(f"  rec.set_metadata('{meta}', '<Your {meta.replace('_', ' ').title()}>')")
            print("Example: rec.set_metadata('subject', 'S001')")

load(filepath, force_generic=False, **kwargs)

Load EMG data from a CSV file.

Args: filepath: Path to the CSV file force_generic: If True, forces using the generic CSV importer even if a specialized format is detected **kwargs: Additional options including: - columns: List of column names or indices to include - time_column: Name or index of column to use as time index (default: auto-detect) - has_header: Whether file has a header row (default: auto-detect) - skiprows: Number of rows to skip at the beginning (default: auto-detect) - delimiter: Column delimiter (default: auto-detect) - sample_frequency: Sampling frequency in Hz (required if no time column) - channel_types: Dict mapping column names to channel types ('EMG', 'ACC', etc.) - physical_dimensions: Dict mapping column names to physical dimensions - metadata: Dict of additional metadata to include

Returns: Recording: Recording object containing the loaded data

Raises: ValueError: If a specialized format is detected and force_generic is False FileNotFoundError: If the file does not exist

Source code in biosigio/importers/csv.py

def load(self, filepath: str, force_generic: bool = False, **kwargs) -> Recording:
    """
    Load EMG data from a CSV file.

    Args:
        filepath: Path to the CSV file
        force_generic: If True, forces using the generic CSV importer even if a
                      specialized format is detected
        **kwargs: Additional options including:
            - columns: List of column names or indices to include
            - time_column: Name or index of column to use as time index (default: auto-detect)
            - has_header: Whether file has a header row (default: auto-detect)
            - skiprows: Number of rows to skip at the beginning (default: auto-detect)
            - delimiter: Column delimiter (default: auto-detect)
            - sample_frequency: Sampling frequency in Hz (required if no time column)
            - channel_types: Dict mapping column names to channel types ('EMG', 'ACC', etc.)
            - physical_dimensions: Dict mapping column names to physical dimensions
            - metadata: Dict of additional metadata to include

    Returns:
        Recording: Recording object containing the loaded data

    Raises:
        ValueError: If a specialized format is detected and force_generic is False
        FileNotFoundError: If the file does not exist
    """
    # Check if this file matches a specialized format
    if not force_generic:
        format_name = self._detect_specialized_format(filepath)
        if format_name:
            importer_messages = {
                "trigno": (
                    "This file appears to be a Delsys Trigno CSV export. "
                    "For better metadata extraction and channel detection, use:\n\n"
                    "recording = Recording.from_file(filepath, importer='trigno')\n\n"
                    "If you still want to use the generic CSV importer, set force_generic=True:\n"
                    "importer = CSVImporter()\n"
                    "recording = importer.load(filepath, force_generic=True, **params)"
                )
                # Add more format-specific messages here as new importers are developed
            }

            if format_name in importer_messages:
                raise ValueError(importer_messages[format_name])

    # Extract kwargs with defaults
    columns = kwargs.get("columns", None)
    time_column = kwargs.get("time_column", None)
    has_header = kwargs.get("has_header", None)
    skiprows = kwargs.get("skiprows", None)
    delimiter = kwargs.get("delimiter", None)
    sample_frequency = kwargs.get("sample_frequency", None)
    channel_names = kwargs.get("channel_names", [])
    channel_types = kwargs.get("channel_types", {})
    physical_dimensions = kwargs.get("physical_dimensions", {})
    metadata = kwargs.get("metadata", {})

    # Analyze file structure if parameters not explicitly provided
    try:
        if any(param is None for param in [has_header, skiprows, delimiter]):
            analyzed_params = self._analyze_csv_structure(filepath)

            # Use analyzed parameters for any not explicitly provided
            has_header = has_header if has_header is not None else analyzed_params["has_header"]
            skiprows = skiprows if skiprows is not None else analyzed_params["skiprows"]
            delimiter = delimiter if delimiter is not None else analyzed_params["delimiter"]
    except FileNotFoundError:
        # Pass through file not found errors
        raise

    # Read the CSV file
    try:
        df = pd.read_csv(
            filepath,
            header=0 if has_header else None,
            skiprows=skiprows,
            delimiter=delimiter,
            index_col=None,
        )
    except FileNotFoundError:
        # Pass through file not found errors
        raise
    except Exception as e:
        raise ValueError(f"Failed to read CSV file: {str(e)}") from e

    # If no header, generate column names
    if not has_header:
        df.columns = [f"Channel_{i}" for i in range(len(df.columns))]

    # If channel names are provided, use them.
    # Also, handle the case where the length of channel_names is less than the number of columns.
    if channel_names:
        if len(channel_names) < len(df.columns):
            raise ValueError(
                "Number of channel names provided is less than the number of columns in the CSV file."
            )
        df.columns = channel_names

    # Filter columns if specified
    if columns is not None:
        if all(isinstance(col, int) for col in columns):
            # Convert numerical indices to column names
            col_names = [df.columns[i] for i in columns]
            # Save original columns for potential renumbering
            df = df[col_names]

            # If using default channel names, renumber them sequentially
            if not has_header and not channel_names:
                # Check if these are auto-generated channel names
                if all(col.startswith("Channel_") for col in col_names):
                    # Rename columns to be sequential
                    new_names = [f"Channel_{i}" for i in range(len(col_names))]
                    rename_map = dict(zip(col_names, new_names, strict=True))
                    df = df.rename(columns=rename_map)
        else:
            # Filter by column names
            df = df[columns]

    # Handle time column
    if time_column is not None:
        # If time_column is an index, convert to column name
        if isinstance(time_column, int):
            time_column = df.columns[time_column]

        # Set time column as index
        if time_column in df.columns:
            df.set_index(time_column, inplace=True)
        else:
            raise ValueError(f"Time column '{time_column}' not found in data")
    elif has_header:
        # When header exists, try to auto-detect time column only if has_header is True
        time_col = self._detect_time_column(df)
        if time_col:
            df.set_index(time_col, inplace=True)
        elif sample_frequency:
            # Create time index based on provided sampling frequency
            time_index = np.arange(len(df)) / sample_frequency
            df.index = time_index
        else:
            # No time column and no sample frequency provided
            raise ValueError(
                "No time column detected and no sample_frequency provided. "
                "Please specify either time_column or sample_frequency."
            )
    else:
        # For headerless data, don't attempt to auto-detect time column
        # to avoid treating the first column as time
        if sample_frequency:
            # Create time index based on provided sampling frequency
            time_index = np.arange(len(df)) / sample_frequency
            df.index = time_index
        else:
            # No sample frequency provided
            raise ValueError(
                "No sample_frequency provided for headerless data. "
                "Please specify sample_frequency for proper time indexing."
            )

    # Create recording object
    rec = Recording()

    # Add metadata
    rec.set_metadata("source_file", filepath)
    rec.set_metadata("file_format", "CSV")

    # Add any user-provided metadata
    for key, value in metadata.items():
        rec.set_metadata(key, value)

    # Default sampling frequency if not specified
    default_sample_frequency = 1000.0  # 1 kHz is a common default for EMG
    if hasattr(df.index, "to_series"):
        # Calculate sampling frequency from time index if possible
        try:
            time_diffs = df.index.to_series().diff().dropna()
            if len(time_diffs) > 0:
                avg_diff = time_diffs.mean()
                if avg_diff > 0:
                    calculated_freq = 1.0 / avg_diff
                    default_sample_frequency = calculated_freq
        except Exception:
            # If calculation fails, keep default
            pass

    # Add each column as a channel
    for column in df.columns:
        # Determine channel type
        if column in channel_types:
            ch_type = channel_types[column]
        else:
            # Try to infer channel type from name
            ch_type = self._infer_channel_type(column)

        # Determine physical dimension
        if column in physical_dimensions:
            phys_dim = physical_dimensions[column]
        else:
            # Default based on channel type
            phys_dim = self._default_physical_dimension(ch_type)

        # Add the channel to the recording
        rec.add_channel(
            label=column,
            data=df[column].values,
            sample_frequency=sample_frequency or default_sample_frequency,
            physical_dimension=phys_dim,
            channel_type=ch_type,
        )

    # Encourage user to add metadata if missing essential information
    self._print_metadata_reminder(rec)

    return rec

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"}:
            return "meg"
        elif extension in {".vhdr"}:
            return "brainvision"
        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 ValueError(f"Unsupported file extension: {extension}")

    @classmethod
    def from_file(
        cls,
        filepath: str,
        importer: ImporterName | None = None,
        force_csv: bool = False,
        bids_channels: str = "auto",
        **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 and CTF .ds; requires the 'meg' extra)
                - 'brainvision': BrainVision .vhdr via MNE (requires 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: When 'auto' (default), look for a sibling BIDS
                      _channels.tsv next to the file and apply its per-channel
                      type/units over the importer's inferred values. Pass 'off'
                      to disable.
            **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)
            "brainvision": "BrainVisionImporter",  # BrainVision via MNE (.vhdr)
            "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)\n"
                "- brainvision: BrainVision via MNE (.vhdr)\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

        # 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".
        if bids_channels != "off":
            from ..bids import apply_channels_tsv, find_channels_tsv

            channels_tsv = find_channels_tsv(filepath)
            if channels_tsv:
                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``, chunk/shard sizing, ...).

        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', **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 and CTF .ds; requires the 'meg' extra) - 'brainvision': BrainVision .vhdr via MNE (requires 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: When 'auto' (default), look for a sibling BIDS _channels.tsv next to the file and apply its per-channel type/units over the importer's inferred values. Pass 'off' to disable. **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 = "auto",
    **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 and CTF .ds; requires the 'meg' extra)
            - 'brainvision': BrainVision .vhdr via MNE (requires 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: When 'auto' (default), look for a sibling BIDS
                  _channels.tsv next to the file and apply its per-channel
                  type/units over the importer's inferred values. Pass 'off'
                  to disable.
        **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)
        "brainvision": "BrainVisionImporter",  # BrainVision via MNE (.vhdr)
        "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)\n"
            "- brainvision: BrainVision via MNE (.vhdr)\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

    # 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".
    if bids_channels != "off":
        from ..bids import apply_channels_tsv, find_channels_tsv

        channels_tsv = find_channels_tsv(filepath)
        if channels_tsv:
            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, chunk/shard sizing, ...).

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``, chunk/shard sizing, ...).

    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)

Usage Example

from biosigio import Recording
from biosigio.importers.csv import CSVImporter

# Method 1: Using Recording.from_file (recommended)
rec = Recording.from_file('data.csv', importer='csv')

# Method 2: Using the importer directly
rec = CSVImporter().load('data.csv', has_header=True, delimiter=',')

Auto-Detection Features

The CSV importer includes several auto-detection capabilities:

• Format Detection: Recognizes specialized formats like Trigno CSV files
• Delimiter Detection: Identifies the most common delimiter (comma, tab, semicolon)
• Header Detection: Determines if the first row is a header based on content
• Time Column Detection: Looks for columns that might represent time

Parameters

CSVImporter().load(filepath, force_generic=False, **kwargs) takes:

• filepath (str): Path to the CSV file.
• force_generic (bool, optional): Force using the generic CSV importer even if a specialized format (e.g., Trigno) is detected. When loading through Recording.from_file(..., importer='csv', force_csv=True), the force_csv argument is forwarded as this force_generic flag.
• kwargs: Additional keyword arguments:
• sample_frequency (float, optional): Sampling frequency in Hz (required if no time column)
• has_header (bool, optional): Whether file has a header row (auto-detected if not specified)
• skiprows (int, optional): Number of rows to skip at beginning (auto-detected if not specified)
• delimiter (str, optional): Column delimiter (auto-detected if not specified)
• time_column (str or int, optional): Name or index of column to use as time index (auto-detected if not specified)
• columns (list, optional): List of column names or indices to include
• channel_names (list, optional): Custom names for channels
• channel_types (dict, optional): Dict mapping column names to channel types ('EMG', 'ACC', etc.)
• physical_dimensions (dict, optional): Dict mapping column names to physical dimensions
• metadata (dict, optional): Dict of additional metadata to include

Return Value

The load() method returns a single Recording object with:

1. signals: signal data with channels as columns and time as index.
2. channels: per-channel information including:
3. channel_type: type of channel (EMG, ACC, GYRO, MISC, OTHER), inferred from the column name when not provided
4. physical_dimension: physical unit (e.g., 'µV', 'g')
5. sample_frequency: sampling rate in Hz
6. metadata: includes source_file, file_format ('CSV'), and any additional metadata passed via the metadata keyword argument.

Implementation Details

The CSV importer uses pandas to:

1. Detect the format and structure of the CSV file
2. Extract time information if available or generate a time index based on sample frequency
3. Convert column data to appropriate formats
4. Apply channel labeling and typing based on provided information
5. Construct a pandas DataFrame with the signal data

Examples

Basic CSV with Headers

# Load CSV with automatic format detection
rec = Recording.from_file('data.csv', importer='csv')

Headerless CSV with Custom Names

# Load headerless CSV with custom channel names
rec = Recording.from_file('data.csv', importer='csv',
                   has_header=False,
                   sample_frequency=1000,  # Required since no time column
                   channel_names=['EMG_L', 'EMG_R', 'ACC_X'])

Setting Channel Types and Units

# Specify channel types and physical dimensions
rec = Recording.from_file('data.csv', importer='csv',
                   channel_types={
                       'EMG1': 'EMG',
                       'EMG2': 'EMG',
                       'ACC1': 'ACC'
                   },
                   physical_dimensions={
                       'EMG1': 'mV',
                       'EMG2': 'mV',
                       'ACC1': 'g'
                   })