objects

SCHEMA_CHANNEL module-attribute

SCHEMA_CHANNEL = 'chromstream-channel/v0.1.0'

SCHEMA_CHROMATOGRAM module-attribute

SCHEMA_CHROMATOGRAM = 'chromstream-chromatogram/v0.1.0'

SCHEMA_EXPERIMENT module-attribute

SCHEMA_EXPERIMENT = 'chromstream-experiment/v0.1.0'

ChannelChromatograms dataclass

ChannelChromatograms(
    channel: str,
    chromatograms: dict[int, Chromatogram] = dict(),
    integrals: DataFrame | None = None,
)

Contains data of a single channel with multiple chromatograms

Parameters:

• channel (str) –

  Name of the channel (e.g., 'FID', 'TCD')

• chromatograms (dict[int, Chromatogram], default: dict() ) –

  Dictionary mapping chromatogram number to Chromatogram objects

• integrals (DataFrame | None, default: None ) –

  DataFrame containing integrated peak areas for each chromatogram (optional)

Methods:

• add_chromatogram –

  Add a chromatogram to the channel

• plot –

  Plot all chromatograms in the channel

• integrate_peaks –

  Integrate peaks for all chromatograms in the channel, requieres dict of peak limits

channel instance-attribute

channel: str

chromatograms class-attribute instance-attribute

chromatograms: dict[int, Chromatogram] = field(
    default_factory=dict
)

integrals class-attribute instance-attribute

integrals: DataFrame | None = None

plot_chromatograms class-attribute instance-attribute

plot_chromatograms = plot

add_chromatogram

add_chromatogram(
    injection_num: int, chromatogram: Chromatogram
)

Add a chromatogram for a specific injection

Source code in src/chromstream/objects.py

def add_chromatogram(self, injection_num: int, chromatogram: Chromatogram):
    """Add a chromatogram for a specific injection"""
    self.chromatograms[injection_num] = chromatogram

apply_baseline

apply_baseline(
    correction_func,
    inplace=False,
    suffix="_BLcorr",
    **kwargs,
)

Apply baseline correction to all chromatograms in the channel

Parameters:

• correction_func –

  Function that takes a pandas DataFrame and returns corrected Series

• inplace (bool, default: False ) –

  If True, modify the original data. If False, add new column

• suffix (str, default: '_BLcorr' ) –

  Suffix to add to the new column name when inplace=False

• **kwargs –

  Additional arguments to pass to the correction function

Returns:

• None –

  Modifies chromatograms in place

Source code in src/chromstream/objects.py

def apply_baseline(
    self, correction_func, inplace=False, suffix="_BLcorr", **kwargs
):
    """
    Apply baseline correction to all chromatograms in the channel

    Args:
        correction_func: Function that takes a pandas DataFrame and returns corrected Series
        inplace (bool): If True, modify the original data. If False, add new column
        suffix (str): Suffix to add to the new column name when inplace=False
        **kwargs: Additional arguments to pass to the correction function

    Returns:
        None: Modifies chromatograms in place
    """
    for chrom in self.chromatograms.values():
        chrom.apply_baseline(
            correction_func, inplace=inplace, suffix=suffix, **kwargs
        )

integrate_peaks

integrate_peaks(
    peaklist: dict, column: None | str = None
) -> DataFrame

Integrate peaks for all chromatograms in the channel

Parameters:

• peaklist (dict) –

  Dictionary defining the peaks to integrate. Example:

• Peaks_TCD = {"N2" –

  [20, 26], "H2": [16, 19]}

• column (None | str, default: None ) –

  Optional column name to use for integration. If None, uses second column.

Returns:

• DataFrame –

  DataFrame with integrated peak areas for each injection

Source code in src/chromstream/objects.py

def integrate_peaks(
    self, peaklist: dict, column: None | str = None
) -> pd.DataFrame:
    """
    Integrate peaks for all chromatograms in the channel

    Args:
        peaklist: Dictionary defining the peaks to integrate. Example:
        Peaks_TCD = {"N2": [20, 26], "H2": [16, 19]}

        The list values must be in the same unit as the chromatogram.

        column: Optional column name to use for integration. If None, uses second column.

    Returns:
        DataFrame with integrated peak areas for each injection
    """
    self.integrals = integrate_channel(self, peaklist, column=column)
    return self.integrals

