objects

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

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

Experiment dataclass

Experiment(
    name: str,
    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

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
)

experiment_endtime class-attribute instance-attribute

experiment_endtime: Timestamp | None = None

experiment_starttime class-attribute instance-attribute

experiment_starttime: Timestamp | None = None

log class-attribute instance-attribute

log: DataFrame | None = None

log_data property

log_data: DataFrame

Get log data, raising an error if not available

name instance-attribute

name: str

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_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