plot

plot(
    ax=None,
    colormap="viridis",
    plot_colorbar=True,
    **kwargs,
)

Plotting all chromatograms of a channel channel

Source code in src/chromstream/objects.py

def plot(self, ax=None, colormap="viridis", plot_colorbar=True, **kwargs):
    """Plotting all chromatograms of a channel channel"""
    if ax is None:
        fig, ax = plt.subplots()
    colormap = plt.get_cmap(colormap)
    colors = colormap(np.linspace(0, 1, len(self.chromatograms)))

    for inj_num, chrom in self.chromatograms.items():
        ax.plot(
            chrom.data[chrom.data.columns[0]],
            chrom.data[chrom.data.columns[1]],
            label=f"Injection {inj_num}",
            color=colors[inj_num],
            **kwargs,
        )

    # Set labels and title (handle empty channel case)
    if len(self.chromatograms) > 0:
        # Use any chromatogram to get column names
        sample_chrom = next(iter(self.chromatograms.values()))
        time_unit = sample_chrom.time_unit
        signal_unit = sample_chrom.signal_unit
        ax.set_xlabel(f"Time ({time_unit})" if time_unit != "unknown" else "Time")
        ax.set_ylabel(
            f"Signal ({signal_unit})" if signal_unit != "unknown" else "Signal"
        )
    else:
        ax.set_xlabel("Time")
        ax.set_ylabel("Signal")
    ax.set_title(f"Channel: {self.channel}")
    # add colorbar
    if plot_colorbar:
        sm = plt.cm.ScalarMappable(
            norm=Normalize(vmin=0, vmax=len(self.chromatograms) - 1)
        )
        sm.set_array([])
        cbar = plt.colorbar(sm, ax=ax)
        cbar.set_label("Injection Number")

    return ax

to_hdf5

to_hdf5(
    path: Path | str,
    *,
    overwrite: bool = False,
    compression: str | None = "gzip",
) -> Path

Write this channel to a standalone HDF5 file.

Parameters:

• path (Path | str) –

  The path to the HDF5 file to write.

• overwrite (bool, default: False ) –

  If True, overwrite the file if it exists.

• compression (str | None, default: 'gzip' ) –

  Dataset compression ("gzip", "lzf", or None).

Source code in src/chromstream/objects.py

def to_hdf5(
    self,
    path: Path | str,
    *,
    overwrite: bool = False,
    compression: str | None = "gzip",
) -> Path:
    """Write this channel to a standalone HDF5 file.

    Args:
        path: The path to the HDF5 file to write.
        overwrite: If True, overwrite the file if it exists.
        compression: Dataset compression ("gzip", "lzf", or None).
    """
    from .writers.hdf5_writer import write_channel_hdf5

    return write_channel_hdf5(
        self, path, overwrite=overwrite, compression=compression
    )

Chromatogram dataclass

Chromatogram(
    data: DataFrame,
    injection_time: Timestamp,
    metadata: dict,
    channel: str,
    path: Path | str | None,
)

Single chromatogram data for one injection on one channel

channel instance-attribute

channel: str

data instance-attribute

data: DataFrame

injection_time instance-attribute

injection_time: Timestamp

metadata instance-attribute

metadata: dict

path instance-attribute

path: Path | str | None

signal_unit property

signal_unit: str

time_unit property

time_unit: str

Get the time unit from metadata, default to 'min' if not found

apply_baseline

apply_baseline(
    correction_func,
    inplace=False,
    suffix="_BLcorr",
    **kwargs,
)

Apply baseline correction to the chromatogram data

Parameters:

• correction_func –

  Function that takes a pandas DataFrame and returns corrected Series

• inplace (bool, default: False ) –

  If True, modify the original data. If False, add new column

• suffix (str, default: '_BLcorr' ) –

  Suffix to add to the new column name when inplace=False

Returns:

• –

  pd.DataFrame: The corrected data (same as self.data if inplace=True)

Source code in src/chromstream/objects.py

def apply_baseline(
    self, correction_func, inplace=False, suffix="_BLcorr", **kwargs
):
    """
    Apply baseline correction to the chromatogram data

    Args:
        correction_func: Function that takes a pandas DataFrame and returns corrected Series
        inplace (bool): If True, modify the original data. If False, add new column
        suffix (str): Suffix to add to the new column name when inplace=False

    Returns:
        pd.DataFrame: The corrected data (same as self.data if inplace=True)
    """
    signal_column = self.data.columns[1]  # Second column (signal data)

    # Apply the correction function - passes entire DataFrame
    corrected_signal = correction_func(self.data, **kwargs)

    if inplace:
        self.data[signal_column] = corrected_signal
    else:
        new_column_name = signal_column + suffix
        self.data[new_column_name] = corrected_signal

    return self.data

integrate_peaks

integrate_peaks(
    peaklist: dict, column: None | str = None
) -> dict

Integrate peaks for this chromatogram

Parameters:

• peaklist (dict) –

  Dictionary defining the peaks to integrate. Example:

• Peaks_TCD = {"N2" –

  [20, 26], "H2": [16, 19]}

• column (None | str, default: None ) –

  Optional column name to use for integration. If None, uses second column.

Returns:

• dict –

  Dictionary with integrated peak areas and timestamp

Source code in src/chromstream/objects.py

def integrate_peaks(self, peaklist: dict, column: None | str = None) -> dict:
    """
    Integrate peaks for this chromatogram

    Args:
        peaklist: Dictionary defining the peaks to integrate. Example:
        Peaks_TCD = {"N2": [20, 26], "H2": [16, 19]}
        The list values must be in the same unit as the chromatogram.
        column: Optional column name to use for integration. If None, uses second column.

    Returns:
        Dictionary with integrated peak areas and timestamp
    """
    from .data_processing import integrate_single_chromatogram

    return integrate_single_chromatogram(self, peaklist, column=column)

plot

plot(ax=None, column=None, **kwargs)

Plot the chromatogram data

Source code in src/chromstream/objects.py

def plot(self, ax=None, column=None, **kwargs):
    """Plot the chromatogram data"""
    if ax is None:
        fig, ax = plt.subplots()

    # Choose which column to plot (default to second column)
    y_column = self.data.columns[1] if column is None else column
    x_column = self.data.columns[0]
    time_unit = self.time_unit
    signal_unit = self.signal_unit

    ax.plot(self.data[x_column], self.data[y_column], **kwargs)
    ax.set_xlabel(f"Time ({time_unit})" if time_unit != "unknown" else "Time")
    ax.set_ylabel(
        f"Signal ({signal_unit})" if signal_unit != "unknown" else "Signal"
    )
    ax.set_title(f"Chromatogram - {self.channel} - {self.path}")

    return ax

to_hdf5

to_hdf5(
    path: Path | str,
    *,
    injection_index: int | None = None,
    overwrite: bool = False,
    compression: str | None = "gzip",
) -> Path

Write this chromatogram to a standalone HDF5 file.

Parameters:

• path (Path | str) –

  The path to the HDF5 file to write.

• injection_index (int | None, default: None ) –

  Optional channel-level injection index to persist so the chromatogram can be reassembled into an Experiment losslessly.

• overwrite (bool, default: False ) –

  If True, overwrite the file if it exists.

• compression (str | None, default: 'gzip' ) –

  Dataset compression ("gzip", "lzf", or None).

Source code in src/chromstream/objects.py

def to_hdf5(
    self,
    path: Path | str,
    *,
    injection_index: int | None = None,
    overwrite: bool = False,
    compression: str | None = "gzip",
) -> Path:
    """Write this chromatogram to a standalone HDF5 file.

    Args:
        path: The path to the HDF5 file to write.
        injection_index: Optional channel-level injection index to persist so
            the chromatogram can be reassembled into an Experiment losslessly.
        overwrite: If True, overwrite the file if it exists.
        compression: Dataset compression ("gzip", "lzf", or None).
    """
    from .writers.hdf5_writer import write_chromatogram_hdf5

    return write_chromatogram_hdf5(
        self,
        path,
        injection_index=injection_index,
        overwrite=overwrite,
        compression=compression,
    )

Experiment dataclass

Experiment(
    name: str,
    schema: str = SCHEMA_EXPERIMENT,
    author: str | None = None,
    creation_date: Timestamp | None = None,
    metadata: dict[str, object] = dict(),
    channels: dict[str, ChannelChromatograms] = dict(),
    experiment_starttime: Timestamp | None = None,
    experiment_endtime: Timestamp | None = None,
    log: DataFrame | None = None,
)

Data for a single experiment containing multiple on-line GC channels

author class-attribute instance-attribute

author: str | None = None

channel_names property

channel_names: list[str]

Get a list of channel names in the experiment

channels class-attribute instance-attribute

channels: dict[str, ChannelChromatograms] = field(
    default_factory=dict
)

creation_date class-attribute instance-attribute

creation_date: Timestamp | None = None

experiment_endtime class-attribute instance-attribute

experiment_endtime: Timestamp | None = None

experiment_starttime class-attribute instance-attribute

experiment_starttime: Timestamp | None = None

label property

label: str

Alias for the experiment name when persisted to file formats.

log class-attribute instance-attribute

log: DataFrame | None = None

log_data property

log_data: DataFrame

Get log data, raising an error if not available

metadata class-attribute instance-attribute

metadata: dict[str, object] = field(default_factory=dict)

name instance-attribute

name: str

schema class-attribute instance-attribute

schema: str = SCHEMA_EXPERIMENT

add_channel

add_channel(
    channel_name: str, channel_data: ChannelChromatograms
)

Add a channel to the experiment

Source code in src/chromstream/objects.py

def add_channel(self, channel_name: str, channel_data: ChannelChromatograms):
    """Add a channel to the experiment"""
    self.channels[channel_name] = channel_data

add_chromatogram

add_chromatogram(
    chromatogram: Path | str | Chromatogram,
    channel_name: str | None = None,
)

Add a chromatogram to the experiment, automatically creating the channel if it does not exist

Parameters:

• chromatogram (Path | str | Chromatogram) –

  Path to the chromatogram file or a Chromatogram object

• channel_name (Optional[str], default: None ) –

  Optional channel name to override

Source code in src/chromstream/objects.py

def add_chromatogram(
    self, chromatogram: Path | str | Chromatogram, channel_name: str | None = None
):
    """Add a chromatogram to the experiment, automatically creating the channel if it does not exist

    Args:
        chromatogram (Path | str | Chromatogram): Path to the chromatogram file or a Chromatogram object
        channel_name (Optional[str], optional): Optional channel name to override


    """
    if isinstance(chromatogram, (str, Path)):
        path = Path(chromatogram)
        if path.suffix.lower() == ".ch":
            from .parsers import parse_agilent_ch

            chrom = parse_agilent_ch(chromatogram)
        else:
            from .parsers import parse_chromatogram_txt

            try:
                chrom = parse_chromatogram_txt(chromatogram)
            except Exception as e:
                raise ValueError(f"Failed to parse chromatogram: {e}")
    elif isinstance(chromatogram, Chromatogram):
        chrom = chromatogram
    else:
        raise ValueError(
            "chromatogram must be a file path or a Chromatogram object"
        )

    channel = channel_name if channel_name else chrom.channel

    if channel not in self.channels:
        self.channels[channel] = ChannelChromatograms(channel=channel)

    injection_num = len(self.channels[channel].chromatograms)
    self.channels[channel].add_chromatogram(injection_num, chrom)

add_chromatogram_hdf5

add_chromatogram_hdf5(
    path: Path | str, *, channel_name: str | None = None
) -> None

Add a chromatogram from a standalone HDF5 file into this experiment.

Note on injection index: a standalone chromatogram file may store the injection_index it had within its channel. When that attr is present we restore the chromatogram under that exact key, so writing each chromatogram of an experiment to its own file and reassembling them round-trips the original (possibly non-contiguous) injection numbering regardless of the order the files are loaded in. When the attr is absent (e.g. a hand-made file) we fall back to sequential numbering, len(existing), matching add_chromatogram's behavior.

Parameters:

• path (Path | str) –

  Path to a chromatogram HDF5 file written by ChromStream.

• channel_name (str | None, default: None ) –

  Optional channel name override; defaults to the channel stored on the chromatogram.

Source code in src/chromstream/objects.py

def add_chromatogram_hdf5(
    self, path: Path | str, *, channel_name: str | None = None
) -> None:
    """Add a chromatogram from a standalone HDF5 file into this experiment.

    Note on injection index: a standalone chromatogram file may
    store the ``injection_index`` it had within its channel. When that attr
    is present we restore the chromatogram under that exact key, so writing
    each chromatogram of an experiment to its own file and reassembling them
    round-trips the original (possibly non-contiguous) injection numbering
    regardless of the order the files are loaded in. When the attr is absent
    (e.g. a hand-made file) we fall back to sequential numbering,
    ``len(existing)``, matching ``add_chromatogram``'s behavior.

    Args:
        path: Path to a chromatogram HDF5 file written by ChromStream.
        channel_name: Optional channel name override; defaults to the
            channel stored on the chromatogram.
    """
    from .parsers.hdf5 import _parse_chromatogram_hdf5_with_index

    injection_index, chrom = _parse_chromatogram_hdf5_with_index(path)
    channel = channel_name if channel_name else chrom.channel
    if channel not in self.channels:
        self.channels[channel] = ChannelChromatograms(channel=channel)

    if injection_index is None:
        injection_index = len(self.channels[channel].chromatograms)
    self.channels[channel].add_chromatogram(injection_index, chrom)

add_log

add_log(log: str | Path | DataFrame)

Adds a log dataframe to the experiment, either from a dataframe or from a path to the log file.

Parameters:

• log (str | Path | DataFrame) –

  Path to the log file or a DataFrame

Source code in src/chromstream/objects.py

def add_log(self, log: str | Path | pd.DataFrame):
    """
    Adds a log dataframe to the experiment, either from a dataframe or from a path to the log file.

    Args:
        log (str | Path | pd.DataFrame): Path to the log file or a DataFrame
    """
    if isinstance(log, (str, Path)):
        from .parsers import parse_log_file

        self.log = parse_log_file(log)
    elif isinstance(log, pd.DataFrame):
        self.log = log
    else:
        raise ValueError("log must be a file path or a DataFrame")

add_mult_chromatograms

add_mult_chromatograms(
    chromatograms: list[Path | str | Chromatogram]
    | Path
    | str,
    channel_name: str | None = None,
)

Add multiple chromatograms to the experiment

Parameters:

• chromatograms (list[Path | str | Chromatogram] | Path | str) –

  Either: - A list of Chromatogram objects - A list of paths to chromatogram files - A path to a .d directory containing .ch files - A path to a .dx archive containing .ch files

• channel_name (str | None, default: None ) –

  Optional channel name to override for all chromatograms

Source code in src/chromstream/objects.py

def add_mult_chromatograms(
    self,
    chromatograms: list[Path | str | Chromatogram] | Path | str,
    channel_name: str | None = None,
):
    """Add multiple chromatograms to the experiment

    Args:
        chromatograms: Either:
            - A list of Chromatogram objects
            - A list of paths to chromatogram files
            - A path to a .d directory containing .ch files
            - A path to a .dx archive containing .ch files
        channel_name: Optional channel name to override for all chromatograms
    """
    # Convert single path to Path object if needed
    if isinstance(chromatograms, (str, Path)):
        path = Path(chromatograms)
        # Check if it's a .d directory
        if path.is_dir() and path.name.lower().endswith(".d"):
            from .parsers.agilent import parse_agilent_dot_d

            chrom_list = parse_agilent_dot_d(path)
        # Check if it's a .dx archive
        elif path.is_file() and path.suffix.lower() == ".dx":
            from .parsers.agilent import parse_agilent_dx

            chrom_list = parse_agilent_dx(path)
        else:
            raise ValueError(
                f"Path must be a .d directory or .dx file. Got: {chromatograms}"
            )

        for chrom in chrom_list:
            self.add_chromatogram(chrom, channel_name=channel_name)
    elif isinstance(chromatograms, list):
        # Handle list of chromatograms or paths
        for chrom in chromatograms:
            self.add_chromatogram(chrom, channel_name=channel_name)
    else:
        raise ValueError(
            "chromatograms must be a list of Chromatogram objects/paths or a path to a .d directory/.dx file"
        )

plot_chromatograms

plot_chromatograms(
    ax=None, channels: str | list = "all", **kwargs
)

Source code in src/chromstream/objects.py

def plot_chromatograms(self, ax=None, channels: str | list = "all", **kwargs):
    if ax is None:
        n_channels_to_plot = (
            len(self.channels) if channels == "all" else len(channels)
        )

        # Handle empty experiment case
        if n_channels_to_plot == 0:
            fig, ax = plt.subplots()
            ax.text(
                0.5,
                0.5,
                "No channels to plot",
                ha="center",
                va="center",
                transform=ax.transAxes,
            )
            ax.set_title("Empty Experiment")
            return

        fig, ax = plt.subplots(
            n_channels_to_plot,
            1,
            # figsize=(7, 3.3 / 1.618 * n_channels_to_plot),
            tight_layout=True,
        )
        if n_channels_to_plot == 1:
            ax = [ax]
    if channels == "all":
        channels = list(self.channels.keys())
    for i, channel in enumerate(channels):
        if channel in self.channels:
            self.channels[channel].plot(ax=ax[i], **kwargs)
        else:
            raise ValueError(f"Channel {channel} not found in experiment.")

plot_log

plot_log(columns: str | list, ax=None, use_exp_time=False)

Plots specified colums of the experiment log. If use_exp_time is True, the x-axis will be the time since the start of the experiment in minutes. Args: columns (str | list): Column name or list of column names to plot ax (matplotlib.axes.Axes, optional): Axes to plot on. If None, a new figure and axes will be created. use_exp_time (bool, optional): Whether to use time since start of experiment as x-axis. Defaults to False.

Source code in src/chromstream/objects.py

def plot_log(self, columns: str | list, ax=None, use_exp_time=False):
    """
    Plots specified colums of the experiment log. If use_exp_time is True, the x-axis will be the time since the start of the experiment in minutes.
    Args:
        columns (str | list): Column name or list of column names to plot
        ax (matplotlib.axes.Axes, optional): Axes to plot on. If None, a new figure and axes will be created.
        use_exp_time (bool, optional): Whether to use time since start of experiment as x-axis. Defaults to False.
    """

    if self.log is None:
        raise ValueError("No log data available to plot.")

    if ax is None:
        fig, ax = plt.subplots()

    if isinstance(columns, str):
        columns = [columns]

    if use_exp_time:
        if self.experiment_starttime is None:
            raise ValueError(
                "Experiment start time is not set. Cannot use experiment time."
            )
        x = (
            pd.to_datetime(self.log["Timestamp"]) - self.experiment_starttime
        ).dt.total_seconds() / 60.0
        x_label = "Experiment Time (min)"
    else:
        x = self.log["Timestamp"]
        x_label = "Timestamp"

    for col in columns:
        if col not in self.log.columns:
            raise ValueError(f"Column {col} not found in log data.")
        ax.plot(x, self.log[col], label=col)

    ax.set_xlabel(x_label)
    ax.set_ylabel("Value")
    ax.set_title("Experiment Log Data")
    ax.legend()

    return ax

to_hdf5

to_hdf5(
    path: Path | str,
    *,
    overwrite: bool = False,
    compression: str | None = "gzip",
) -> Path

Write this experiment to a single HDF5 file. Args: path: The path to the HDF5 file to write. overwrite: If True, overwrite the file if it exists. compression: The compression algorithm to use for datasets. Available options include "gzip", "lzf", or None for no compression. Compression can reduce file size but may increase read/write time. Turned on by default with "gzip". Set to None to disable compression.

Source code in src/chromstream/objects.py

def to_hdf5(
    self,
    path: Path | str,
    *,
    overwrite: bool = False,
    compression: str | None = "gzip",
) -> Path:
    """Write this experiment to a single HDF5 file.
    Args:
        path: The path to the HDF5 file to write.
        overwrite: If True, overwrite the file if it exists.
        compression: The compression algorithm to use for datasets. Available options include "gzip", "lzf", or None for no compression. Compression can reduce file size but may increase read/write time. Turned on by default with "gzip". Set to None to disable compression.
    """

    from .writers.hdf5_writer import write_experiment_hdf5

    return write_experiment_hdf5(
        self, path, overwrite=overwrite, compression=compression
    )