API Reference

This page documents the full public API of lasertram, auto-generated from the source code docstrings.

LaserTRAM

The core class for processing individual LA-ICP-MS time-series analyses — background subtraction, signal selection, normalization, and output report generation.

lasertram.tram.LaserTRAM

Time resolved analysis of a single LA-ICP-MS spot.

LaserTRAM contains all the information and methods related to reducing one individual spot analysis from raw counts-per-second data to internally normalised ratios. To be used in conjunction with LaserCalc.

Parameters:

Name	Type	Description	Default
name	str	Sample name, i.e. the value in the SampleLabel column of the LT-ready file.	required

Examples:

>>> from lasertram import LaserTRAM, helpers
>>> raw_data = helpers.preprocessing.load_test_rawdata()
>>> spot = LaserTRAM(name="GSD-1G_-_1")
>>> spot.get_data(raw_data.loc["GSD-1G_-_1", :])
>>> spot.assign_int_std("29Si")
>>> spot.assign_intervals(bkgd=(5, 10), keep=(25, 40))
>>> spot.get_bkgd_data()
>>> spot.subtract_bkgd()
>>> spot.get_detection_limits()
>>> spot.normalize_interval()
>>> spot.make_output_report()

Source code in src/lasertram/tram/tram.py

class LaserTRAM:
    """Time resolved analysis of a single LA-ICP-MS spot.

    ``LaserTRAM`` contains all the information and methods related to
    reducing one individual spot analysis from raw counts-per-second
    data to internally normalised ratios. To be used in conjunction
    with ``LaserCalc``.

    Parameters
    ----------
    name : str
        Sample name, i.e. the value in the ``SampleLabel`` column of the
        LT-ready file.

    Examples
    --------
    >>> from lasertram import LaserTRAM, helpers
    >>> raw_data = helpers.preprocessing.load_test_rawdata()
    >>> spot = LaserTRAM(name="GSD-1G_-_1")
    >>> spot.get_data(raw_data.loc["GSD-1G_-_1", :])
    >>> spot.assign_int_std("29Si")
    >>> spot.assign_intervals(bkgd=(5, 10), keep=(25, 40))
    >>> spot.get_bkgd_data()
    >>> spot.subtract_bkgd()
    >>> spot.get_detection_limits()
    >>> spot.normalize_interval()
    >>> spot.make_output_report()
    """

    def __init__(self, name: str) -> None:
        """Initialise a LaserTRAM spot object.

        Parameters
        ----------
        name : str
            Sample name, i.e. the value in the ``SampleLabel`` column of
            the LT-ready file.
        """
        # all attributes in relative chronological order that they are created in
        # if everything is done correctly. These all will get rewritten throughout the
        # data processing pipeline but this allows us to see what all the potential attributes
        # are going to be from the beginning (PEP convention)

        # for the math involved please see:

        # name of the lasertram spot object
        self.name = name

        # boolean flag for whether or not the data have been
        # despiked
        self.despiked = False

        # list of elements that have been despiked. Also may be 'all'
        self.despiked_elements = None

        # data from a single spot to be processed. 2D pandas dataframe
        self.data = None

        # self.data but as a 2D numpy matrix. Equivalent to self.data.values
        self.data_matrix = None

        # list of analytes in the analysis
        self.analytes = None

        # datetime corresponding to the analysis
        self.timestamp = None

        # string representation internal standard analyte for the processing.
        # this is just the column header of the analyte chosen as the internal
        # standard e.g., "29Si"
        self.int_std = None

        # column number in self.data_matrix that denotes the internal standard analyte
        # data. Remember python starts counting at 0!
        self.int_std_loc = None

        # background interval start time
        self.bkgd_start = None

        # background interval stop time
        self.bkgd_stop = None

        # desired ablation interval start time
        self.int_start = None

        # desired ablation interval stop time
        self.int_stop = None

        # row in self.data corresponding to self.bkgd_start
        self.bkgd_start_idx = None

        # row in self.data corresponding to self.bkgd_stop
        self.bkgd_stop_idx = None

        # row in self.data corresponding to self.int_start
        self.int_start_idx = None

        # row in self.data corresponding to self.int_stop
        self.int_stop_idx = None

        # desired omitted region start time
        self.omit_start = None

        # desired omitted region stop time
        self.omit_stop = None

        # row in self.data corresponding to self.omit_start
        self.omit_start_idx = None
        # row in self.data corresponding to self.omit_stop
        self.omit_stop_idx = None

        #
        self.omitted_region = None

        # 1D array of median background values [self.bkgd_start - self.bkgd_stop)
        # that is len(analytes) in shape
        self.bkgd_data_median = None

        # 1D array of detection limits in counts per second
        # that is len(analytes) in shape
        self.detection_limits = None

        # 2D array of background corrected data over the self.int_start - self.int_stop
        # region
        self.bkgd_subtract_data = None

        # 2D array of background corrected data over the self.int_start - self.int_stop
        # region that is normalized to the internal standard
        self.bkgd_subtract_normal_data = None

        # 1D array of median background corrected normalized values over the self.int_start - self.int_stop
        # retion that is len(analytes) in shape
        self.bkgd_subtract_med = None

        # 1D array of 1 standard error of the mean values for each analyte over the
        # self.int_start - self.int_stop region
        self.bkgd_subtract_std_err = None

        #
        self.bkgd_subtract_std_err_rel = None

        # 1D pandas dataframe that contains many of the attributes created during the
        # LaserTRAM process:
        # |timestamp|Spot|despiked|omitted_region|bkgd_start|bkgd_stop|int_start|int_stop|norm|norm_cps|analyte vals and uncertainties -->|
        # |---------|----|--------|--------------|----------|---------|---------|--------|----|--------|----------------------------------|
        self.output_report = None

    def get_data(
        self, df: pd.DataFrame, time_units: str = "ms", verbose: bool = True
    ) -> None:
        """Assign raw counts-per-second data to the spot object.

        Parameters
        ----------
        df : pandas.DataFrame
            Raw data corresponding to the spot being processed, e.g.
            ``all_data.loc[spot, :]`` where *all_data* is the LT-ready
            DataFrame.
        time_units : str, optional
            Units for the ``Time`` column. Used to convert input time
            values to seconds. ``'ms'`` (default) or ``'s'``.
        verbose : bool, optional
            Whether to print status messages during data loading,
            by default ``True``.

        Examples
        --------
        >>> spot = LaserTRAM(name="GSD-1G_-_1")
        >>> spot.get_data(raw_data.loc["GSD-1G_-_1", :])
        """

        # TODO: get_data
        # - [x] add: check to make sure data are in right format else throw an error. do this by having a list of required columns and checking for them.

        # get data and set index to "SampleLabel" column
        self.data = df.reset_index()

        col_check, type_check = formatting.check_lt_input_format(self.data)
        if verbose:
            print(
                "checking LaserTRAM input data format for correct column headers and data types..."
            )

        if col_check is not None:
            raise ValueError(
                f"It looks like your input data are missing the following required columns: {col_check}. Please fix before continuing with processing."
            )
        if type_check is not None:
            raise TypeError(
                f"It looks like your input data have incorrect types in the following column indices: {type_check}. Please fix before continuing with processing."
            )

        if verbose:
            print("check complete...input data format looks good!")

        self.data = self.data.set_index("SampleLabel")

        # convert time units from ms --> s if applicable
        if time_units == "ms":
            self.data["Time"] = self.data["Time"] / 1000
        elif time_units == "s":
            pass

        # just numpy matrix for data
        self.data_matrix = self.data.iloc[:, 1:].to_numpy()

        # list of analytes in experiment
        self.analytes = self.data.loc[:, "Time":].columns.tolist()[1:]

        # need to add check for if this exists otherwise there is no timestamp attribute
        self.timestamp = str(self.data.loc[:, "timestamp"].unique()[0])
        self.timestamp = pd.to_datetime(self.timestamp)

    def assign_int_std(self, int_std: str) -> None:
        """Assign the internal standard analyte for normalisation.

        Parameters
        ----------
        int_std : str
            Column name for the internal standard analyte, e.g.
            ``"29Si"``.

        Examples
        --------
        >>> spot.assign_int_std("29Si")
        """

        # set the internal standard analyte
        self.int_std = int_std

        # get the internal standard array index
        self.int_std_loc = np.where(np.array(self.analytes) == self.int_std)[0][0]

    def assign_intervals(
        self,
        bkgd: tuple[float, float],
        keep: tuple[float, float],
        omit: tuple[float, float] | None = None,
    ) -> None:
        """Assign background, signal, and optional omission intervals.

        Parameters
        ----------
        bkgd : tuple of float
            ``(start, stop)`` pair of values (in seconds) for the
            background signal region.
        keep : tuple of float
            ``(start, stop)`` pair of values (in seconds) for the
            ablation signal region used in concentration calculations.
        omit : tuple of float or None, optional
            ``(start, stop)`` pair of values (in seconds) to be omitted
            from the ``keep`` interval, by default ``None``.

        Examples
        --------
        >>> spot.assign_intervals(bkgd=(5, 10), keep=(25, 40))

        With a region to omit:

        >>> spot.assign_intervals(bkgd=(5, 10), keep=(23, 40), omit=(30, 33))
        """

        # set background and interval times in s
        self.bkgd_start = bkgd[0]
        self.bkgd_stop = bkgd[1]
        self.int_start = keep[0]
        self.int_stop = keep[1]

        # equivalent background and interval times but as indices
        # in their respective arrays
        self.bkgd_start_idx = np.where(self.data["Time"] > self.bkgd_start)[0][0]
        self.bkgd_stop_idx = np.where(self.data["Time"] > self.bkgd_stop)[0][0]
        self.int_start_idx = np.where(self.data["Time"] > self.int_start)[0][0]
        self.int_stop_idx = np.where((self.data["Time"] > self.int_stop))[0][0]

        # boolean whether or not there is an omitted region
        self.omitted_region = False
        # if omission is true, set those start and stop times like above
        if omit:
            self.omit_start = omit[0]
            self.omit_stop = omit[1]
            self.omit_start_idx = (
                np.where(self.data["Time"] > self.omit_start)[0][0] - self.int_start_idx
            )
            self.omit_stop_idx = (
                np.where(self.data["Time"] > self.omit_stop)[0][0] - self.int_start_idx
            )

            self.omitted_region = True

    def get_bkgd_data(self) -> None:
        """Calculate median background values for all analytes.

        Uses the intervals assigned in ``assign_intervals()`` to take
        the median value of all analytes within the background range.
        These are later subtracted from the ablation signal in
        ``subtract_bkgd()``.
        """
        # median background data values
        self.bkgd_data_median = np.median(
            self.data_matrix[self.bkgd_start_idx : self.bkgd_stop_idx, 1:], axis=0
        )

    def get_detection_limits(self) -> None:
        """Calculate detection limits in counts per second.

        Defined as the median background plus three standard deviations
        of the background signal for each analyte.
        """

        self.detection_limits = np.std(
            self.data_matrix[self.bkgd_start_idx : self.bkgd_stop_idx, 1:], axis=0
        ) * 3 + np.median(
            self.data_matrix[self.bkgd_start_idx : self.bkgd_stop_idx, 1:], axis=0
        )

    def subtract_bkgd(self) -> None:
        """Subtract the median background from the ablation signal.

        Removes the median background values calculated in
        ``get_bkgd_data()`` from the signal in the *keep* interval
        established in ``assign_intervals()``.
        """
        self.bkgd_subtract_data = (
            self.data_matrix[self.int_start_idx : self.int_stop_idx, 1:]
            - self.bkgd_data_median
        )

    def normalize_interval(self) -> None:
        """Normalise the ablation interval to the internal standard.

        Divides all analytes in the *keep* portion of the signal by the
        internal standard analyte. Also calculates the median normalised
        value, its standard error of the mean, and relative standard
        error of the mean.
        """

        # set the detection limit thresholds to be checked against
        # with the interval data. This basically takes the detection limits
        threshold = self.detection_limits - self.bkgd_data_median

        # if there's an omitted region, remove it from the data to be further processed
        # for the chosen interval
        if self.omitted_region is True:
            self.bkgd_subtract_normal_data = np.delete(
                self.bkgd_subtract_data,
                np.arange(self.omit_start_idx, self.omit_stop_idx),
                axis=0,
            ) / np.delete(
                self.bkgd_subtract_data[:, self.int_std_loc][:, None],
                np.arange(self.omit_start_idx, self.omit_stop_idx),
                axis=0,
            )

        else:
            self.bkgd_subtract_normal_data = (
                self.bkgd_subtract_data
                / self.bkgd_subtract_data[:, self.int_std_loc][:, None]
            )

        # get background corrected and normalized median values for an interval
        self.bkgd_subtract_med = np.median(self.bkgd_subtract_normal_data, axis=0)
        self.bkgd_subtract_med[
            np.median(self.bkgd_subtract_data, axis=0) <= threshold
        ] = -9999
        self.bkgd_subtract_med[np.median(self.bkgd_subtract_data, axis=0) == 0] = -9999

        # standard error of the mean for the interval region
        self.bkgd_subtract_std_err = self.bkgd_subtract_normal_data.std(
            axis=0
        ) / np.sqrt(abs(self.int_stop_idx - self.int_start_idx))

        self.bkgd_subtract_std_err_rel = 100 * (
            self.bkgd_subtract_std_err / self.bkgd_subtract_med
        )

    def make_output_report(self) -> None:
        """Create an output report summarising the spot processing.

        Generates a single-row ``pandas.DataFrame`` stored in
        ``self.output_report`` with the following columns:

        ``timestamp | Spot | despiked | omitted_region | bkgd_start |
        bkgd_stop | int_start | int_stop | norm | norm_cps |``
        followed by normalised analyte values and their standard errors.

        Examples
        --------
        >>> spot.make_output_report()
        >>> spot.output_report.head()
        """
        if self.despiked is True:
            despike_col = self.despiked_elements
        else:
            despike_col = "None"

        if self.omitted_region is True:
            omitted_col = (
                self.data["Time"].iloc[self.omit_start_idx + self.int_start_idx],
                self.data["Time"].iloc[self.omit_stop_idx + self.int_start_idx],
            )
        else:
            omitted_col = "None"

        spot_data = pd.DataFrame(
            [
                self.timestamp,
                self.name,
                despike_col,
                omitted_col,
                self.data["Time"].iloc[self.bkgd_start_idx],
                self.data["Time"].iloc[self.bkgd_stop_idx],
                self.data["Time"].iloc[self.int_start_idx],
                self.data["Time"].iloc[self.int_stop_idx],
                self.int_std,
                np.median(self.bkgd_subtract_data[:, self.int_std_loc]),
            ]
        ).T
        spot_data.columns = [
            "timestamp",
            "Spot",
            "despiked",
            "omitted_region",
            "bkgd_start",
            "bkgd_stop",
            "int_start",
            "int_stop",
            "norm",
            "norm_cps",
        ]
        spot_data["timestamp"] = pd.to_datetime(spot_data["timestamp"])
        spot_data = pd.concat(
            [
                spot_data,
                pd.DataFrame(
                    self.bkgd_subtract_med[np.newaxis, :], columns=self.analytes
                ),
                pd.DataFrame(
                    self.bkgd_subtract_std_err_rel[np.newaxis, :],
                    columns=[f"{analyte}_se" for analyte in self.analytes],
                ),
            ],
            axis="columns",
        )

        for col in ["bkgd_start", "bkgd_stop", "int_start", "int_stop", "norm_cps"]:
            spot_data[col] = spot_data[col].astype(np.float64)

        self.output_report = spot_data

    def despike_data(
        self,
        analyte_list: str | list[str] = "all",
        std_devs: int = 4,
        window: int = 25,
    ) -> None:
        """Despike normalised data using a rolling z-score filter.

        Applies a z-score filter to the internally normalised data to
        remove analytical spikes. Must be called **after**
        ``normalize_interval()`` and **before**
        ``make_output_report()``.

        Parameters
        ----------
        analyte_list : str or list of str, optional
            Analytes to despike. Accepts a single analyte
            (e.g., ``"29Si"``), a list (e.g., ``["7Li", "29Si"]``),
            or ``"all"`` to despike every analyte, by default ``"all"``.
        std_devs : int, optional
            Number of standard deviations from the rolling mean to be
            considered an outlier, by default 4.
        window : int, optional
            Size of the rolling average window, by default 25.

        Examples
        --------
        Despike all analytes with default settings:

        >>> spot.despike_data()

        Despike a single analyte with custom parameters:

        >>> spot.despike_data(analyte_list="208Pb", std_devs=3, window=30)
        """

        assert (
            self.bkgd_subtract_normal_data is not None
        ), "please normalize your data prior to despiking"

        self.despiked = True

        if analyte_list == "all":
            filter_list = self.analytes
        else:
            if isinstance(analyte_list, list):
                pass
            else:
                analyte_list = [analyte_list]

            filter_list = analyte_list

        self.despiked_elements = filter_list

        df = pd.DataFrame(self.bkgd_subtract_normal_data, columns=self.analytes)

        for analyte in filter_list:

            filtered = _z_filter(df[analyte], window=window, std_devs=std_devs)

            # replaces data with despiked data
            df[analyte] = filtered

        self.bkgd_subtract_normal_data = df.loc[:, self.analytes].values

        # now recalculate uncertainties after despiking
        # standard error of the mean for the interval region
        self.bkgd_subtract_std_err = self.bkgd_subtract_normal_data.std(
            axis=0
        ) / np.sqrt(abs(self.int_stop_idx - self.int_start_idx))

        self.bkgd_subtract_std_err_rel = 100 * (
            self.bkgd_subtract_std_err / self.bkgd_subtract_med
        )

__init__(name)

Initialise a LaserTRAM spot object.

Parameters:

Name	Type	Description	Default
name	str	Sample name, i.e. the value in the SampleLabel column of the LT-ready file.	required

Source code in src/lasertram/tram/tram.py

def __init__(self, name: str) -> None:
    """Initialise a LaserTRAM spot object.

    Parameters
    ----------
    name : str
        Sample name, i.e. the value in the ``SampleLabel`` column of
        the LT-ready file.
    """
    # all attributes in relative chronological order that they are created in
    # if everything is done correctly. These all will get rewritten throughout the
    # data processing pipeline but this allows us to see what all the potential attributes
    # are going to be from the beginning (PEP convention)

    # for the math involved please see:

    # name of the lasertram spot object
    self.name = name

    # boolean flag for whether or not the data have been
    # despiked
    self.despiked = False

    # list of elements that have been despiked. Also may be 'all'
    self.despiked_elements = None

    # data from a single spot to be processed. 2D pandas dataframe
    self.data = None

    # self.data but as a 2D numpy matrix. Equivalent to self.data.values
    self.data_matrix = None

    # list of analytes in the analysis
    self.analytes = None

    # datetime corresponding to the analysis
    self.timestamp = None

    # string representation internal standard analyte for the processing.
    # this is just the column header of the analyte chosen as the internal
    # standard e.g., "29Si"
    self.int_std = None

    # column number in self.data_matrix that denotes the internal standard analyte
    # data. Remember python starts counting at 0!
    self.int_std_loc = None

    # background interval start time
    self.bkgd_start = None

    # background interval stop time
    self.bkgd_stop = None

    # desired ablation interval start time
    self.int_start = None

    # desired ablation interval stop time
    self.int_stop = None

    # row in self.data corresponding to self.bkgd_start
    self.bkgd_start_idx = None

    # row in self.data corresponding to self.bkgd_stop
    self.bkgd_stop_idx = None

    # row in self.data corresponding to self.int_start
    self.int_start_idx = None

    # row in self.data corresponding to self.int_stop
    self.int_stop_idx = None

    # desired omitted region start time
    self.omit_start = None

    # desired omitted region stop time
    self.omit_stop = None

    # row in self.data corresponding to self.omit_start
    self.omit_start_idx = None
    # row in self.data corresponding to self.omit_stop
    self.omit_stop_idx = None

    #
    self.omitted_region = None

    # 1D array of median background values [self.bkgd_start - self.bkgd_stop)
    # that is len(analytes) in shape
    self.bkgd_data_median = None

    # 1D array of detection limits in counts per second
    # that is len(analytes) in shape
    self.detection_limits = None

    # 2D array of background corrected data over the self.int_start - self.int_stop
    # region
    self.bkgd_subtract_data = None

    # 2D array of background corrected data over the self.int_start - self.int_stop
    # region that is normalized to the internal standard
    self.bkgd_subtract_normal_data = None

    # 1D array of median background corrected normalized values over the self.int_start - self.int_stop
    # retion that is len(analytes) in shape
    self.bkgd_subtract_med = None

    # 1D array of 1 standard error of the mean values for each analyte over the
    # self.int_start - self.int_stop region
    self.bkgd_subtract_std_err = None

    #
    self.bkgd_subtract_std_err_rel = None

    # 1D pandas dataframe that contains many of the attributes created during the
    # LaserTRAM process:
    # |timestamp|Spot|despiked|omitted_region|bkgd_start|bkgd_stop|int_start|int_stop|norm|norm_cps|analyte vals and uncertainties -->|
    # |---------|----|--------|--------------|----------|---------|---------|--------|----|--------|----------------------------------|
    self.output_report = None

get_data(df, time_units='ms', verbose=True)

Assign raw counts-per-second data to the spot object.

Parameters:

Name	Type	Description	Default
df	DataFrame	Raw data corresponding to the spot being processed, e.g. all_data.loc[spot, :] where all_data is the LT-ready DataFrame.	required
time_units	str	Units for the Time column. Used to convert input time values to seconds. 'ms' (default) or 's'.	'ms'
verbose	bool	Whether to print status messages during data loading, by default True.	True

Examples:

>>> spot = LaserTRAM(name="GSD-1G_-_1")
>>> spot.get_data(raw_data.loc["GSD-1G_-_1", :])

Source code in src/lasertram/tram/tram.py

def get_data(
    self, df: pd.DataFrame, time_units: str = "ms", verbose: bool = True
) -> None:
    """Assign raw counts-per-second data to the spot object.

    Parameters
    ----------
    df : pandas.DataFrame
        Raw data corresponding to the spot being processed, e.g.
        ``all_data.loc[spot, :]`` where *all_data* is the LT-ready
        DataFrame.
    time_units : str, optional
        Units for the ``Time`` column. Used to convert input time
        values to seconds. ``'ms'`` (default) or ``'s'``.
    verbose : bool, optional
        Whether to print status messages during data loading,
        by default ``True``.

    Examples
    --------
    >>> spot = LaserTRAM(name="GSD-1G_-_1")
    >>> spot.get_data(raw_data.loc["GSD-1G_-_1", :])
    """

    # TODO: get_data
    # - [x] add: check to make sure data are in right format else throw an error. do this by having a list of required columns and checking for them.

    # get data and set index to "SampleLabel" column
    self.data = df.reset_index()

    col_check, type_check = formatting.check_lt_input_format(self.data)
    if verbose:
        print(
            "checking LaserTRAM input data format for correct column headers and data types..."
        )

    if col_check is not None:
        raise ValueError(
            f"It looks like your input data are missing the following required columns: {col_check}. Please fix before continuing with processing."
        )
    if type_check is not None:
        raise TypeError(
            f"It looks like your input data have incorrect types in the following column indices: {type_check}. Please fix before continuing with processing."
        )

    if verbose:
        print("check complete...input data format looks good!")

    self.data = self.data.set_index("SampleLabel")

    # convert time units from ms --> s if applicable
    if time_units == "ms":
        self.data["Time"] = self.data["Time"] / 1000
    elif time_units == "s":
        pass

    # just numpy matrix for data
    self.data_matrix = self.data.iloc[:, 1:].to_numpy()

    # list of analytes in experiment
    self.analytes = self.data.loc[:, "Time":].columns.tolist()[1:]

    # need to add check for if this exists otherwise there is no timestamp attribute
    self.timestamp = str(self.data.loc[:, "timestamp"].unique()[0])
    self.timestamp = pd.to_datetime(self.timestamp)

assign_int_std(int_std)

Assign the internal standard analyte for normalisation.

Parameters:

Name	Type	Description	Default
int_std	str	Column name for the internal standard analyte, e.g. "29Si".	required

Examples:

>>> spot.assign_int_std("29Si")

Source code in src/lasertram/tram/tram.py

def assign_int_std(self, int_std: str) -> None:
    """Assign the internal standard analyte for normalisation.

    Parameters
    ----------
    int_std : str
        Column name for the internal standard analyte, e.g.
        ``"29Si"``.

    Examples
    --------
    >>> spot.assign_int_std("29Si")
    """

    # set the internal standard analyte
    self.int_std = int_std

    # get the internal standard array index
    self.int_std_loc = np.where(np.array(self.analytes) == self.int_std)[0][0]

assign_intervals(bkgd, keep, omit=None)

Assign background, signal, and optional omission intervals.

Parameters:

Name	Type	Description	Default
bkgd	tuple of float	(start, stop) pair of values (in seconds) for the background signal region.	required
keep	tuple of float	(start, stop) pair of values (in seconds) for the ablation signal region used in concentration calculations.	required
omit	tuple of float or None	(start, stop) pair of values (in seconds) to be omitted from the keep interval, by default None.	None

Examples:

>>> spot.assign_intervals(bkgd=(5, 10), keep=(25, 40))

With a region to omit:

>>> spot.assign_intervals(bkgd=(5, 10), keep=(23, 40), omit=(30, 33))

Source code in src/lasertram/tram/tram.py

def assign_intervals(
    self,
    bkgd: tuple[float, float],
    keep: tuple[float, float],
    omit: tuple[float, float] | None = None,
) -> None:
    """Assign background, signal, and optional omission intervals.

    Parameters
    ----------
    bkgd : tuple of float
        ``(start, stop)`` pair of values (in seconds) for the
        background signal region.
    keep : tuple of float
        ``(start, stop)`` pair of values (in seconds) for the
        ablation signal region used in concentration calculations.
    omit : tuple of float or None, optional
        ``(start, stop)`` pair of values (in seconds) to be omitted
        from the ``keep`` interval, by default ``None``.

    Examples
    --------
    >>> spot.assign_intervals(bkgd=(5, 10), keep=(25, 40))

    With a region to omit:

    >>> spot.assign_intervals(bkgd=(5, 10), keep=(23, 40), omit=(30, 33))
    """

    # set background and interval times in s
    self.bkgd_start = bkgd[0]
    self.bkgd_stop = bkgd[1]
    self.int_start = keep[0]
    self.int_stop = keep[1]

    # equivalent background and interval times but as indices
    # in their respective arrays
    self.bkgd_start_idx = np.where(self.data["Time"] > self.bkgd_start)[0][0]
    self.bkgd_stop_idx = np.where(self.data["Time"] > self.bkgd_stop)[0][0]
    self.int_start_idx = np.where(self.data["Time"] > self.int_start)[0][0]
    self.int_stop_idx = np.where((self.data["Time"] > self.int_stop))[0][0]

    # boolean whether or not there is an omitted region
    self.omitted_region = False
    # if omission is true, set those start and stop times like above
    if omit:
        self.omit_start = omit[0]
        self.omit_stop = omit[1]
        self.omit_start_idx = (
            np.where(self.data["Time"] > self.omit_start)[0][0] - self.int_start_idx
        )
        self.omit_stop_idx = (
            np.where(self.data["Time"] > self.omit_stop)[0][0] - self.int_start_idx
        )

        self.omitted_region = True

get_bkgd_data()

Calculate median background values for all analytes.

Uses the intervals assigned in assign_intervals() to take the median value of all analytes within the background range. These are later subtracted from the ablation signal in subtract_bkgd().

Source code in src/lasertram/tram/tram.py

def get_bkgd_data(self) -> None:
    """Calculate median background values for all analytes.

    Uses the intervals assigned in ``assign_intervals()`` to take
    the median value of all analytes within the background range.
    These are later subtracted from the ablation signal in
    ``subtract_bkgd()``.
    """
    # median background data values
    self.bkgd_data_median = np.median(
        self.data_matrix[self.bkgd_start_idx : self.bkgd_stop_idx, 1:], axis=0
    )

get_detection_limits()

Calculate detection limits in counts per second.

Defined as the median background plus three standard deviations of the background signal for each analyte.

Source code in src/lasertram/tram/tram.py

def get_detection_limits(self) -> None:
    """Calculate detection limits in counts per second.

    Defined as the median background plus three standard deviations
    of the background signal for each analyte.
    """

    self.detection_limits = np.std(
        self.data_matrix[self.bkgd_start_idx : self.bkgd_stop_idx, 1:], axis=0
    ) * 3 + np.median(
        self.data_matrix[self.bkgd_start_idx : self.bkgd_stop_idx, 1:], axis=0
    )

subtract_bkgd()

Subtract the median background from the ablation signal.

Removes the median background values calculated in get_bkgd_data() from the signal in the keep interval established in assign_intervals().

Source code in src/lasertram/tram/tram.py

def subtract_bkgd(self) -> None:
    """Subtract the median background from the ablation signal.

    Removes the median background values calculated in
    ``get_bkgd_data()`` from the signal in the *keep* interval
    established in ``assign_intervals()``.
    """
    self.bkgd_subtract_data = (
        self.data_matrix[self.int_start_idx : self.int_stop_idx, 1:]
        - self.bkgd_data_median
    )

normalize_interval()

Normalise the ablation interval to the internal standard.

Divides all analytes in the keep portion of the signal by the internal standard analyte. Also calculates the median normalised value, its standard error of the mean, and relative standard error of the mean.

Source code in src/lasertram/tram/tram.py

def normalize_interval(self) -> None:
    """Normalise the ablation interval to the internal standard.

    Divides all analytes in the *keep* portion of the signal by the
    internal standard analyte. Also calculates the median normalised
    value, its standard error of the mean, and relative standard
    error of the mean.
    """

    # set the detection limit thresholds to be checked against
    # with the interval data. This basically takes the detection limits
    threshold = self.detection_limits - self.bkgd_data_median

    # if there's an omitted region, remove it from the data to be further processed
    # for the chosen interval
    if self.omitted_region is True:
        self.bkgd_subtract_normal_data = np.delete(
            self.bkgd_subtract_data,
            np.arange(self.omit_start_idx, self.omit_stop_idx),
            axis=0,
        ) / np.delete(
            self.bkgd_subtract_data[:, self.int_std_loc][:, None],
            np.arange(self.omit_start_idx, self.omit_stop_idx),
            axis=0,
        )

    else:
        self.bkgd_subtract_normal_data = (
            self.bkgd_subtract_data
            / self.bkgd_subtract_data[:, self.int_std_loc][:, None]
        )

    # get background corrected and normalized median values for an interval
    self.bkgd_subtract_med = np.median(self.bkgd_subtract_normal_data, axis=0)
    self.bkgd_subtract_med[
        np.median(self.bkgd_subtract_data, axis=0) <= threshold
    ] = -9999
    self.bkgd_subtract_med[np.median(self.bkgd_subtract_data, axis=0) == 0] = -9999

    # standard error of the mean for the interval region
    self.bkgd_subtract_std_err = self.bkgd_subtract_normal_data.std(
        axis=0
    ) / np.sqrt(abs(self.int_stop_idx - self.int_start_idx))

    self.bkgd_subtract_std_err_rel = 100 * (
        self.bkgd_subtract_std_err / self.bkgd_subtract_med
    )

make_output_report()

Create an output report summarising the spot processing.

Generates a single-row pandas.DataFrame stored in self.output_report with the following columns:

timestamp | Spot | despiked | omitted_region | bkgd_start | bkgd_stop | int_start | int_stop | norm | norm_cps | followed by normalised analyte values and their standard errors.

Examples:

>>> spot.make_output_report()
>>> spot.output_report.head()

Source code in src/lasertram/tram/tram.py

def make_output_report(self) -> None:
    """Create an output report summarising the spot processing.

    Generates a single-row ``pandas.DataFrame`` stored in
    ``self.output_report`` with the following columns:

    ``timestamp | Spot | despiked | omitted_region | bkgd_start |
    bkgd_stop | int_start | int_stop | norm | norm_cps |``
    followed by normalised analyte values and their standard errors.

    Examples
    --------
    >>> spot.make_output_report()
    >>> spot.output_report.head()
    """
    if self.despiked is True:
        despike_col = self.despiked_elements
    else:
        despike_col = "None"

    if self.omitted_region is True:
        omitted_col = (
            self.data["Time"].iloc[self.omit_start_idx + self.int_start_idx],
            self.data["Time"].iloc[self.omit_stop_idx + self.int_start_idx],
        )
    else:
        omitted_col = "None"

    spot_data = pd.DataFrame(
        [
            self.timestamp,
            self.name,
            despike_col,
            omitted_col,
            self.data["Time"].iloc[self.bkgd_start_idx],
            self.data["Time"].iloc[self.bkgd_stop_idx],
            self.data["Time"].iloc[self.int_start_idx],
            self.data["Time"].iloc[self.int_stop_idx],
            self.int_std,
            np.median(self.bkgd_subtract_data[:, self.int_std_loc]),
        ]
    ).T
    spot_data.columns = [
        "timestamp",
        "Spot",
        "despiked",
        "omitted_region",
        "bkgd_start",
        "bkgd_stop",
        "int_start",
        "int_stop",
        "norm",
        "norm_cps",
    ]
    spot_data["timestamp"] = pd.to_datetime(spot_data["timestamp"])
    spot_data = pd.concat(
        [
            spot_data,
            pd.DataFrame(
                self.bkgd_subtract_med[np.newaxis, :], columns=self.analytes
            ),
            pd.DataFrame(
                self.bkgd_subtract_std_err_rel[np.newaxis, :],
                columns=[f"{analyte}_se" for analyte in self.analytes],
            ),
        ],
        axis="columns",
    )

    for col in ["bkgd_start", "bkgd_stop", "int_start", "int_stop", "norm_cps"]:
        spot_data[col] = spot_data[col].astype(np.float64)

    self.output_report = spot_data

despike_data(analyte_list='all', std_devs=4, window=25)

Despike normalised data using a rolling z-score filter.

Applies a z-score filter to the internally normalised data to remove analytical spikes. Must be called after normalize_interval() and before make_output_report().

Parameters:

Name	Type	Description	Default
analyte_list	str or list of str	Analytes to despike. Accepts a single analyte (e.g., "29Si"), a list (e.g., ["7Li", "29Si"]), or "all" to despike every analyte, by default "all".	'all'
std_devs	int	Number of standard deviations from the rolling mean to be considered an outlier, by default 4.	4
window	int	Size of the rolling average window, by default 25.	25

Examples:

Despike all analytes with default settings:

>>> spot.despike_data()

Despike a single analyte with custom parameters:

>>> spot.despike_data(analyte_list="208Pb", std_devs=3, window=30)

Source code in src/lasertram/tram/tram.py

def despike_data(
    self,
    analyte_list: str | list[str] = "all",
    std_devs: int = 4,
    window: int = 25,
) -> None:
    """Despike normalised data using a rolling z-score filter.

    Applies a z-score filter to the internally normalised data to
    remove analytical spikes. Must be called **after**
    ``normalize_interval()`` and **before**
    ``make_output_report()``.

    Parameters
    ----------
    analyte_list : str or list of str, optional
        Analytes to despike. Accepts a single analyte
        (e.g., ``"29Si"``), a list (e.g., ``["7Li", "29Si"]``),
        or ``"all"`` to despike every analyte, by default ``"all"``.
    std_devs : int, optional
        Number of standard deviations from the rolling mean to be
        considered an outlier, by default 4.
    window : int, optional
        Size of the rolling average window, by default 25.

    Examples
    --------
    Despike all analytes with default settings:

    >>> spot.despike_data()

    Despike a single analyte with custom parameters:

    >>> spot.despike_data(analyte_list="208Pb", std_devs=3, window=30)
    """

    assert (
        self.bkgd_subtract_normal_data is not None
    ), "please normalize your data prior to despiking"

    self.despiked = True

    if analyte_list == "all":
        filter_list = self.analytes
    else:
        if isinstance(analyte_list, list):
            pass
        else:
            analyte_list = [analyte_list]

        filter_list = analyte_list

    self.despiked_elements = filter_list

    df = pd.DataFrame(self.bkgd_subtract_normal_data, columns=self.analytes)

    for analyte in filter_list:

        filtered = _z_filter(df[analyte], window=window, std_devs=std_devs)

        # replaces data with despiked data
        df[analyte] = filtered

    self.bkgd_subtract_normal_data = df.loc[:, self.analytes].values

    # now recalculate uncertainties after despiking
    # standard error of the mean for the interval region
    self.bkgd_subtract_std_err = self.bkgd_subtract_normal_data.std(
        axis=0
    ) / np.sqrt(abs(self.int_stop_idx - self.int_start_idx))

    self.bkgd_subtract_std_err_rel = 100 * (
        self.bkgd_subtract_std_err / self.bkgd_subtract_med
    )

LaserCalc

Concentration calculations from processed LA-ICP-MS data — calibration standard management, drift correction, and element quantification using the methodology of Longerich et al. (1996).

lasertram.calc.LaserCalc

Calculate concentrations from LA-ICP-MS normalised ratios.

Implements the methodology of Longerich et al. (1996) and Kent & Ungerer (2006) for converting internally normalised ratios produced by LaserTRAM into absolute concentrations (ppm). The basic workflow is:

1. Upload SRM compositions via get_SRM_comps().
2. Upload LaserTRAM output via get_data().
3. Set the calibration standard via set_calibration_standard().
4. Check for drift via drift_check().
5. Set internal standard concentrations via set_int_std_concentrations().
6. Calculate concentrations via calculate_concentrations().

Parameters:

Name	Type	Description	Default
name	str	Name for the experiment / processing run.	required

References

.. [1] Longerich, H. P., Jackson, S. E., & Günther, D. (1996). Inter-laboratory note. Laser ablation inductively coupled plasma mass spectrometric transient signal data acquisition and analyte concentration calculation. J. Anal. At. Spectrom., 11(9), 899–904. .. [2] Kent, A. J., & Ungerer, C. A. (2006). Analysis of light lithophile elements (Li, Be, B) by laser ablation ICP-MS: comparison between magnetic sector and quadrupole ICP-MS. Am. Mineral., 91(8–9), 1401–1411.

Examples:

>>> from lasertram import LaserCalc, helpers
>>> concentrations = LaserCalc(name="tutorial")
>>> concentrations.get_SRM_comps(srm_data)
>>> concentrations.get_data(processed_df)
>>> concentrations.set_calibration_standard("GSD-1G")
>>> concentrations.drift_check()
>>> concentrations.get_calibration_std_ratios()
>>> concentrations.set_int_std_concentrations(spots, values, uncertainties)
>>> concentrations.calculate_concentrations()

Source code in src/lasertram/calc/calc.py

class LaserCalc:
    """Calculate concentrations from LA-ICP-MS normalised ratios.

    Implements the methodology of Longerich et al. (1996) and
    Kent & Ungerer (2006) for converting internally normalised ratios
    produced by ``LaserTRAM`` into absolute
    concentrations (ppm). The basic workflow is:

    1. Upload SRM compositions via ``get_SRM_comps()``.
    2. Upload LaserTRAM output via ``get_data()``.
    3. Set the calibration standard via ``set_calibration_standard()``.
    4. Check for drift via ``drift_check()``.
    5. Set internal standard concentrations via
       ``set_int_std_concentrations()``.
    6. Calculate concentrations via ``calculate_concentrations()``.

    Parameters
    ----------
    name : str
        Name for the experiment / processing run.

    References
    ----------
    .. [1] Longerich, H. P., Jackson, S. E., & Günther, D. (1996).
       Inter-laboratory note. Laser ablation inductively coupled plasma
       mass spectrometric transient signal data acquisition and analyte
       concentration calculation. *J. Anal. At. Spectrom.*, 11(9), 899–904.
    .. [2] Kent, A. J., & Ungerer, C. A. (2006). Analysis of light
       lithophile elements (Li, Be, B) by laser ablation ICP-MS:
       comparison between magnetic sector and quadrupole ICP-MS.
       *Am. Mineral.*, 91(8–9), 1401–1411.

    Examples
    --------
    >>> from lasertram import LaserCalc, helpers
    >>> concentrations = LaserCalc(name="tutorial")
    >>> concentrations.get_SRM_comps(srm_data)
    >>> concentrations.get_data(processed_df)
    >>> concentrations.set_calibration_standard("GSD-1G")
    >>> concentrations.drift_check()
    >>> concentrations.get_calibration_std_ratios()
    >>> concentrations.set_int_std_concentrations(spots, values, uncertainties)
    >>> concentrations.calculate_concentrations()
    """

    def __init__(self, name: str, auto_load_srm: bool = True) -> None:
        """Initialise a LaserCalc object.

        Parameters
        ----------
        name : str
            Name for the experiment / processing run.
        auto_load_srm : bool, optional
            If True (the default), automatically load the bundled GeoReM
            SRM database during construction. Set to False to retain the
            legacy behaviour where get_SRM_comps() must be called explicitly.
        """
        # all attributes in relative chronological order that they are created in
        # if everything is done correctly. These all will get rewritten throughout the
        # data processing pipeline but this allows us to see what all the potential attributes
        # are going to be from the beginning (PEP convention)

        # for the math involved please see:

        # name for the lasercalc object
        # for notekeeping
        self.name = name

        # 2D pandas dataframe of standards reference material preferred compositions
        # from georem
        self.standards_data = None

        # List of standard reference materials in self.standards_data
        self.database_standards = None

        # list of standard reference material elements/oxides in self.standards_data
        self.standard_elements = None

        # list of standard reference material element/oxide 1 sigma uncertainties in self.standards_data
        self.standard_element_uncertainties = None

        # list of spot analyses for which concentrations are being calculated
        # this is the equivalent of self.data['Spot']
        self.spots = None

        # list of analytes for which concentrations are being calculated
        # these are column headers in self.data
        self.analytes = None

        # dict mapping each analyte to its corresponding _se column name
        self.analyte_se_map = None

        # ordered list of SE column names matching analyte order
        self.analyte_se_columns = None

        # 1 sigma standard deviation of the calibration standard values
        # in self.data. Is len(analytes) in shape
        self.calibration_std_stdevs = None

        # 2D pandas dataframe that represents the metadata and data for numerous
        # spot analyses. Each row is the equivalent of a LaserTRAM.output_report
        # and has the following columns:
        # |timestamp|Spot|despiked|omitted_region|bkgd_start|bkgd_stop|int_start|int_stop|norm|norm_cps|analyte vals and uncertainties -->|
        # |---------|----|--------|--------------|----------|---------|---------|--------|----|--------|----------------------------------|
        self.data = None

        # element used as internal standard. NOT to be confused with analyte
        # e.g. self.int_std_element == 'Si' NOT '29Si'
        self.int_std_element = None
        self.int_std_units = None

        # list of standard reference materials in found in self.data that are
        # also found in self.database_standards. This lets you know which standard reference
        # materials you can use as potential calibration standards
        self.potential_calibration_standards = None

        # list of samples in self.data with the self.potential_calibration_standards
        # removed
        self.samples_nostandards = None

        # list of elements for which concentrations are being calculated
        # this is the equivalent to self.analytes with the atomic masses
        # removed
        self.elements = None

        # string representing the standard reference material used
        # as the calibration standard for calculating concentrations
        self.calibration_std = None

        # 2D pandas dataframe which is a subset of self.data for only the
        # calibration standard data. This is essentially self.data.loc[self.calibration_std,:]
        self.calibration_std_data = None

        # mean calibration standard values for all analytes
        # equivalent of self.calibration_std_data.mean(axis = 0)
        self.calibration_std_means = None

        # calibration standard standard error of the mean for all analytes
        self.calibration_std_ses = None

        # 2D dataframe that is contains statistics for each analyte in self.calibration_std_data
        # columns are:
        # drift_correct | f_pval | f_value | f_crit_value | rmse | slope | intercept | mean | std_dev | percent_std_err
        # These stats are based on the following regression:
        # for each analyte
        # x = self.calibration_std_data.loc[:,'timestamp']
        # y = self.calibration_std_data.loc[:, analyte]

        # X = sm.add_constant(x)
        # Note the difference in argument order
        # model = sm.OLS(y, X).fit()
        # now generate predictions
        # ypred = model.predict(X)

        # calc rmse
        # RMSE = rmse(y, ypred)

        self.calibration_std_stats = None

        # the ratio of concentrations between an analyte and the internal standard
        # in the georem calibration standard values
        self.calibration_std_conc_ratios = None

        # list of standard reference materials that are not used as calibration standard
        # this is effectively self.potential_calibration_standards with self.calibration_std
        # removed
        self.secondary_standards = None

        # 2D pandas dataframe of calculated concentrations for all spots in self.secondary_standards and all
        # analytes in self.analytes. This is self.data.loc[self.secondary_standards,self.analytes].shape in shape
        self.SRM_concentrations = None

        # 2D pandas dataframe of calculated concentrations for all spots in self.spots and all
        # analytes in self.analytes. This is self.data.loc[self.spots,self.analytes].shape in shape
        self.unknown_concentrations = None

        # 2D pandas dataframe of calculated accuracies for all spots in self.secondary_standards and all
        # analytes in self.analytes. This is self.data.loc[self.secondary_standards,self.analytes].shape in shape
        # here accuracy is just 100*measured_concentration / georem_concentration
        self.SRM_accuracies = None

        if auto_load_srm:
            try:
                from ..helpers.preprocessing import load_srm_database

                self.get_SRM_comps(load_srm_database())
            except Exception as exc:
                warnings.warn(
                    f"Auto-loading the bundled SRM database failed ({exc}). "
                    "Call get_SRM_comps() manually to supply a standards database "
                    "before running the pipeline.",
                    UserWarning,
                    stacklevel=2,
                )

    def get_SRM_comps(self, df: pd.DataFrame) -> None:
        """Load a database of standard reference material compositions.

        This call is optional when ``auto_load_srm=True``. Calling this
        method after auto-load will replace the bundled database and emit
        a UserWarning.

        Must be called **before** ``get_data()`` so that potential
        calibration standards can be identified.

        Parameters
        ----------
        df : pandas.DataFrame
            DataFrame of standard reference materials where each row
            represents data for one SRM. The first column must be named
            ``"Standard"``. All other columns are elemental
            concentrations. Standard names must match those found in
            [GEOREM](http://georem.mpch-mainz.gwdg.de/).

        Examples
        --------
        >>> import pandas as pd
        >>> srm_data = pd.read_excel("laicpms_stds_tidy.xlsx")
        >>> concentrations.get_SRM_comps(srm_data)
        """
        # TODO: get_SRM_comps
        # - [x] add: check to make sure data are in right format else throw an error. do this by having a list of required columns and checking for them.

        if self.standards_data is not None:
            warnings.warn(
                "A standards database is already loaded (bundled database). "
                "Calling get_SRM_comps() is overriding it with the supplied DataFrame. "
                "If this is intentional, no action is needed.",
                UserWarning,
                stacklevel=2,
            )

        print(
            "checking uploaded standards database format for correct column headers and data types..."
        )
        col_check, type_check = formatting.check_srm_format(df)

        if col_check is not None:
            raise ValueError(
                f"It looks like your input data are missing the following required columns: {col_check}. Please fix before continuing with processing."
            )
        if type_check is not None:
            raise TypeError(
                f"It looks like your input data have incorrect types in the following columns: {type_check}. Please fix before continuing with processing."
            )
        self.standards_data = df.set_index("Standard")

        print("check complete...standards database format looks good!")

        self.database_standards = self.standards_data.index.unique().to_list()
        # Get a list of all of the elements supported in the published standard datasheet
        # Get a second list for the same elements but their corresponding uncertainty columns
        self.standard_elements = [
            analyte
            for analyte in self.standards_data.columns.tolist()
            if "_std" not in analyte
        ]
        self.standard_element_uncertainties = [
            analyte + "_std" for analyte in self.standard_elements
        ]

    def get_data(self, df: pd.DataFrame, verbose: bool = True) -> None:
        """Load LaserTRAM output for concentration calculations.

        Automatically identifies potential calibration standards by
        comparing spot names to the SRM database loaded via
        ``get_SRM_comps()``.

        Parameters
        ----------
        df : pandas.DataFrame
            Concatenated ``LaserTRAM.output_report`` rows for
            many spots.
        verbose : bool, optional
            Whether to print status messages, by default ``True``.

        Examples
        --------
        >>> concentrations.get_data(processed_df)
        """
        # check if first row is nan (output from GUI does this).
        # If so, remove it
        df = df[df.iloc[:, 0].isna() == False]

        # TODO: get_data
        # - [x] add: check to make sure data are in right format else throw an error. do this by having a list of required columns and checking for them. this can't include analytes though - just the metadata.
        # - [x] add: in some checking for analytes to make sure that the measured analytes exist within the standards database. compare self.elements to self.standard_elements by going through each standard and checking for nan for that element
        # - [ ] add: duplicate name check.
        if verbose:
            print(
                "checking LaserCalc input data format for correct column headers and data types..."
            )

        col_check, type_check = formatting.check_lt_complete_format(df)
        if col_check is not None:
            raise ValueError(
                f"It looks like your input data are missing the following required columns: {col_check}. Please fix before continuing with processing."
            )
        if type_check is not None:
            raise TypeError(
                f"It looks like your input data have incorrect types in the following columns: {type_check}. Please fix before continuing with processing."
            )

        if verbose:
            print("check complete...input data format looks good!")

        data = df.set_index("Spot")
        data.insert(loc=1, column="index", value=np.arange(1, len(data) + 1))

        self.spots = data.index.unique().dropna().tolist()

        # Check for potential calibration standards. This will let us know what our options
        # are for choosing calibration standards by looking for spots that have the same string
        # as the standard spreadsheet

        stds_column = [
            [std for std in self.database_standards if std in spot]
            for spot in self.spots
        ]

        stds_column = [["unknown"] if not l else l for l in stds_column]

        stds_column = [std for sublist in stds_column for std in sublist]

        # standards that can be used as calibrations standards (must have more than 1 analysis)
        # potential_standards = list(np.unique(stds_column))
        potential_standards = [
            std for std in np.unique(stds_column) if stds_column.count(std) > 1
        ]
        potential_standards.remove("unknown")

        # all of the samples in your input sheet that are NOT potential standards
        all_standards = list(np.unique(stds_column))
        all_standards.remove("unknown")

        data["sample"] = stds_column

        data.reset_index(inplace=True)
        data.set_index("sample", inplace=True)

        self.data = data
        self.potential_calibration_standards = potential_standards
        self.samples_nostandards = list(np.setdiff1d(stds_column, all_standards))

        self.analytes = [
            col for col in data.columns.tolist()
            if ANALYTE_PATTERN.match(col)
        ]
        # elements without isotopes in the front
        self.elements = [re.split(r"(\d+)", analyte)[2] for analyte in self.analytes]

        # SE column pairing: map each analyte to its corresponding _se column
        analyte_set = set(self.analytes)
        self.analyte_se_map = {}

        for col in data.columns:
            if col.endswith('_se'):
                base = col[:-3]  # Remove '_se' suffix
                if base in analyte_set:
                    self.analyte_se_map[base] = col

        # Ordered list of SE columns matching analyte order
        self.analyte_se_columns = [
            self.analyte_se_map[a] for a in self.analytes
            if a in self.analyte_se_map
        ]

        # first check to make sure that the element exists within the standards database
        for el in self.elements:
            if el not in self.standard_elements:
                raise ValueError(
                    f"{el} is not in the standards database. Please remove it from your data before proceeding with processing."
                )

        if len(self.potential_calibration_standards) == 0:
            raise ValueError(
                f"No potential calibration standards found. The following spot names "
                f"were evaluated against the standards database: {list(np.unique(stds_column))}. "
                f"Ensure your data contains at least one SRM spot name that matches a "
                f"standard in the GeoReM database, with at least two analyses per standard."
            )

        print(
            f"Your potential calibration standards are: {[str(s) for s in self.potential_calibration_standards]}"
        )

        # internal standard analyte from lasertram
        self.int_std_element = re.split(r"(\d+)", self.data["norm"].unique()[0])[2]

    def set_calibration_standard(self, std: str) -> None:
        """Designate the calibration standard for concentration calculations.

        Calculates mean values, standard deviations, and relative
        standard errors for the chosen SRM across all analytes.

        Parameters
        ----------
        std : str
            Name of the standard reference material, e.g.
            ``"GSD-1G"``, ``"NIST-612"``, ``"BCR-2G"``.

        Examples
        --------
        >>> concentrations.set_calibration_standard("GSD-1G")
        """
        self.calibration_std = std

        self.calibration_std_data = self.data.loc[std, :]
        # Calibration standard information
        # mean
        self.calibration_std_means = self.calibration_std_data.loc[
            :, self.analytes
        ].mean()
        # std deviation
        self.calibration_std_stdevs = self.calibration_std_data.loc[
            :, self.analytes
        ].std()
        # relative standard error
        self.calibration_std_ses = 100 * (
            (self.calibration_std_stdevs / self.calibration_std_means)
            / np.sqrt(self.calibration_std_data.shape[0])
        )

        # Warn about fringe analytes with no published SRM value
        for analyte, element in zip(self.analytes, self.elements):
            if pd.isna(self.standards_data.loc[self.calibration_std, element]):
                warnings.warn(
                    f"Analyte '{analyte}' (element '{element}') has no published "
                    f"concentration in '{self.calibration_std}'. Concentrations for "
                    f"this analyte will be reported as 'n.c.' (not calibrated). "
                    f"Verify your SRM choice or switch to an SRM with a published value.",
                    UserWarning,
                    stacklevel=2,
                )

    def drift_check(self, pval: float = 0.01) -> None:
        """Assess instrument drift for each analyte.

        Performs a linear regression of each analyte's normalised ratio
        through time for the calibration standard. If the regression is
        statistically significant (F-test *p*-value < ``pval`` **and**
        F > F_crit), the analyte is flagged for drift correction
        in ``calculate_concentrations()``.

        Results are stored in ``self.calibration_std_stats``.

        Parameters
        ----------
        pval : float, optional
            Significance threshold to reject the null hypothesis of no
            drift, by default 0.01.

        Examples
        --------
        >>> concentrations.drift_check()
        >>> concentrations.calibration_std_stats.head()
        """
        calib_std_rmses = []
        calib_std_slopes = []
        calib_std_intercepts = []
        drift_check = []

        f_pvals = []
        f_vals = []
        f_crits = []
        for analyte in self.analytes:
            # Getting regression statistics on analyte normalized ratios through time
            # for the calibration standard. This is what we use to check to see if it needs
            # to be drift corrected
            if "timestamp" in self.calibration_std_data.columns.tolist():
                # get an array in time units based on timestamp column. This is
                # is in seconds
                x = np.array(
                    [
                        np.datetime64(d, "m")
                        for d in self.calibration_std_data["timestamp"]
                    ]
                ).astype(np.float64)
                # x = np.cumsum(np.diff(x))
                # x = np.insert(x, 0, 0).astype(np.float64)

            else:
                x = self.calibration_std_data["index"].to_numpy()

            y = self.calibration_std_data.loc[:, analyte].astype("float64")

            X = sm.add_constant(x)
            # Note the difference in argument order
            model = sm.OLS(y, X).fit()
            # now generate predictions
            ypred = model.predict(X)

            # calc rmse
            RMSE = rmse(y, ypred)

            calib_std_rmses.append(RMSE)

            if model.params.shape[0] < 2:
                calib_std_slopes.append(model.params.loc["x1"])
                calib_std_intercepts.append(0)

            else:
                calib_std_slopes.append(model.params.loc["x1"])
                calib_std_intercepts.append(model.params.loc["const"])

            # new stuff
            # confidence limit 99%

            # f value stuff

            fvalue = model.fvalue
            f_vals.append(fvalue)
            f_pvalue = model.f_pvalue
            f_pvals.append(f_pvalue)
            fcrit = stats.f.ppf(q=1 - pval, dfn=len(x) - 1, dfd=len(y) - 1)
            f_crits.append(fcrit)
            if (f_pvalue < pval) and (fvalue > fcrit):
                drift = "True"
                drift_check.append(drift)
            else:
                drift = "False"
                drift_check.append(drift)

        self.calibration_std_stats = pd.DataFrame(
            {
                "drift_correct": drift_check,
                "f_pval": f_pvals,
                "f_value": f_vals,
                "f_crit_value": f_crits,
                "rmse": calib_std_rmses,
                "slope": calib_std_slopes,
                "intercept": calib_std_intercepts,
                "mean": self.calibration_std_means[self.analytes].to_numpy(),
                "std_dev": self.calibration_std_stdevs[self.analytes].to_numpy(),
                "percent_std_err": self.calibration_std_ses[self.analytes].to_numpy(),
            },
            index=self.analytes,
        )

    def get_calibration_std_ratios(self) -> None:
        """Calculate concentration ratios for the calibration standard.

        For the chosen calibration standard, compute the ratio of each
        analyte's concentration to the internal standard element's
        concentration. Results are stored in
        ``self.calibration_std_conc_ratios``.

        Examples
        --------
        >>> concentrations.get_calibration_std_ratios()
        """

        # For our calibration standard, calculate the concentration ratio
        # of each analyte to the element used as the internal standard
        std_conc_ratios = []

        for element in self.elements:
            if element in self.standard_elements:
                std_conc_ratios.append(
                    self.standards_data.loc[self.calibration_std, element]
                    / self.standards_data.loc[
                        self.calibration_std, self.int_std_element
                    ]
                )

        # make our list an array for easier math going forward
        self.calibration_std_conc_ratios = np.array(std_conc_ratios)

    def set_int_std_concentrations(
        self,
        spots: pd.Series | None = None,
        concentrations: npt.ArrayLike | None = None,
        uncertainties: npt.ArrayLike | None = None,
        units: str = "wt_per_ox",
    ) -> None:
        """Assign internal standard concentrations to spots.

        A linear change in the concentration value produces a
        proportional linear change in the calculated concentration.

        Parameters
        ----------
        spots : pandas.Series or None, optional
            Spot names to assign. Corresponds to the ``Spot`` column
            from the LaserTRAM output.
        concentrations : array-like or None, optional
            Internal standard concentration values. Must be the
            same length as ``spots``.
        uncertainties : array-like or None, optional
            Relative uncertainty in percent for the internal standard.
            Must be the same length as ``spots``.
        units : str, optional
            Units for the concentration values. One of ``"wt_per_ox"``
            (weight percent oxide, default), ``"wt_per_el"`` (weight
            percent element), or ``"ppm_el"`` (ppm element).

        Examples
        --------
        >>> concentrations.set_int_std_concentrations(
        ...     internal_std_comps["Spot"],
        ...     internal_std_comps["SiO2"],
        ...     internal_std_comps["SiO2_std%"],
        ... )
        """

        assert units in [
            "wt_per_ox",
            "wt_per_el",
            "ppm_el",
        ], f"{units} is not a supported unit for concentrations. accepted units are: ['wt_per_ox','wt_per_el', 'ppm_el']"

        if spots is None:
            spots = (self.data["Spot"],)
            concentrations = (np.full(self.data["Spot"].shape[0], 10),)
            uncertainties = (np.full(self.data["Spot"].shape[0], 1),)

        self.data["int_std_comp"] = 10.0
        self.data["int_std_rel_unc"] = 1.0
        df = self.data.reset_index().set_index("Spot")

        for spot, concentration, uncertainty in zip(
            spots, concentrations, uncertainties
        ):
            df.loc[spot, "int_std_comp"] = concentration
            df.loc[spot, "int_std_rel_unc"] = uncertainty

        self.data["int_std_comp"] = df["int_std_comp"].to_numpy()
        self.data["int_std_rel_unc"] = df["int_std_rel_unc"].to_numpy()

        self.int_std_units = units

    def calculate_concentrations(self) -> None:
        """Calculate concentrations and uncertainties for all spots.

        Uses the calibration standard, internal standard concentrations,
        and drift correction information to compute absolute
        concentrations. Stores results in
        ``self.unknown_concentrations`` and ``self.SRM_concentrations``.

        Values below the detection limit are replaced with ``"b.d.l."``.

        Examples
        --------
        >>> concentrations.calculate_concentrations()
        >>> concentrations.unknown_concentrations.head()
        """

        secondary_standards = self.potential_calibration_standards.copy()
        secondary_standards.remove(self.calibration_std)
        self.secondary_standards = secondary_standards
        secondary_standards_concentrations_list = []
        unknown_concentrations_list = []

        for sample in secondary_standards:
            Cn_u = self.standards_data.loc[
                sample,
                re.split(
                    r"(\d+)",
                    self.calibration_std_data["norm"].unique()[0],
                )[2],
            ]
            Cin_std = self.calibration_std_conc_ratios
            Ni_std = self.calibration_std_stats["mean"][self.analytes]
            Ni_u = self.data.loc[sample, self.analytes]

            concentrations = Cn_u * (Cin_std / Ni_std) * Ni_u

            drift_concentrations_list = []

            for j, analyte, slope, intercept, drift in zip(
                range(len(self.analytes)),
                self.analytes,
                self.calibration_std_stats["slope"],
                self.calibration_std_stats["intercept"],
                self.calibration_std_stats["drift_correct"],
            ):
                if "True" in drift:
                    if "timestamp" in self.data.columns.tolist():
                        frac = (
                            slope
                            * np.array(
                                [
                                    np.datetime64(d, "m")
                                    for d in self.data.loc[sample, "timestamp"]
                                ]
                            ).astype(np.float64)
                            + intercept
                        )
                    else:
                        frac = slope * self.data.loc[sample, "index"] + intercept

                    Ni_std = frac

                    drift_concentrations = Cn_u * (Cin_std[j] / Ni_std) * Ni_u[analyte]

                    if isinstance(drift_concentrations, np.float64):
                        df = pd.DataFrame(
                            np.array([drift_concentrations]), columns=[analyte]
                        )

                    else:
                        df = pd.DataFrame(drift_concentrations, columns=[analyte])

                    drift_concentrations_list.append(df)

            if len(drift_concentrations_list) > 0:
                drift_df = pd.concat(drift_concentrations_list, axis="columns")

                if drift_df.shape[0] == 1:
                    drift_df["sample"] = sample
                    drift_df.set_index("sample", inplace=True)
            else:
                drift_df = pd.DataFrame()

            for column in drift_df.columns.tolist():
                if isinstance(concentrations, pd.Series):
                    concentrations.loc[column] = drift_df[column].to_numpy()[0]

                else:
                    concentrations[column] = drift_df[column].to_numpy()

            if isinstance(concentrations, pd.Series):
                concentrations = pd.DataFrame(concentrations).T
                concentrations["sample"] = sample
                concentrations.set_index("sample", inplace=True)

            secondary_standards_concentrations_list.append(concentrations)

        ###############################

        for sample in self.samples_nostandards:
            # Cn_u = conversions.oxide_to_ppm(
            #     self.data.loc[sample, "int_std_comp"],
            #     self.data.loc[sample, "norm"].unique()[0],
            # ).to_numpy()
            int_std_element = "".join(
                [
                    i
                    for i in self.data.loc[sample, "norm"].unique()[0]
                    if not i.isdigit()
                ]
            )

            # handle conversions from various units to all end up at ppm element
            if self.int_std_units == "wt_per_ox":
                Cn_u = conversions.oxide_to_ppm(
                    self.data.loc[sample, "int_std_comp"], int_std_element
                )
            elif self.int_std_units == "wt_per_el":
                Cn_u = self.data.loc[sample, "int_std_comp"].values * 1e4
            elif self.int_std_units == "ppm_el":
                Cn_u = self.data.loc[sample, "int_std_comp"].values

            Cin_std = self.calibration_std_conc_ratios
            Ni_std = self.calibration_std_stats["mean"][self.analytes].to_numpy()
            Ni_u = self.data.loc[sample, self.analytes].to_numpy()

            concentrations = pd.DataFrame(
                Cn_u[:, np.newaxis] * (Cin_std / Ni_std) * Ni_u, columns=self.analytes
            )

            drift_concentrations_list = []

            for j, analyte, slope, intercept, drift in zip(
                range(len(self.analytes)),
                self.analytes,
                self.calibration_std_stats["slope"],
                self.calibration_std_stats["intercept"],
                self.calibration_std_stats["drift_correct"],
            ):
                if "True" in drift:
                    if "timestamp" in self.data.columns.tolist():
                        frac = (
                            slope
                            * np.array(
                                [
                                    np.datetime64(d, "m")
                                    for d in self.data.loc[sample, "timestamp"]
                                ]
                            ).astype(np.float64)
                            + intercept
                        )
                    else:
                        frac = slope * self.data.loc[sample, "index"] + intercept
                    frac = np.array(frac)
                    drift_concentrations = (
                        Cn_u[:, np.newaxis]
                        * (Cin_std[j] / frac)[:, np.newaxis]
                        * Ni_u[:, j][:, np.newaxis]
                    )

                    if isinstance(drift_concentrations, np.float64):
                        df = pd.DataFrame(
                            np.array([drift_concentrations]), columns=[analyte]
                        )

                    else:
                        df = pd.DataFrame(drift_concentrations, columns=[analyte])

                    drift_concentrations_list.append(df)

            if len(drift_concentrations_list) > 0:
                drift_df = pd.concat(drift_concentrations_list, axis="columns")

                if drift_df.shape[0] == 1:
                    drift_df["sample"] = sample
                    drift_df.set_index("sample", inplace=True)
            else:
                drift_df = pd.DataFrame()

            for column in drift_df.columns.tolist():
                if isinstance(concentrations, pd.Series):
                    concentrations.loc[column] = drift_df[column].to_numpy()[0]

                else:
                    concentrations[column] = drift_df[column].to_numpy()

            if isinstance(concentrations, pd.Series):
                concentrations = pd.DataFrame(concentrations).T
                concentrations["sample"] = sample
                concentrations.set_index("sample", inplace=True)

            unknown_concentrations_list.append(concentrations)

        self.SRM_concentrations = pd.concat(secondary_standards_concentrations_list)
        self.unknown_concentrations = pd.concat(unknown_concentrations_list)

        self.calculate_uncertainties()

        # Fill NaN concentrations with SENTINEL_VALUE_NC for fringe analytes
        for analyte in self.analytes:
            if self.unknown_concentrations[analyte].isna().any():
                self.unknown_concentrations[analyte] = self.unknown_concentrations[analyte].fillna(SENTINEL_VALUE_NC)
            if self.SRM_concentrations[analyte].isna().any():
                self.SRM_concentrations[analyte] = self.SRM_concentrations[analyte].fillna(SENTINEL_VALUE_NC)

        # INSERT IN SPOT METADATA NOW
        # OLD WAY OF REPLACING NEGATIVES. WILL THROW ERROR IN FUTURE FOR MIXING
        # STRINGS WITH FLOATS
        # self.unknown_concentrations[self.unknown_concentrations < 0] = "b.d.l."
        # self.SRM_concentrations[self.SRM_concentrations < 0] = "b.d.l."

        # THE NEW WAY OF DOING IT IS TO GO THROUGH COLUMN BY COLUMN AND CHECK FOR BELOW
        # 0 VALUES, CHANGE THE DTYPE TO OBJECT, AND THEN REPLACE THE NEGATIVE VALUES WITH BDL STRING
        # THEN CHANGE THE UNCERTAINTY VALUES TO BDL STRINGS BASED ON THE ROW WE DID FOR THE ACTUAL CONCENTRATION VALUE
        for analyte in self.analytes:
            # n.c. branch — MUST come before b.d.l. because SENTINEL_VALUE_NC < 0
            if any(self.unknown_concentrations[analyte] == SENTINEL_VALUE_NC):
                self.unknown_concentrations[analyte] = self.unknown_concentrations[analyte].astype("object")
                self.unknown_concentrations[f"{analyte}_interr"] = self.unknown_concentrations[f"{analyte}_interr"].astype("object")
                self.unknown_concentrations[f"{analyte}_exterr"] = self.unknown_concentrations[f"{analyte}_exterr"].astype("object")
                self.unknown_concentrations.loc[self.unknown_concentrations[analyte] == SENTINEL_VALUE_NC, analyte] = "n.c."
                self.unknown_concentrations.loc[self.unknown_concentrations[analyte] == "n.c.", f"{analyte}_interr"] = "n.c."
                self.unknown_concentrations.loc[self.unknown_concentrations[analyte] == "n.c.", f"{analyte}_exterr"] = "n.c."

            if any(self.SRM_concentrations[analyte] == SENTINEL_VALUE_NC):
                self.SRM_concentrations[analyte] = self.SRM_concentrations[analyte].astype("object")
                self.SRM_concentrations[f"{analyte}_interr"] = self.SRM_concentrations[f"{analyte}_interr"].astype("object")
                self.SRM_concentrations[f"{analyte}_exterr"] = self.SRM_concentrations[f"{analyte}_exterr"].astype("object")
                self.SRM_concentrations.loc[self.SRM_concentrations[analyte] == SENTINEL_VALUE_NC, analyte] = "n.c."
                self.SRM_concentrations.loc[self.SRM_concentrations[analyte] == "n.c.", f"{analyte}_interr"] = "n.c."
                self.SRM_concentrations.loc[self.SRM_concentrations[analyte] == "n.c.", f"{analyte}_exterr"] = "n.c."
            if self.unknown_concentrations[analyte].dtype != "object" and any(self.unknown_concentrations[analyte] < 0):
                self.unknown_concentrations[analyte] = self.unknown_concentrations[
                    analyte
                ].astype("object")
                self.unknown_concentrations[f"{analyte}_interr"] = (
                    self.unknown_concentrations[f"{analyte}_interr"].astype("object")
                )
                self.unknown_concentrations[f"{analyte}_exterr"] = (
                    self.unknown_concentrations[f"{analyte}_exterr"].astype("object")
                )

                self.unknown_concentrations.loc[
                    self.unknown_concentrations[analyte] < 0, analyte
                ] = "b.d.l."
                self.unknown_concentrations.loc[
                    self.unknown_concentrations[analyte] == "b.d.l.",
                    f"{analyte}_interr",
                ] = "b.d.l."
                self.unknown_concentrations.loc[
                    self.unknown_concentrations[analyte] == "b.d.l", f"{analyte}_interr"
                ] = "b.d.l."

            if self.SRM_concentrations[analyte].dtype != "object" and any(self.SRM_concentrations[analyte] < 0):
                self.SRM_concentrations[analyte] = self.SRM_concentrations[
                    analyte
                ].astype("object")
                self.SRM_concentrations[f"{analyte}_interr"] = self.SRM_concentrations[
                    f"{analyte}_interr"
                ].astype("object")
                self.SRM_concentrations[f"{analyte}_exterr"] = self.SRM_concentrations[
                    f"{analyte}_exterr"
                ].astype("object")

                self.SRM_concentrations.loc[
                    self.SRM_concentrations[analyte] < 0, analyte
                ] = "b.d.l."
                self.SRM_concentrations.loc[
                    self.SRM_concentrations[analyte] == "b.d.l.", f"{analyte}_interr"
                ] = "b.d.l."
                self.SRM_concentrations.loc[
                    self.SRM_concentrations[analyte] == "b.d.l", f"{analyte}_interr"
                ] = "b.d.l."

        self.SRM_concentrations.insert(
            0, "Spot", list(self.data.loc[self.secondary_standards, "Spot"])
        )

        if "timestamp" in self.data.columns.tolist():
            self.SRM_concentrations.insert(
                0,
                "timestamp",
                list(self.data.loc[self.secondary_standards, "timestamp"]),
            )
        else:
            self.SRM_concentrations.insert(
                0, "index", list(self.data.loc[self.secondary_standards, "index"])
            )
        self.unknown_concentrations.insert(
            0, "Spot", list(self.data.loc[self.samples_nostandards, "Spot"])
        )
        if "timestamp" in self.data.columns.tolist():
            self.unknown_concentrations.insert(
                0,
                "timestamp",
                list(self.data.loc[self.samples_nostandards, "timestamp"]),
            )
        else:
            self.unknown_concentrations.insert(
                0, "index", list(self.data.loc[self.samples_nostandards, "index"])
            )

        self.unknown_concentrations.index = [
            "unknown"
        ] * self.unknown_concentrations.shape[0]
        self.unknown_concentrations.index.name = "sample"

    def calculate_uncertainties(self) -> None:
        """Calculate internal and external uncertainties for each analysis.

        Called automatically by ``calculate_concentrations()``. The
        results are appended as ``<analyte>_interr`` and
        ``<analyte>_exterr`` columns to ``self.unknown_concentrations``
        and ``self.SRM_concentrations``.
        """

        myuncertainties = [analyte + "_se" for analyte in self.analytes]
        srm_rel_ext_uncertainties_list = []
        unk_rel_ext_uncertainties_list = []
        srm_rel_int_uncertainties_list = []
        unk_rel_int_uncertainties_list = []
        # use RMSE of regression for elements where drift correction is applied rather than the standard error
        # of the mean of all the calibration standard normalized ratios
        rse_i_std = []
        for analyte in self.analytes:
            if "True" in self.calibration_std_stats.loc[analyte, "drift_correct"]:
                rse_i_std.append(
                    100
                    * self.calibration_std_stats.loc[analyte, "rmse"]
                    / self.calibration_std_stats.loc[analyte, "mean"]
                )
            else:
                rse_i_std.append(
                    self.calibration_std_stats.loc[analyte, "percent_std_err"]
                )

        rse_i_std = np.array(rse_i_std)

        for sample in self.secondary_standards:
            t1 = (
                self.standards_data.loc[sample, f"{self.int_std_element}_std"]
                / self.standards_data.loc[sample, f"{self.int_std_element}"]
            ) ** 2

            # concentration of internal standard in calibration standard uncertainties
            t2 = (
                self.standards_data.loc[
                    self.calibration_std, f"{self.int_std_element}_std"
                ]
                / self.standards_data.loc[
                    self.calibration_std, f"{self.int_std_element}"
                ]
            ) ** 2

            # concentration of each analyte in calibration standard uncertainties
            std_conc_stds = []
            for element in self.elements:
                # if our element is in the list of standard elements take the ratio
                if element in self.standard_elements:
                    std_conc_stds.append(
                        (
                            self.standards_data.loc[
                                self.calibration_std, f"{element}_std"
                            ]
                            / self.standards_data.loc[self.calibration_std, element]
                        )
                        ** 2
                    )

            std_conc_stds = np.array(std_conc_stds)

            # Overall uncertainties
            # Need to loop through each row?

            rel_ext_uncertainty = pd.DataFrame(
                np.sqrt(
                    np.array(
                        t1
                        + t2
                        + std_conc_stds
                        + (rse_i_std[np.newaxis, :] / 100) ** 2
                        + (self.data.loc[sample, myuncertainties].to_numpy() / 100) ** 2
                    ).astype(np.float64)
                )
            )
            rel_int_uncertainty = pd.DataFrame(
                np.sqrt(
                    np.array(
                        t1
                        # +t2
                        # + std_conc_stds
                        + (rse_i_std[np.newaxis, :] / 100) ** 2
                        + (self.data.loc[sample, myuncertainties].to_numpy() / 100) ** 2
                    ).astype(np.float64)
                )
            )
            rel_ext_uncertainty.columns = [f"{a}_exterr" for a in self.analytes]
            srm_rel_ext_uncertainties_list.append(rel_ext_uncertainty)
            rel_int_uncertainty.columns = [f"{a}_interr" for a in self.analytes]
            srm_rel_int_uncertainties_list.append(rel_int_uncertainty)

        srm_rel_ext_uncertainties = pd.concat(srm_rel_ext_uncertainties_list)
        srm_rel_int_uncertainties = pd.concat(srm_rel_int_uncertainties_list)

        srm_ext_uncertainties = pd.DataFrame(
            srm_rel_ext_uncertainties.values
            * self.SRM_concentrations.loc[:, self.analytes].values,
            columns=[f"{a}_exterr" for a in self.analytes],
            index=self.SRM_concentrations.index,
        )
        srm_int_uncertainties = pd.DataFrame(
            srm_rel_int_uncertainties.values
            * self.SRM_concentrations.loc[:, self.analytes].values,
            columns=[f"{a}_interr" for a in self.analytes],
            index=self.SRM_concentrations.index,
        )

        # Mask fringe analytes (SENTINEL_VALUE_NC) so uncertainties become NaN
        for analyte in self.analytes:
            mask = self.SRM_concentrations[analyte] == SENTINEL_VALUE_NC
            if mask.any():
                srm_ext_uncertainties.loc[mask, f"{analyte}_exterr"] = np.nan
                srm_int_uncertainties.loc[mask, f"{analyte}_interr"] = np.nan

        self.SRM_concentrations = pd.concat(
            [self.SRM_concentrations, srm_ext_uncertainties, srm_int_uncertainties],
            axis="columns",
        )

        ######################################

        for sample in self.samples_nostandards:
            # concentration of internal standard in unknown uncertainties
            int_std_element = re.split(
                r"(\d+)", self.calibration_std_data["norm"].unique()[0]
            )[2]
            # concentration of internal standard in unknown uncertainties
            t1 = (self.data.loc[sample, "int_std_rel_unc"] / 100) ** 2
            t1 = np.array(t1)
            t1 = t1[:, np.newaxis]

            # concentration of internal standard in calibration standard uncertainties
            t2 = (
                self.standards_data.loc[self.calibration_std, f"{int_std_element}_std"]
                / self.standards_data.loc[self.calibration_std, f"{int_std_element}"]
            ) ** 2

            # concentration of each analyte in calibration standard uncertainties
            std_conc_stds = []
            for element in self.elements:
                # # if our element is in the list of standard elements take the ratio
                if element in self.standard_elements:
                    std_conc_stds.append(
                        (
                            self.standards_data.loc[
                                self.calibration_std, f"{element}_std"
                            ]
                            / self.standards_data.loc[self.calibration_std, element]
                        )
                        ** 2
                    )

            std_conc_stds = np.array(std_conc_stds)

            # Overall uncertainties
            # Need to loop through each row?

            rel_ext_uncertainty = pd.DataFrame(
                np.sqrt(
                    np.array(
                        t1
                        + t2
                        + std_conc_stds
                        + (rse_i_std[np.newaxis, :] / 100) ** 2
                        + (self.data.loc[sample, myuncertainties].to_numpy() / 100) ** 2
                    ).astype(np.float64)
                )
            )
            rel_int_uncertainty = pd.DataFrame(
                np.sqrt(
                    np.array(
                        t1
                        # +t2
                        # + std_conc_stds
                        + (rse_i_std[np.newaxis, :] / 100) ** 2
                        + (self.data.loc[sample, myuncertainties].to_numpy() / 100) ** 2
                    ).astype(np.float64)
                )
            )
            rel_ext_uncertainty.columns = [f"{a}_exterr" for a in self.analytes]
            unk_rel_ext_uncertainties_list.append(rel_ext_uncertainty)
            rel_int_uncertainty.columns = [f"{a}_interr" for a in self.analytes]
            unk_rel_int_uncertainties_list.append(rel_int_uncertainty)

        unk_rel_ext_uncertainties = pd.concat(unk_rel_ext_uncertainties_list)
        unk_rel_int_uncertainties = pd.concat(unk_rel_int_uncertainties_list)

        unknown_ext_uncertainties = pd.DataFrame(
            unk_rel_ext_uncertainties.values
            * self.unknown_concentrations.loc[:, self.analytes].values,
            columns=[f"{a}_exterr" for a in self.analytes],
            index=self.unknown_concentrations.index,
        )

        unknown_int_uncertainties = pd.DataFrame(
            unk_rel_int_uncertainties.values
            * self.unknown_concentrations.loc[:, self.analytes].values,
            columns=[f"{a}_interr" for a in self.analytes],
            index=self.unknown_concentrations.index,
        )

        # Mask fringe analytes (SENTINEL_VALUE_NC) so uncertainties become NaN
        for analyte in self.analytes:
            mask = self.unknown_concentrations[analyte] == SENTINEL_VALUE_NC
            if mask.any():
                unknown_ext_uncertainties.loc[mask, f"{analyte}_exterr"] = np.nan
                unknown_int_uncertainties.loc[mask, f"{analyte}_interr"] = np.nan

        self.unknown_concentrations = pd.concat(
            [
                self.unknown_concentrations,
                unknown_ext_uncertainties,
                unknown_int_uncertainties,
            ],
            axis="columns",
        )

    # make an accuracy checking function
    # need to use analytes no mass to check SRM vals
    def get_secondary_standard_accuracies(self) -> None:
        """Calculate accuracy of secondary standard measurements.

        Accuracy is defined as ``100 * measured / accepted`` where
        *accepted* is the GEOREM preferred value for that SRM–analyte
        pair. Results are stored in ``self.SRM_accuracies``.

        Examples
        --------
        >>> concentrations.get_secondary_standard_accuracies()
        >>> concentrations.SRM_accuracies.head()
        """
        df_list = []

        for standard in self.secondary_standards:
            # need to go through column by column and check for bdl and then
            # replace with nan for numeric calculation. This explicit type declaration
            # is now required by pandas.

            nc_analytes = set()
            for analyte in self.analytes:
                if self.SRM_concentrations[analyte].dtype == "object":
                    if self.SRM_concentrations[analyte].str.contains("n.c.").any():
                        nc_analytes.add(analyte)
                    if self.SRM_concentrations[analyte].str.contains("b.d.l.").any():
                        ser = pd.to_numeric(
                            self.SRM_concentrations[analyte], errors="coerce"
                        )
                        self.SRM_concentrations[analyte] = ser

            df = pd.DataFrame(
                100
                * pd.to_numeric(self.SRM_concentrations.loc[standard, self.analytes].values.flatten(), errors="coerce").reshape(
                    self.SRM_concentrations.loc[standard, self.analytes].values.shape
                )
                / self.standards_data.loc[standard, self.elements].values[
                    np.newaxis, :
                ],
                columns=self.analytes,
                index=self.SRM_concentrations.loc[standard, :].index,
            ).fillna("b.d.l.")

            # Overwrite not-calibrated columns with "n.c."
            for analyte in nc_analytes:
                df[analyte] = "n.c."

            df.insert(0, "Spot", self.SRM_concentrations.loc[standard, "Spot"])
            if "timestamp" in self.data.columns:
                df.insert(
                    0, "timestamp", self.SRM_concentrations.loc[standard, "timestamp"]
                )
            else:
                df.insert(0, "index", self.SRM_concentrations.loc[standard, "index"])

            df_list.append(df)

        self.SRM_accuracies = pd.concat(df_list)

__init__(name, auto_load_srm=True)

Initialise a LaserCalc object.

Parameters:

Name	Type	Description	Default
name	str	Name for the experiment / processing run.	required
auto_load_srm	bool	If True (the default), automatically load the bundled GeoReM SRM database during construction. Set to False to retain the legacy behaviour where get_SRM_comps() must be called explicitly.	True

Source code in src/lasertram/calc/calc.py

def __init__(self, name: str, auto_load_srm: bool = True) -> None:
    """Initialise a LaserCalc object.

    Parameters
    ----------
    name : str
        Name for the experiment / processing run.
    auto_load_srm : bool, optional
        If True (the default), automatically load the bundled GeoReM
        SRM database during construction. Set to False to retain the
        legacy behaviour where get_SRM_comps() must be called explicitly.
    """
    # all attributes in relative chronological order that they are created in
    # if everything is done correctly. These all will get rewritten throughout the
    # data processing pipeline but this allows us to see what all the potential attributes
    # are going to be from the beginning (PEP convention)

    # for the math involved please see:

    # name for the lasercalc object
    # for notekeeping
    self.name = name

    # 2D pandas dataframe of standards reference material preferred compositions
    # from georem
    self.standards_data = None

    # List of standard reference materials in self.standards_data
    self.database_standards = None

    # list of standard reference material elements/oxides in self.standards_data
    self.standard_elements = None

    # list of standard reference material element/oxide 1 sigma uncertainties in self.standards_data
    self.standard_element_uncertainties = None

    # list of spot analyses for which concentrations are being calculated
    # this is the equivalent of self.data['Spot']
    self.spots = None

    # list of analytes for which concentrations are being calculated
    # these are column headers in self.data
    self.analytes = None

    # dict mapping each analyte to its corresponding _se column name
    self.analyte_se_map = None

    # ordered list of SE column names matching analyte order
    self.analyte_se_columns = None

    # 1 sigma standard deviation of the calibration standard values
    # in self.data. Is len(analytes) in shape
    self.calibration_std_stdevs = None

    # 2D pandas dataframe that represents the metadata and data for numerous
    # spot analyses. Each row is the equivalent of a LaserTRAM.output_report
    # and has the following columns:
    # |timestamp|Spot|despiked|omitted_region|bkgd_start|bkgd_stop|int_start|int_stop|norm|norm_cps|analyte vals and uncertainties -->|
    # |---------|----|--------|--------------|----------|---------|---------|--------|----|--------|----------------------------------|
    self.data = None

    # element used as internal standard. NOT to be confused with analyte
    # e.g. self.int_std_element == 'Si' NOT '29Si'
    self.int_std_element = None
    self.int_std_units = None

    # list of standard reference materials in found in self.data that are
    # also found in self.database_standards. This lets you know which standard reference
    # materials you can use as potential calibration standards
    self.potential_calibration_standards = None

    # list of samples in self.data with the self.potential_calibration_standards
    # removed
    self.samples_nostandards = None

    # list of elements for which concentrations are being calculated
    # this is the equivalent to self.analytes with the atomic masses
    # removed
    self.elements = None

    # string representing the standard reference material used
    # as the calibration standard for calculating concentrations
    self.calibration_std = None

    # 2D pandas dataframe which is a subset of self.data for only the
    # calibration standard data. This is essentially self.data.loc[self.calibration_std,:]
    self.calibration_std_data = None

    # mean calibration standard values for all analytes
    # equivalent of self.calibration_std_data.mean(axis = 0)
    self.calibration_std_means = None

    # calibration standard standard error of the mean for all analytes
    self.calibration_std_ses = None

    # 2D dataframe that is contains statistics for each analyte in self.calibration_std_data
    # columns are:
    # drift_correct | f_pval | f_value | f_crit_value | rmse | slope | intercept | mean | std_dev | percent_std_err
    # These stats are based on the following regression:
    # for each analyte
    # x = self.calibration_std_data.loc[:,'timestamp']
    # y = self.calibration_std_data.loc[:, analyte]

    # X = sm.add_constant(x)
    # Note the difference in argument order
    # model = sm.OLS(y, X).fit()
    # now generate predictions
    # ypred = model.predict(X)

    # calc rmse
    # RMSE = rmse(y, ypred)

    self.calibration_std_stats = None

    # the ratio of concentrations between an analyte and the internal standard
    # in the georem calibration standard values
    self.calibration_std_conc_ratios = None

    # list of standard reference materials that are not used as calibration standard
    # this is effectively self.potential_calibration_standards with self.calibration_std
    # removed
    self.secondary_standards = None

    # 2D pandas dataframe of calculated concentrations for all spots in self.secondary_standards and all
    # analytes in self.analytes. This is self.data.loc[self.secondary_standards,self.analytes].shape in shape
    self.SRM_concentrations = None

    # 2D pandas dataframe of calculated concentrations for all spots in self.spots and all
    # analytes in self.analytes. This is self.data.loc[self.spots,self.analytes].shape in shape
    self.unknown_concentrations = None

    # 2D pandas dataframe of calculated accuracies for all spots in self.secondary_standards and all
    # analytes in self.analytes. This is self.data.loc[self.secondary_standards,self.analytes].shape in shape
    # here accuracy is just 100*measured_concentration / georem_concentration
    self.SRM_accuracies = None

    if auto_load_srm:
        try:
            from ..helpers.preprocessing import load_srm_database

            self.get_SRM_comps(load_srm_database())
        except Exception as exc:
            warnings.warn(
                f"Auto-loading the bundled SRM database failed ({exc}). "
                "Call get_SRM_comps() manually to supply a standards database "
                "before running the pipeline.",
                UserWarning,
                stacklevel=2,
            )

get_SRM_comps(df)

Load a database of standard reference material compositions.

This call is optional when auto_load_srm=True. Calling this method after auto-load will replace the bundled database and emit a UserWarning.

Must be called before get_data() so that potential calibration standards can be identified.

Parameters:

Name	Type	Description	Default
df	DataFrame	DataFrame of standard reference materials where each row represents data for one SRM. The first column must be named "Standard". All other columns are elemental concentrations. Standard names must match those found in GEOREM.	required

Examples:

>>> import pandas as pd
>>> srm_data = pd.read_excel("laicpms_stds_tidy.xlsx")
>>> concentrations.get_SRM_comps(srm_data)

Source code in src/lasertram/calc/calc.py

def get_SRM_comps(self, df: pd.DataFrame) -> None:
    """Load a database of standard reference material compositions.

    This call is optional when ``auto_load_srm=True``. Calling this
    method after auto-load will replace the bundled database and emit
    a UserWarning.

    Must be called **before** ``get_data()`` so that potential
    calibration standards can be identified.

    Parameters
    ----------
    df : pandas.DataFrame
        DataFrame of standard reference materials where each row
        represents data for one SRM. The first column must be named
        ``"Standard"``. All other columns are elemental
        concentrations. Standard names must match those found in
        [GEOREM](http://georem.mpch-mainz.gwdg.de/).

    Examples
    --------
    >>> import pandas as pd
    >>> srm_data = pd.read_excel("laicpms_stds_tidy.xlsx")
    >>> concentrations.get_SRM_comps(srm_data)
    """
    # TODO: get_SRM_comps
    # - [x] add: check to make sure data are in right format else throw an error. do this by having a list of required columns and checking for them.

    if self.standards_data is not None:
        warnings.warn(
            "A standards database is already loaded (bundled database). "
            "Calling get_SRM_comps() is overriding it with the supplied DataFrame. "
            "If this is intentional, no action is needed.",
            UserWarning,
            stacklevel=2,
        )

    print(
        "checking uploaded standards database format for correct column headers and data types..."
    )
    col_check, type_check = formatting.check_srm_format(df)

    if col_check is not None:
        raise ValueError(
            f"It looks like your input data are missing the following required columns: {col_check}. Please fix before continuing with processing."
        )
    if type_check is not None:
        raise TypeError(
            f"It looks like your input data have incorrect types in the following columns: {type_check}. Please fix before continuing with processing."
        )
    self.standards_data = df.set_index("Standard")

    print("check complete...standards database format looks good!")

    self.database_standards = self.standards_data.index.unique().to_list()
    # Get a list of all of the elements supported in the published standard datasheet
    # Get a second list for the same elements but their corresponding uncertainty columns
    self.standard_elements = [
        analyte
        for analyte in self.standards_data.columns.tolist()
        if "_std" not in analyte
    ]
    self.standard_element_uncertainties = [
        analyte + "_std" for analyte in self.standard_elements
    ]

get_data(df, verbose=True)

Load LaserTRAM output for concentration calculations.

Automatically identifies potential calibration standards by comparing spot names to the SRM database loaded via get_SRM_comps().

Parameters:

Name	Type	Description	Default
df	DataFrame	Concatenated LaserTRAM.output_report rows for many spots.	required
verbose	bool	Whether to print status messages, by default True.	True

Examples:

>>> concentrations.get_data(processed_df)

Source code in src/lasertram/calc/calc.py

def get_data(self, df: pd.DataFrame, verbose: bool = True) -> None:
    """Load LaserTRAM output for concentration calculations.

    Automatically identifies potential calibration standards by
    comparing spot names to the SRM database loaded via
    ``get_SRM_comps()``.

    Parameters
    ----------
    df : pandas.DataFrame
        Concatenated ``LaserTRAM.output_report`` rows for
        many spots.
    verbose : bool, optional
        Whether to print status messages, by default ``True``.

    Examples
    --------
    >>> concentrations.get_data(processed_df)
    """
    # check if first row is nan (output from GUI does this).
    # If so, remove it
    df = df[df.iloc[:, 0].isna() == False]

    # TODO: get_data
    # - [x] add: check to make sure data are in right format else throw an error. do this by having a list of required columns and checking for them. this can't include analytes though - just the metadata.
    # - [x] add: in some checking for analytes to make sure that the measured analytes exist within the standards database. compare self.elements to self.standard_elements by going through each standard and checking for nan for that element
    # - [ ] add: duplicate name check.
    if verbose:
        print(
            "checking LaserCalc input data format for correct column headers and data types..."
        )

    col_check, type_check = formatting.check_lt_complete_format(df)
    if col_check is not None:
        raise ValueError(
            f"It looks like your input data are missing the following required columns: {col_check}. Please fix before continuing with processing."
        )
    if type_check is not None:
        raise TypeError(
            f"It looks like your input data have incorrect types in the following columns: {type_check}. Please fix before continuing with processing."
        )

    if verbose:
        print("check complete...input data format looks good!")

    data = df.set_index("Spot")
    data.insert(loc=1, column="index", value=np.arange(1, len(data) + 1))

    self.spots = data.index.unique().dropna().tolist()

    # Check for potential calibration standards. This will let us know what our options
    # are for choosing calibration standards by looking for spots that have the same string
    # as the standard spreadsheet

    stds_column = [
        [std for std in self.database_standards if std in spot]
        for spot in self.spots
    ]

    stds_column = [["unknown"] if not l else l for l in stds_column]

    stds_column = [std for sublist in stds_column for std in sublist]

    # standards that can be used as calibrations standards (must have more than 1 analysis)
    # potential_standards = list(np.unique(stds_column))
    potential_standards = [
        std for std in np.unique(stds_column) if stds_column.count(std) > 1
    ]
    potential_standards.remove("unknown")

    # all of the samples in your input sheet that are NOT potential standards
    all_standards = list(np.unique(stds_column))
    all_standards.remove("unknown")

    data["sample"] = stds_column

    data.reset_index(inplace=True)
    data.set_index("sample", inplace=True)

    self.data = data
    self.potential_calibration_standards = potential_standards
    self.samples_nostandards = list(np.setdiff1d(stds_column, all_standards))

    self.analytes = [
        col for col in data.columns.tolist()
        if ANALYTE_PATTERN.match(col)
    ]
    # elements without isotopes in the front
    self.elements = [re.split(r"(\d+)", analyte)[2] for analyte in self.analytes]

    # SE column pairing: map each analyte to its corresponding _se column
    analyte_set = set(self.analytes)
    self.analyte_se_map = {}

    for col in data.columns:
        if col.endswith('_se'):
            base = col[:-3]  # Remove '_se' suffix
            if base in analyte_set:
                self.analyte_se_map[base] = col

    # Ordered list of SE columns matching analyte order
    self.analyte_se_columns = [
        self.analyte_se_map[a] for a in self.analytes
        if a in self.analyte_se_map
    ]

    # first check to make sure that the element exists within the standards database
    for el in self.elements:
        if el not in self.standard_elements:
            raise ValueError(
                f"{el} is not in the standards database. Please remove it from your data before proceeding with processing."
            )

    if len(self.potential_calibration_standards) == 0:
        raise ValueError(
            f"No potential calibration standards found. The following spot names "
            f"were evaluated against the standards database: {list(np.unique(stds_column))}. "
            f"Ensure your data contains at least one SRM spot name that matches a "
            f"standard in the GeoReM database, with at least two analyses per standard."
        )

    print(
        f"Your potential calibration standards are: {[str(s) for s in self.potential_calibration_standards]}"
    )

    # internal standard analyte from lasertram
    self.int_std_element = re.split(r"(\d+)", self.data["norm"].unique()[0])[2]

set_calibration_standard(std)

Designate the calibration standard for concentration calculations.

Calculates mean values, standard deviations, and relative standard errors for the chosen SRM across all analytes.

Parameters:

Name	Type	Description	Default
std	str	Name of the standard reference material, e.g. "GSD-1G", "NIST-612", "BCR-2G".	required

Examples:

>>> concentrations.set_calibration_standard("GSD-1G")

Source code in src/lasertram/calc/calc.py

def set_calibration_standard(self, std: str) -> None:
    """Designate the calibration standard for concentration calculations.

    Calculates mean values, standard deviations, and relative
    standard errors for the chosen SRM across all analytes.

    Parameters
    ----------
    std : str
        Name of the standard reference material, e.g.
        ``"GSD-1G"``, ``"NIST-612"``, ``"BCR-2G"``.

    Examples
    --------
    >>> concentrations.set_calibration_standard("GSD-1G")
    """
    self.calibration_std = std

    self.calibration_std_data = self.data.loc[std, :]
    # Calibration standard information
    # mean
    self.calibration_std_means = self.calibration_std_data.loc[
        :, self.analytes
    ].mean()
    # std deviation
    self.calibration_std_stdevs = self.calibration_std_data.loc[
        :, self.analytes
    ].std()
    # relative standard error
    self.calibration_std_ses = 100 * (
        (self.calibration_std_stdevs / self.calibration_std_means)
        / np.sqrt(self.calibration_std_data.shape[0])
    )

    # Warn about fringe analytes with no published SRM value
    for analyte, element in zip(self.analytes, self.elements):
        if pd.isna(self.standards_data.loc[self.calibration_std, element]):
            warnings.warn(
                f"Analyte '{analyte}' (element '{element}') has no published "
                f"concentration in '{self.calibration_std}'. Concentrations for "
                f"this analyte will be reported as 'n.c.' (not calibrated). "
                f"Verify your SRM choice or switch to an SRM with a published value.",
                UserWarning,
                stacklevel=2,
            )

drift_check(pval=0.01)

Assess instrument drift for each analyte.

Performs a linear regression of each analyte's normalised ratio through time for the calibration standard. If the regression is statistically significant (F-test p-value < pval and F > F_crit), the analyte is flagged for drift correction in calculate_concentrations().

Results are stored in self.calibration_std_stats.

Parameters:

Name	Type	Description	Default
pval	float	Significance threshold to reject the null hypothesis of no drift, by default 0.01.	0.01

Examples:

>>> concentrations.drift_check()
>>> concentrations.calibration_std_stats.head()

Source code in src/lasertram/calc/calc.py

def drift_check(self, pval: float = 0.01) -> None:
    """Assess instrument drift for each analyte.

    Performs a linear regression of each analyte's normalised ratio
    through time for the calibration standard. If the regression is
    statistically significant (F-test *p*-value < ``pval`` **and**
    F > F_crit), the analyte is flagged for drift correction
    in ``calculate_concentrations()``.

    Results are stored in ``self.calibration_std_stats``.

    Parameters
    ----------
    pval : float, optional
        Significance threshold to reject the null hypothesis of no
        drift, by default 0.01.

    Examples
    --------
    >>> concentrations.drift_check()
    >>> concentrations.calibration_std_stats.head()
    """
    calib_std_rmses = []
    calib_std_slopes = []
    calib_std_intercepts = []
    drift_check = []

    f_pvals = []
    f_vals = []
    f_crits = []
    for analyte in self.analytes:
        # Getting regression statistics on analyte normalized ratios through time
        # for the calibration standard. This is what we use to check to see if it needs
        # to be drift corrected
        if "timestamp" in self.calibration_std_data.columns.tolist():
            # get an array in time units based on timestamp column. This is
            # is in seconds
            x = np.array(
                [
                    np.datetime64(d, "m")
                    for d in self.calibration_std_data["timestamp"]
                ]
            ).astype(np.float64)
            # x = np.cumsum(np.diff(x))
            # x = np.insert(x, 0, 0).astype(np.float64)

        else:
            x = self.calibration_std_data["index"].to_numpy()

        y = self.calibration_std_data.loc[:, analyte].astype("float64")

        X = sm.add_constant(x)
        # Note the difference in argument order
        model = sm.OLS(y, X).fit()
        # now generate predictions
        ypred = model.predict(X)

        # calc rmse
        RMSE = rmse(y, ypred)

        calib_std_rmses.append(RMSE)

        if model.params.shape[0] < 2:
            calib_std_slopes.append(model.params.loc["x1"])
            calib_std_intercepts.append(0)

        else:
            calib_std_slopes.append(model.params.loc["x1"])
            calib_std_intercepts.append(model.params.loc["const"])

        # new stuff
        # confidence limit 99%

        # f value stuff

        fvalue = model.fvalue
        f_vals.append(fvalue)
        f_pvalue = model.f_pvalue
        f_pvals.append(f_pvalue)
        fcrit = stats.f.ppf(q=1 - pval, dfn=len(x) - 1, dfd=len(y) - 1)
        f_crits.append(fcrit)
        if (f_pvalue < pval) and (fvalue > fcrit):
            drift = "True"
            drift_check.append(drift)
        else:
            drift = "False"
            drift_check.append(drift)

    self.calibration_std_stats = pd.DataFrame(
        {
            "drift_correct": drift_check,
            "f_pval": f_pvals,
            "f_value": f_vals,
            "f_crit_value": f_crits,
            "rmse": calib_std_rmses,
            "slope": calib_std_slopes,
            "intercept": calib_std_intercepts,
            "mean": self.calibration_std_means[self.analytes].to_numpy(),
            "std_dev": self.calibration_std_stdevs[self.analytes].to_numpy(),
            "percent_std_err": self.calibration_std_ses[self.analytes].to_numpy(),
        },
        index=self.analytes,
    )

get_calibration_std_ratios()

Calculate concentration ratios for the calibration standard.

For the chosen calibration standard, compute the ratio of each analyte's concentration to the internal standard element's concentration. Results are stored in self.calibration_std_conc_ratios.

Examples:

>>> concentrations.get_calibration_std_ratios()

Source code in src/lasertram/calc/calc.py

def get_calibration_std_ratios(self) -> None:
    """Calculate concentration ratios for the calibration standard.

    For the chosen calibration standard, compute the ratio of each
    analyte's concentration to the internal standard element's
    concentration. Results are stored in
    ``self.calibration_std_conc_ratios``.

    Examples
    --------
    >>> concentrations.get_calibration_std_ratios()
    """

    # For our calibration standard, calculate the concentration ratio
    # of each analyte to the element used as the internal standard
    std_conc_ratios = []

    for element in self.elements:
        if element in self.standard_elements:
            std_conc_ratios.append(
                self.standards_data.loc[self.calibration_std, element]
                / self.standards_data.loc[
                    self.calibration_std, self.int_std_element
                ]
            )

    # make our list an array for easier math going forward
    self.calibration_std_conc_ratios = np.array(std_conc_ratios)

set_int_std_concentrations(spots=None, concentrations=None, uncertainties=None, units='wt_per_ox')

Assign internal standard concentrations to spots.

A linear change in the concentration value produces a proportional linear change in the calculated concentration.

Parameters:

Name	Type	Description	Default
spots	Series or None	Spot names to assign. Corresponds to the Spot column from the LaserTRAM output.	None
concentrations	array - like or None	Internal standard concentration values. Must be the same length as spots.	None
uncertainties	array - like or None	Relative uncertainty in percent for the internal standard. Must be the same length as spots.	None
units	str	Units for the concentration values. One of "wt_per_ox" (weight percent oxide, default), "wt_per_el" (weight percent element), or "ppm_el" (ppm element).	'wt_per_ox'

Examples:

>>> concentrations.set_int_std_concentrations(
...     internal_std_comps["Spot"],
...     internal_std_comps["SiO2"],
...     internal_std_comps["SiO2_std%"],
... )

Source code in src/lasertram/calc/calc.py

def set_int_std_concentrations(
    self,
    spots: pd.Series | None = None,
    concentrations: npt.ArrayLike | None = None,
    uncertainties: npt.ArrayLike | None = None,
    units: str = "wt_per_ox",
) -> None:
    """Assign internal standard concentrations to spots.

    A linear change in the concentration value produces a
    proportional linear change in the calculated concentration.

    Parameters
    ----------
    spots : pandas.Series or None, optional
        Spot names to assign. Corresponds to the ``Spot`` column
        from the LaserTRAM output.
    concentrations : array-like or None, optional
        Internal standard concentration values. Must be the
        same length as ``spots``.
    uncertainties : array-like or None, optional
        Relative uncertainty in percent for the internal standard.
        Must be the same length as ``spots``.
    units : str, optional
        Units for the concentration values. One of ``"wt_per_ox"``
        (weight percent oxide, default), ``"wt_per_el"`` (weight
        percent element), or ``"ppm_el"`` (ppm element).

    Examples
    --------
    >>> concentrations.set_int_std_concentrations(
    ...     internal_std_comps["Spot"],
    ...     internal_std_comps["SiO2"],
    ...     internal_std_comps["SiO2_std%"],
    ... )
    """

    assert units in [
        "wt_per_ox",
        "wt_per_el",
        "ppm_el",
    ], f"{units} is not a supported unit for concentrations. accepted units are: ['wt_per_ox','wt_per_el', 'ppm_el']"

    if spots is None:
        spots = (self.data["Spot"],)
        concentrations = (np.full(self.data["Spot"].shape[0], 10),)
        uncertainties = (np.full(self.data["Spot"].shape[0], 1),)

    self.data["int_std_comp"] = 10.0
    self.data["int_std_rel_unc"] = 1.0
    df = self.data.reset_index().set_index("Spot")

    for spot, concentration, uncertainty in zip(
        spots, concentrations, uncertainties
    ):
        df.loc[spot, "int_std_comp"] = concentration
        df.loc[spot, "int_std_rel_unc"] = uncertainty

    self.data["int_std_comp"] = df["int_std_comp"].to_numpy()
    self.data["int_std_rel_unc"] = df["int_std_rel_unc"].to_numpy()

    self.int_std_units = units

calculate_concentrations()

Calculate concentrations and uncertainties for all spots.

Uses the calibration standard, internal standard concentrations, and drift correction information to compute absolute concentrations. Stores results in self.unknown_concentrations and self.SRM_concentrations.

Values below the detection limit are replaced with "b.d.l.".

Examples:

>>> concentrations.calculate_concentrations()
>>> concentrations.unknown_concentrations.head()

Source code in src/lasertram/calc/calc.py

def calculate_concentrations(self) -> None:
    """Calculate concentrations and uncertainties for all spots.

    Uses the calibration standard, internal standard concentrations,
    and drift correction information to compute absolute
    concentrations. Stores results in
    ``self.unknown_concentrations`` and ``self.SRM_concentrations``.

    Values below the detection limit are replaced with ``"b.d.l."``.

    Examples
    --------
    >>> concentrations.calculate_concentrations()
    >>> concentrations.unknown_concentrations.head()
    """

    secondary_standards = self.potential_calibration_standards.copy()
    secondary_standards.remove(self.calibration_std)
    self.secondary_standards = secondary_standards
    secondary_standards_concentrations_list = []
    unknown_concentrations_list = []

    for sample in secondary_standards:
        Cn_u = self.standards_data.loc[
            sample,
            re.split(
                r"(\d+)",
                self.calibration_std_data["norm"].unique()[0],
            )[2],
        ]
        Cin_std = self.calibration_std_conc_ratios
        Ni_std = self.calibration_std_stats["mean"][self.analytes]
        Ni_u = self.data.loc[sample, self.analytes]

        concentrations = Cn_u * (Cin_std / Ni_std) * Ni_u

        drift_concentrations_list = []

        for j, analyte, slope, intercept, drift in zip(
            range(len(self.analytes)),
            self.analytes,
            self.calibration_std_stats["slope"],
            self.calibration_std_stats["intercept"],
            self.calibration_std_stats["drift_correct"],
        ):
            if "True" in drift:
                if "timestamp" in self.data.columns.tolist():
                    frac = (
                        slope
                        * np.array(
                            [
                                np.datetime64(d, "m")
                                for d in self.data.loc[sample, "timestamp"]
                            ]
                        ).astype(np.float64)
                        + intercept
                    )
                else:
                    frac = slope * self.data.loc[sample, "index"] + intercept

                Ni_std = frac

                drift_concentrations = Cn_u * (Cin_std[j] / Ni_std) * Ni_u[analyte]

                if isinstance(drift_concentrations, np.float64):
                    df = pd.DataFrame(
                        np.array([drift_concentrations]), columns=[analyte]
                    )

                else:
                    df = pd.DataFrame(drift_concentrations, columns=[analyte])

                drift_concentrations_list.append(df)

        if len(drift_concentrations_list) > 0:
            drift_df = pd.concat(drift_concentrations_list, axis="columns")

            if drift_df.shape[0] == 1:
                drift_df["sample"] = sample
                drift_df.set_index("sample", inplace=True)
        else:
            drift_df = pd.DataFrame()

        for column in drift_df.columns.tolist():
            if isinstance(concentrations, pd.Series):
                concentrations.loc[column] = drift_df[column].to_numpy()[0]

            else:
                concentrations[column] = drift_df[column].to_numpy()

        if isinstance(concentrations, pd.Series):
            concentrations = pd.DataFrame(concentrations).T
            concentrations["sample"] = sample
            concentrations.set_index("sample", inplace=True)

        secondary_standards_concentrations_list.append(concentrations)

    ###############################

    for sample in self.samples_nostandards:
        # Cn_u = conversions.oxide_to_ppm(
        #     self.data.loc[sample, "int_std_comp"],
        #     self.data.loc[sample, "norm"].unique()[0],
        # ).to_numpy()
        int_std_element = "".join(
            [
                i
                for i in self.data.loc[sample, "norm"].unique()[0]
                if not i.isdigit()
            ]
        )

        # handle conversions from various units to all end up at ppm element
        if self.int_std_units == "wt_per_ox":
            Cn_u = conversions.oxide_to_ppm(
                self.data.loc[sample, "int_std_comp"], int_std_element
            )
        elif self.int_std_units == "wt_per_el":
            Cn_u = self.data.loc[sample, "int_std_comp"].values * 1e4
        elif self.int_std_units == "ppm_el":
            Cn_u = self.data.loc[sample, "int_std_comp"].values

        Cin_std = self.calibration_std_conc_ratios
        Ni_std = self.calibration_std_stats["mean"][self.analytes].to_numpy()
        Ni_u = self.data.loc[sample, self.analytes].to_numpy()

        concentrations = pd.DataFrame(
            Cn_u[:, np.newaxis] * (Cin_std / Ni_std) * Ni_u, columns=self.analytes
        )

        drift_concentrations_list = []

        for j, analyte, slope, intercept, drift in zip(
            range(len(self.analytes)),
            self.analytes,
            self.calibration_std_stats["slope"],
            self.calibration_std_stats["intercept"],
            self.calibration_std_stats["drift_correct"],
        ):
            if "True" in drift:
                if "timestamp" in self.data.columns.tolist():
                    frac = (
                        slope
                        * np.array(
                            [
                                np.datetime64(d, "m")
                                for d in self.data.loc[sample, "timestamp"]
                            ]
                        ).astype(np.float64)
                        + intercept
                    )
                else:
                    frac = slope * self.data.loc[sample, "index"] + intercept
                frac = np.array(frac)
                drift_concentrations = (
                    Cn_u[:, np.newaxis]
                    * (Cin_std[j] / frac)[:, np.newaxis]
                    * Ni_u[:, j][:, np.newaxis]
                )

                if isinstance(drift_concentrations, np.float64):
                    df = pd.DataFrame(
                        np.array([drift_concentrations]), columns=[analyte]
                    )

                else:
                    df = pd.DataFrame(drift_concentrations, columns=[analyte])

                drift_concentrations_list.append(df)

        if len(drift_concentrations_list) > 0:
            drift_df = pd.concat(drift_concentrations_list, axis="columns")

            if drift_df.shape[0] == 1:
                drift_df["sample"] = sample
                drift_df.set_index("sample", inplace=True)
        else:
            drift_df = pd.DataFrame()

        for column in drift_df.columns.tolist():
            if isinstance(concentrations, pd.Series):
                concentrations.loc[column] = drift_df[column].to_numpy()[0]

            else:
                concentrations[column] = drift_df[column].to_numpy()

        if isinstance(concentrations, pd.Series):
            concentrations = pd.DataFrame(concentrations).T
            concentrations["sample"] = sample
            concentrations.set_index("sample", inplace=True)

        unknown_concentrations_list.append(concentrations)

    self.SRM_concentrations = pd.concat(secondary_standards_concentrations_list)
    self.unknown_concentrations = pd.concat(unknown_concentrations_list)

    self.calculate_uncertainties()

    # Fill NaN concentrations with SENTINEL_VALUE_NC for fringe analytes
    for analyte in self.analytes:
        if self.unknown_concentrations[analyte].isna().any():
            self.unknown_concentrations[analyte] = self.unknown_concentrations[analyte].fillna(SENTINEL_VALUE_NC)
        if self.SRM_concentrations[analyte].isna().any():
            self.SRM_concentrations[analyte] = self.SRM_concentrations[analyte].fillna(SENTINEL_VALUE_NC)

    # INSERT IN SPOT METADATA NOW
    # OLD WAY OF REPLACING NEGATIVES. WILL THROW ERROR IN FUTURE FOR MIXING
    # STRINGS WITH FLOATS
    # self.unknown_concentrations[self.unknown_concentrations < 0] = "b.d.l."
    # self.SRM_concentrations[self.SRM_concentrations < 0] = "b.d.l."

    # THE NEW WAY OF DOING IT IS TO GO THROUGH COLUMN BY COLUMN AND CHECK FOR BELOW
    # 0 VALUES, CHANGE THE DTYPE TO OBJECT, AND THEN REPLACE THE NEGATIVE VALUES WITH BDL STRING
    # THEN CHANGE THE UNCERTAINTY VALUES TO BDL STRINGS BASED ON THE ROW WE DID FOR THE ACTUAL CONCENTRATION VALUE
    for analyte in self.analytes:
        # n.c. branch — MUST come before b.d.l. because SENTINEL_VALUE_NC < 0
        if any(self.unknown_concentrations[analyte] == SENTINEL_VALUE_NC):
            self.unknown_concentrations[analyte] = self.unknown_concentrations[analyte].astype("object")
            self.unknown_concentrations[f"{analyte}_interr"] = self.unknown_concentrations[f"{analyte}_interr"].astype("object")
            self.unknown_concentrations[f"{analyte}_exterr"] = self.unknown_concentrations[f"{analyte}_exterr"].astype("object")
            self.unknown_concentrations.loc[self.unknown_concentrations[analyte] == SENTINEL_VALUE_NC, analyte] = "n.c."
            self.unknown_concentrations.loc[self.unknown_concentrations[analyte] == "n.c.", f"{analyte}_interr"] = "n.c."
            self.unknown_concentrations.loc[self.unknown_concentrations[analyte] == "n.c.", f"{analyte}_exterr"] = "n.c."

        if any(self.SRM_concentrations[analyte] == SENTINEL_VALUE_NC):
            self.SRM_concentrations[analyte] = self.SRM_concentrations[analyte].astype("object")
            self.SRM_concentrations[f"{analyte}_interr"] = self.SRM_concentrations[f"{analyte}_interr"].astype("object")
            self.SRM_concentrations[f"{analyte}_exterr"] = self.SRM_concentrations[f"{analyte}_exterr"].astype("object")
            self.SRM_concentrations.loc[self.SRM_concentrations[analyte] == SENTINEL_VALUE_NC, analyte] = "n.c."
            self.SRM_concentrations.loc[self.SRM_concentrations[analyte] == "n.c.", f"{analyte}_interr"] = "n.c."
            self.SRM_concentrations.loc[self.SRM_concentrations[analyte] == "n.c.", f"{analyte}_exterr"] = "n.c."
        if self.unknown_concentrations[analyte].dtype != "object" and any(self.unknown_concentrations[analyte] < 0):
            self.unknown_concentrations[analyte] = self.unknown_concentrations[
                analyte
            ].astype("object")
            self.unknown_concentrations[f"{analyte}_interr"] = (
                self.unknown_concentrations[f"{analyte}_interr"].astype("object")
            )
            self.unknown_concentrations[f"{analyte}_exterr"] = (
                self.unknown_concentrations[f"{analyte}_exterr"].astype("object")
            )

            self.unknown_concentrations.loc[
                self.unknown_concentrations[analyte] < 0, analyte
            ] = "b.d.l."
            self.unknown_concentrations.loc[
                self.unknown_concentrations[analyte] == "b.d.l.",
                f"{analyte}_interr",
            ] = "b.d.l."
            self.unknown_concentrations.loc[
                self.unknown_concentrations[analyte] == "b.d.l", f"{analyte}_interr"
            ] = "b.d.l."

        if self.SRM_concentrations[analyte].dtype != "object" and any(self.SRM_concentrations[analyte] < 0):
            self.SRM_concentrations[analyte] = self.SRM_concentrations[
                analyte
            ].astype("object")
            self.SRM_concentrations[f"{analyte}_interr"] = self.SRM_concentrations[
                f"{analyte}_interr"
            ].astype("object")
            self.SRM_concentrations[f"{analyte}_exterr"] = self.SRM_concentrations[
                f"{analyte}_exterr"
            ].astype("object")

            self.SRM_concentrations.loc[
                self.SRM_concentrations[analyte] < 0, analyte
            ] = "b.d.l."
            self.SRM_concentrations.loc[
                self.SRM_concentrations[analyte] == "b.d.l.", f"{analyte}_interr"
            ] = "b.d.l."
            self.SRM_concentrations.loc[
                self.SRM_concentrations[analyte] == "b.d.l", f"{analyte}_interr"
            ] = "b.d.l."

    self.SRM_concentrations.insert(
        0, "Spot", list(self.data.loc[self.secondary_standards, "Spot"])
    )

    if "timestamp" in self.data.columns.tolist():
        self.SRM_concentrations.insert(
            0,
            "timestamp",
            list(self.data.loc[self.secondary_standards, "timestamp"]),
        )
    else:
        self.SRM_concentrations.insert(
            0, "index", list(self.data.loc[self.secondary_standards, "index"])
        )
    self.unknown_concentrations.insert(
        0, "Spot", list(self.data.loc[self.samples_nostandards, "Spot"])
    )
    if "timestamp" in self.data.columns.tolist():
        self.unknown_concentrations.insert(
            0,
            "timestamp",
            list(self.data.loc[self.samples_nostandards, "timestamp"]),
        )
    else:
        self.unknown_concentrations.insert(
            0, "index", list(self.data.loc[self.samples_nostandards, "index"])
        )

    self.unknown_concentrations.index = [
        "unknown"
    ] * self.unknown_concentrations.shape[0]
    self.unknown_concentrations.index.name = "sample"

calculate_uncertainties()

Calculate internal and external uncertainties for each analysis.

Called automatically by calculate_concentrations(). The results are appended as <analyte>_interr and <analyte>_exterr columns to self.unknown_concentrations and self.SRM_concentrations.

Source code in src/lasertram/calc/calc.py

def calculate_uncertainties(self) -> None:
    """Calculate internal and external uncertainties for each analysis.

    Called automatically by ``calculate_concentrations()``. The
    results are appended as ``<analyte>_interr`` and
    ``<analyte>_exterr`` columns to ``self.unknown_concentrations``
    and ``self.SRM_concentrations``.
    """

    myuncertainties = [analyte + "_se" for analyte in self.analytes]
    srm_rel_ext_uncertainties_list = []
    unk_rel_ext_uncertainties_list = []
    srm_rel_int_uncertainties_list = []
    unk_rel_int_uncertainties_list = []
    # use RMSE of regression for elements where drift correction is applied rather than the standard error
    # of the mean of all the calibration standard normalized ratios
    rse_i_std = []
    for analyte in self.analytes:
        if "True" in self.calibration_std_stats.loc[analyte, "drift_correct"]:
            rse_i_std.append(
                100
                * self.calibration_std_stats.loc[analyte, "rmse"]
                / self.calibration_std_stats.loc[analyte, "mean"]
            )
        else:
            rse_i_std.append(
                self.calibration_std_stats.loc[analyte, "percent_std_err"]
            )

    rse_i_std = np.array(rse_i_std)

    for sample in self.secondary_standards:
        t1 = (
            self.standards_data.loc[sample, f"{self.int_std_element}_std"]
            / self.standards_data.loc[sample, f"{self.int_std_element}"]
        ) ** 2

        # concentration of internal standard in calibration standard uncertainties
        t2 = (
            self.standards_data.loc[
                self.calibration_std, f"{self.int_std_element}_std"
            ]
            / self.standards_data.loc[
                self.calibration_std, f"{self.int_std_element}"
            ]
        ) ** 2

        # concentration of each analyte in calibration standard uncertainties
        std_conc_stds = []
        for element in self.elements:
            # if our element is in the list of standard elements take the ratio
            if element in self.standard_elements:
                std_conc_stds.append(
                    (
                        self.standards_data.loc[
                            self.calibration_std, f"{element}_std"
                        ]
                        / self.standards_data.loc[self.calibration_std, element]
                    )
                    ** 2
                )

        std_conc_stds = np.array(std_conc_stds)

        # Overall uncertainties
        # Need to loop through each row?

        rel_ext_uncertainty = pd.DataFrame(
            np.sqrt(
                np.array(
                    t1
                    + t2
                    + std_conc_stds
                    + (rse_i_std[np.newaxis, :] / 100) ** 2
                    + (self.data.loc[sample, myuncertainties].to_numpy() / 100) ** 2
                ).astype(np.float64)
            )
        )
        rel_int_uncertainty = pd.DataFrame(
            np.sqrt(
                np.array(
                    t1
                    # +t2
                    # + std_conc_stds
                    + (rse_i_std[np.newaxis, :] / 100) ** 2
                    + (self.data.loc[sample, myuncertainties].to_numpy() / 100) ** 2
                ).astype(np.float64)
            )
        )
        rel_ext_uncertainty.columns = [f"{a}_exterr" for a in self.analytes]
        srm_rel_ext_uncertainties_list.append(rel_ext_uncertainty)
        rel_int_uncertainty.columns = [f"{a}_interr" for a in self.analytes]
        srm_rel_int_uncertainties_list.append(rel_int_uncertainty)

    srm_rel_ext_uncertainties = pd.concat(srm_rel_ext_uncertainties_list)
    srm_rel_int_uncertainties = pd.concat(srm_rel_int_uncertainties_list)

    srm_ext_uncertainties = pd.DataFrame(
        srm_rel_ext_uncertainties.values
        * self.SRM_concentrations.loc[:, self.analytes].values,
        columns=[f"{a}_exterr" for a in self.analytes],
        index=self.SRM_concentrations.index,
    )
    srm_int_uncertainties = pd.DataFrame(
        srm_rel_int_uncertainties.values
        * self.SRM_concentrations.loc[:, self.analytes].values,
        columns=[f"{a}_interr" for a in self.analytes],
        index=self.SRM_concentrations.index,
    )

    # Mask fringe analytes (SENTINEL_VALUE_NC) so uncertainties become NaN
    for analyte in self.analytes:
        mask = self.SRM_concentrations[analyte] == SENTINEL_VALUE_NC
        if mask.any():
            srm_ext_uncertainties.loc[mask, f"{analyte}_exterr"] = np.nan
            srm_int_uncertainties.loc[mask, f"{analyte}_interr"] = np.nan

    self.SRM_concentrations = pd.concat(
        [self.SRM_concentrations, srm_ext_uncertainties, srm_int_uncertainties],
        axis="columns",
    )

    ######################################

    for sample in self.samples_nostandards:
        # concentration of internal standard in unknown uncertainties
        int_std_element = re.split(
            r"(\d+)", self.calibration_std_data["norm"].unique()[0]
        )[2]
        # concentration of internal standard in unknown uncertainties
        t1 = (self.data.loc[sample, "int_std_rel_unc"] / 100) ** 2
        t1 = np.array(t1)
        t1 = t1[:, np.newaxis]

        # concentration of internal standard in calibration standard uncertainties
        t2 = (
            self.standards_data.loc[self.calibration_std, f"{int_std_element}_std"]
            / self.standards_data.loc[self.calibration_std, f"{int_std_element}"]
        ) ** 2

        # concentration of each analyte in calibration standard uncertainties
        std_conc_stds = []
        for element in self.elements:
            # # if our element is in the list of standard elements take the ratio
            if element in self.standard_elements:
                std_conc_stds.append(
                    (
                        self.standards_data.loc[
                            self.calibration_std, f"{element}_std"
                        ]
                        / self.standards_data.loc[self.calibration_std, element]
                    )
                    ** 2
                )

        std_conc_stds = np.array(std_conc_stds)

        # Overall uncertainties
        # Need to loop through each row?

        rel_ext_uncertainty = pd.DataFrame(
            np.sqrt(
                np.array(
                    t1
                    + t2
                    + std_conc_stds
                    + (rse_i_std[np.newaxis, :] / 100) ** 2
                    + (self.data.loc[sample, myuncertainties].to_numpy() / 100) ** 2
                ).astype(np.float64)
            )
        )
        rel_int_uncertainty = pd.DataFrame(
            np.sqrt(
                np.array(
                    t1
                    # +t2
                    # + std_conc_stds
                    + (rse_i_std[np.newaxis, :] / 100) ** 2
                    + (self.data.loc[sample, myuncertainties].to_numpy() / 100) ** 2
                ).astype(np.float64)
            )
        )
        rel_ext_uncertainty.columns = [f"{a}_exterr" for a in self.analytes]
        unk_rel_ext_uncertainties_list.append(rel_ext_uncertainty)
        rel_int_uncertainty.columns = [f"{a}_interr" for a in self.analytes]
        unk_rel_int_uncertainties_list.append(rel_int_uncertainty)

    unk_rel_ext_uncertainties = pd.concat(unk_rel_ext_uncertainties_list)
    unk_rel_int_uncertainties = pd.concat(unk_rel_int_uncertainties_list)

    unknown_ext_uncertainties = pd.DataFrame(
        unk_rel_ext_uncertainties.values
        * self.unknown_concentrations.loc[:, self.analytes].values,
        columns=[f"{a}_exterr" for a in self.analytes],
        index=self.unknown_concentrations.index,
    )

    unknown_int_uncertainties = pd.DataFrame(
        unk_rel_int_uncertainties.values
        * self.unknown_concentrations.loc[:, self.analytes].values,
        columns=[f"{a}_interr" for a in self.analytes],
        index=self.unknown_concentrations.index,
    )

    # Mask fringe analytes (SENTINEL_VALUE_NC) so uncertainties become NaN
    for analyte in self.analytes:
        mask = self.unknown_concentrations[analyte] == SENTINEL_VALUE_NC
        if mask.any():
            unknown_ext_uncertainties.loc[mask, f"{analyte}_exterr"] = np.nan
            unknown_int_uncertainties.loc[mask, f"{analyte}_interr"] = np.nan

    self.unknown_concentrations = pd.concat(
        [
            self.unknown_concentrations,
            unknown_ext_uncertainties,
            unknown_int_uncertainties,
        ],
        axis="columns",
    )

get_secondary_standard_accuracies()

Calculate accuracy of secondary standard measurements.

Accuracy is defined as 100 * measured / accepted where accepted is the GEOREM preferred value for that SRM–analyte pair. Results are stored in self.SRM_accuracies.

Examples:

>>> concentrations.get_secondary_standard_accuracies()
>>> concentrations.SRM_accuracies.head()

Source code in src/lasertram/calc/calc.py

def get_secondary_standard_accuracies(self) -> None:
    """Calculate accuracy of secondary standard measurements.

    Accuracy is defined as ``100 * measured / accepted`` where
    *accepted* is the GEOREM preferred value for that SRM–analyte
    pair. Results are stored in ``self.SRM_accuracies``.

    Examples
    --------
    >>> concentrations.get_secondary_standard_accuracies()
    >>> concentrations.SRM_accuracies.head()
    """
    df_list = []

    for standard in self.secondary_standards:
        # need to go through column by column and check for bdl and then
        # replace with nan for numeric calculation. This explicit type declaration
        # is now required by pandas.

        nc_analytes = set()
        for analyte in self.analytes:
            if self.SRM_concentrations[analyte].dtype == "object":
                if self.SRM_concentrations[analyte].str.contains("n.c.").any():
                    nc_analytes.add(analyte)
                if self.SRM_concentrations[analyte].str.contains("b.d.l.").any():
                    ser = pd.to_numeric(
                        self.SRM_concentrations[analyte], errors="coerce"
                    )
                    self.SRM_concentrations[analyte] = ser

        df = pd.DataFrame(
            100
            * pd.to_numeric(self.SRM_concentrations.loc[standard, self.analytes].values.flatten(), errors="coerce").reshape(
                self.SRM_concentrations.loc[standard, self.analytes].values.shape
            )
            / self.standards_data.loc[standard, self.elements].values[
                np.newaxis, :
            ],
            columns=self.analytes,
            index=self.SRM_concentrations.loc[standard, :].index,
        ).fillna("b.d.l.")

        # Overwrite not-calibrated columns with "n.c."
        for analyte in nc_analytes:
            df[analyte] = "n.c."

        df.insert(0, "Spot", self.SRM_concentrations.loc[standard, "Spot"])
        if "timestamp" in self.data.columns:
            df.insert(
                0, "timestamp", self.SRM_concentrations.loc[standard, "timestamp"]
            )
        else:
            df.insert(0, "index", self.SRM_concentrations.loc[standard, "index"])

        df_list.append(df)

    self.SRM_accuracies = pd.concat(df_list)

Batch Processing

Helper for processing multiple spots in parallel.

lasertram.helpers.batch

Batch module.

Batch processing operations for LaserTRAM.

process_spot(spot, raw_data, bkgd, keep, int_std, omit=None, despike=False, output_report=True, verbose=False)

Process a single spot through the full LaserTRAM workflow.

Runs all methods of the LaserTRAM class on a single spot in a compact, batch-friendly call.

Parameters:

Name	Type	Description	Default
spot	LaserTRAM	An empty LaserTRAM object to be processed.	required
raw_data	DataFrame	Raw counts-per-second data for this spot. Shape (m, n) where m is the number of cycles through the mass range.	required
bkgd	tuple of float	(start, stop) analysis times defining the background interval.	required
keep	tuple of float	(start, stop) analysis times defining the signal interval.	required
int_std	str	Column name for the internal standard analyte (e.g., "29Si").	required
omit	tuple of float	(start, stop) analysis times to omit from the keep interval. Default is None.	None
despike	bool	Whether to despike all analyte signals using the standard deviation filter from LaserTRAM.despike_data(). Default is False.	False
output_report	bool	Whether to create a single-row pandas.DataFrame output report. Default is True.	True
verbose	bool	Whether to print status messages during processing. Default is False.	False

Source code in src/lasertram/helpers/batch.py

def process_spot(
    spot,
    raw_data,
    bkgd,
    keep,
    int_std,
    omit=None,
    despike=False,
    output_report=True,
    verbose = False
):
    """Process a single spot through the full LaserTRAM workflow.

    Runs all methods of the ``LaserTRAM`` class on a single spot in a
    compact, batch-friendly call.

    Parameters
    ----------
    spot : LaserTRAM
        An empty ``LaserTRAM`` object to be processed.
    raw_data : pandas.DataFrame
        Raw counts-per-second data for this spot. Shape ``(m, n)``
        where *m* is the number of cycles through the mass range.
    bkgd : tuple of float
        ``(start, stop)`` analysis times defining the background interval.
    keep : tuple of float
        ``(start, stop)`` analysis times defining the signal interval.
    int_std : str
        Column name for the internal standard analyte (e.g., ``"29Si"``).
    omit : tuple of float, optional
        ``(start, stop)`` analysis times to omit from the *keep* interval.
        Default is ``None``.
    despike : bool, optional
        Whether to despike all analyte signals using the standard
        deviation filter from ``LaserTRAM.despike_data()``.
        Default is ``False``.
    output_report : bool, optional
        Whether to create a single-row ``pandas.DataFrame`` output
        report. Default is ``True``.
    verbose : bool, optional
        Whether to print status messages during processing.
        Default is ``False``.
    """
    # assign data to the spot
    spot.get_data(raw_data, verbose = verbose)

    # assign the internal standard analyte
    spot.assign_int_std(int_std)
    # assign intervals for background and ablation signal
    spot.assign_intervals(bkgd=bkgd, keep=keep, omit=omit)
    # assign and save the median background values
    spot.get_bkgd_data()
    # remove the median background values from the ablation interval
    spot.subtract_bkgd()
    # calculate detection limits based off background values
    spot.get_detection_limits()
    # normalize the ablation interval to the internal standard analyte,
    # get the median values, and the standard error
    spot.normalize_interval()
    # despike the data if desired
    if despike is True:
        spot.despike_data(analyte_list="all")

    if output_report is True:
        spot.make_output_report()

Conversions

Oxide-to-element and element-to-oxide concentration conversions.

lasertram.helpers.conversions

Conversions module.

Convert between weight-percent oxide and parts-per-million element concentrations for common geologic oxide species.

See supported_internal_standard_oxides for the full list of supported oxides.

wt_percent_to_oxide(y, element)

Convert weight-percent element to weight-percent oxide.

Parameters:

Name	Type	Description	Default
y	int, float, list, numpy.ndarray, or pandas.Series	Concentration values in weight-percent element.	required
element	str	Cation symbol, e.g. "Si", "Ca", "Fe".	required

Returns:

Type	Description
int, float, or numpy.ndarray	Weight-percent oxide, same shape as y.

Examples:

>>> from lasertram.helpers import conversions
>>> conversions.wt_percent_to_oxide(46.83, "Si")
100.17...

Source code in src/lasertram/helpers/conversions.py

def wt_percent_to_oxide(
    y: int | float | list | npt.NDArray | pd.Series,
    element: str,
) -> int | float | npt.NDArray:
    """Convert weight-percent element to weight-percent oxide.

    Parameters
    ----------
    y : int, float, list, numpy.ndarray, or pandas.Series
        Concentration values in weight-percent element.
    element : str
        Cation symbol, e.g. ``"Si"``, ``"Ca"``, ``"Fe"``.

    Returns
    -------
    int, float, or numpy.ndarray
        Weight-percent oxide, same shape as ``y``.

    Examples
    --------
    >>> from lasertram.helpers import conversions
    >>> conversions.wt_percent_to_oxide(46.83, "Si")
    100.17...
    """
    assert (
        element in oxide_dict
    ), f"{element} is not a supported oxide for conversion to parts per million. please use conversions.supported_internal_standard_oxides for supported oxides"
    assert isinstance(
        y, (int, float, list, pd.core.series.Series, np.ndarray)
    ), "y should be something you can do math on - float, integer, list with numbers, numpy array, pandas Series"

    el_dict = oxide_dict[element]

    if isinstance(y, pd.Series):
        y = y.values

    wt_per_el = y

    return (el_dict["molecular_weight"] * wt_per_el) / (
        el_dict["cation_atomic_weight"] * el_dict["num_cations"]
    )

oxide_to_ppm(y, element)

Convert weight-percent oxide to parts-per-million element.

Parameters:

Name	Type	Description	Default
y	int, float, list, numpy.ndarray, or pandas.Series	Concentration values in weight-percent oxide.	required
element	str	Cation symbol, e.g. "Si", "Ca", "Fe".	required

Returns:

Type	Description
int, float, or numpy.ndarray	Concentration in ppm element, same shape as y.

Examples:

>>> from lasertram.helpers import conversions
>>> conversions.oxide_to_ppm(50.0, "Si")
233694...

Source code in src/lasertram/helpers/conversions.py

def oxide_to_ppm(
    y: int | float | list | npt.NDArray | pd.Series,
    element: str,
) -> int | float | npt.NDArray:
    """Convert weight-percent oxide to parts-per-million element.

    Parameters
    ----------
    y : int, float, list, numpy.ndarray, or pandas.Series
        Concentration values in weight-percent oxide.
    element : str
        Cation symbol, e.g. ``"Si"``, ``"Ca"``, ``"Fe"``.

    Returns
    -------
    int, float, or numpy.ndarray
        Concentration in ppm element, same shape as ``y``.

    Examples
    --------
    >>> from lasertram.helpers import conversions
    >>> conversions.oxide_to_ppm(50.0, "Si")
    233694...
    """

    assert (
        element in oxide_dict
    ), f"{element} is not a supported oxide for conversion to parts per million. please use conversions.supported_internal_standard_oxides for supported oxides"
    assert isinstance(
        y, (int, float, list, pd.core.series.Series, np.ndarray)
    ), "y should be something you can do math on - float, integer, list with numbers, numpy array, pandas Series"

    el_dict = oxide_dict[element]

    if isinstance(y, pd.Series):
        y = y.values

    wt_per_oxide = y

    return (
        1e4 * wt_per_oxide * el_dict["cation_atomic_weight"] * el_dict["num_cations"]
    ) / el_dict["molecular_weight"]

Formatting

Validation helpers for checking DataFrame column names and types at each stage of the lasertram workflow.

lasertram.helpers.formatting

Formatting module.

Validation helpers that check whether uploaded DataFrames conform to the expected column names and types for the various stages of the lasertram workflow.

check_srm_format(df)

Validate the format of an SRM composition database.

Parameters:

Name	Type	Description	Default
df	DataFrame	Candidate SRM database.	required

Returns:

Type	Description
tuple	(missing_columns, bad_type_indices) where each element is None when valid.

Source code in src/lasertram/helpers/formatting.py

def check_srm_format(df: pd.DataFrame) -> tuple[list[str] | None, list[int] | None]:
    """Validate the format of an SRM composition database.

    Parameters
    ----------
    df : pandas.DataFrame
        Candidate SRM database.

    Returns
    -------
    tuple
        ``(missing_columns, bad_type_indices)`` where each element is
        ``None`` when valid.
    """

    in_cols = df.columns.to_list()
    correct_cols = CORRECT_COLS

    type_map = {correct_cols[0]: 'str'}
    for col in correct_cols[1:]:
        type_map[col] = 'float'

    col_names_check = _check_cols(in_cols, correct_cols)
    if col_names_check is not None:
        return col_names_check, None
    col_types_check = _check_col_types(df, type_map)

    return col_names_check, col_types_check

check_lt_input_format(df)

Validate the format of a LaserTRAM input DataFrame.

Checks that the first three columns are SampleLabel, timestamp, and Time with appropriate types.

Parameters:

Name	Type	Description	Default
df	DataFrame	Candidate LaserTRAM input.	required

Returns:

Type	Description
tuple	(missing_columns, bad_type_indices).

Source code in src/lasertram/helpers/formatting.py

def check_lt_input_format(
    df: pd.DataFrame,
) -> tuple[list[str] | None, list[int] | None]:
    """Validate the format of a LaserTRAM input DataFrame.

    Checks that the first three columns are ``SampleLabel``,
    ``timestamp``, and ``Time`` with appropriate types.

    Parameters
    ----------
    df : pandas.DataFrame
        Candidate LaserTRAM input.

    Returns
    -------
    tuple
        ``(missing_columns, bad_type_indices)``.
    """
    in_cols = df.columns.to_list()
    in_cols = in_cols[:3]

    # columns and types to check for
    correct_cols = ['SampleLabel','timestamp','Time']
    type_map = {'SampleLabel': 'str', 'timestamp': 'datetime', 'Time': 'float'}

    col_names_check = _check_cols(in_cols, correct_cols)

    if col_names_check is not None:
        return col_names_check, None
    col_types_check = _check_col_types(df[in_cols], type_map)

    return col_names_check, col_types_check

check_lt_complete_format(df)

Validate the format of a LaserTRAM-complete DataFrame for LaserCalc.

Selects required columns by name (not position), so extra columns in the DataFrame are simply ignored.

Parameters:

Name	Type	Description	Default
df	DataFrame	Input DataFrame to be used as input to LaserCalc after processing in LaserTRAM.	required

Returns:

Type	Description
tuple of (list of str or None, list of str or None)	(missing_columns, type_errors). If columns are missing, returns (list_of_names, None). If types are wrong, returns (None, list_of_names). If all valid, returns (None, None).

Source code in src/lasertram/helpers/formatting.py

def check_lt_complete_format(df: pd.DataFrame) -> tuple[None | list[str], None | list[str]]:
    """Validate the format of a LaserTRAM-complete DataFrame for LaserCalc.

    Selects required columns by name (not position), so extra columns in the
    DataFrame are simply ignored.

    Parameters
    ----------
    df : pandas.DataFrame
        Input DataFrame to be used as input to LaserCalc after processing
        in LaserTRAM.

    Returns
    -------
    tuple of (list of str or None, list of str or None)
        ``(missing_columns, type_errors)``. If columns are missing,
        returns ``(list_of_names, None)``. If types are wrong, returns
        ``(None, list_of_names)``. If all valid, returns ``(None, None)``.
    """

    REQUIRED_COLS = [
        "timestamp", "Spot", "despiked", "omitted_region",
        "bkgd_start", "bkgd_stop", "int_start", "int_stop",
        "norm", "norm_cps",
    ]

    TYPE_MAP = {
        "timestamp": "datetime",
        "Spot": "str",
        "despiked": "bool_or_str",
        "omitted_region": "bool_or_str",
        "bkgd_start": "float",
        "bkgd_stop": "float",
        "int_start": "float",
        "int_stop": "float",
        "norm": "str",
        "norm_cps": "float",
    }

    in_cols = df.columns.tolist()
    col_names_check = _check_cols(in_cols, REQUIRED_COLS)

    if col_names_check is not None:
        # Skip type validation if columns are missing
        return col_names_check, None

    col_types_check = _check_col_types(df, TYPE_MAP)
    return None, col_types_check

check_duplicate_values(df, col, print_output=True)

Find duplicate values in a DataFrame column.

Parameters:

Name	Type	Description	Default
df	DataFrame	Input DataFrame.	required
col	str	Column name to check for duplicates.	required
print_output	bool	Whether to print a formatted table of duplicates, by default True.	True

Returns:

Type	Description
Series or None	Duplicate values and their indices, or None if no duplicates exist.

Source code in src/lasertram/helpers/formatting.py

def check_duplicate_values(
    df: pd.DataFrame, col: str, print_output: bool = True
) -> pd.Series | None:
    """Find duplicate values in a DataFrame column.

    Parameters
    ----------
    df : pandas.DataFrame
        Input DataFrame.
    col : str
        Column name to check for duplicates.
    print_output : bool, optional
        Whether to print a formatted table of duplicates, by default
        ``True``.

    Returns
    -------
    pandas.Series or None
        Duplicate values and their indices, or ``None`` if no
        duplicates exist.
    """

    assert isinstance(print_output, bool), "print_output must be boolean"
    assert isinstance(df, pd.core.frame.DataFrame), "df must be a pandas dataframe"
    assert (
        col in df.columns
    ), f"'{col}' is not in the input dataframe columns - please choose a column that exists in the dataframe"

    duplicates = df[col][df[col].duplicated(keep=False).values]
    if duplicates.shape[0] > 0:
        if print_output:

            print("duplicate sample names found:\n")
            print(tabulate(pd.DataFrame(duplicates), headers="keys", tablefmt="pipe"))
    else:
        duplicates = None
        if print_output:

            print(f"No duplicate values in column {col} found")

    return duplicates

rename_duplicate_values(df, col, print_output=True)

Rename duplicate values by appending -a, -b, etc.

Designed for string-valued columns such as sample names.

Parameters:

Name	Type	Description	Default
df	DataFrame	Input DataFrame.	required
col	str	Column containing duplicate values to rename.	required
print_output	bool	Whether to print renaming details, by default True.	True

Returns:

Type	Description
DataFrame	Copy of df with duplicates in col renamed.

Source code in src/lasertram/helpers/formatting.py

def rename_duplicate_values(
    df: pd.DataFrame, col: str, print_output: bool = True
) -> pd.DataFrame:
    """Rename duplicate values by appending ``-a``, ``-b``, etc.

    Designed for string-valued columns such as sample names.

    Parameters
    ----------
    df : pandas.DataFrame
        Input DataFrame.
    col : str
        Column containing duplicate values to rename.
    print_output : bool, optional
        Whether to print renaming details, by default ``True``.

    Returns
    -------
    pandas.DataFrame
        Copy of ``df`` with duplicates in ``col`` renamed.
    """

    assert isinstance(print_output, bool), "print_output must be boolean"
    assert isinstance(df, pd.core.frame.DataFrame), "df must be a pandas dataframe"
    assert (
        col in df.columns
    ), f"'{col}' is not in the input dataframe columns - please choose a column that exists in the dataframe"

    df_copy = df.copy()

    duplicates = check_duplicate_values(df_copy, col, print_output)

    if duplicates is not None:
        print("Renaming columns:")

        unique_duplicates = duplicates.unique()

        alphabet = list(string.ascii_lowercase)
        for duplicate in unique_duplicates:
            subset = df_copy[col][df_copy[col] == duplicate]
            old_vals = subset.values.copy()

            for val, letter in zip(range(len(subset)), alphabet):
                subset.iloc[val] = f"{subset.iloc[val]}-{letter}"

            new_vals = subset.values
            if print_output:

                print(
                    f"Rename {col} indices {subset.index.values} from {old_vals} ---> {new_vals} complete"
                )

            df_copy.loc[subset.index, col] = subset.values
    else:
        print("No duplicates to rename, returning copy of unmodified DataFrame")

    return df_copy

Plotting

Plotting utilities for time-series data and uncertainty visualisation.

lasertram.helpers.plotting

Plotting module.

Visualisation helpers for time-series LA-ICP-MS data.

plot_timeseries_data(df, analytes='all', marker='', fig=None, ax=None, **kwargs)

Plot time-series LA-ICP-MS data.

The x-axis is analysis time and the y-axis is counts per second (or a quantity derived from it). A legend is placed in a second axes panel to the right.

Parameters:

Name	Type	Description	Default
df	DataFrame	Data to plot. Must contain a "Time" column.	required
analytes	str or list of str	Column names to plot. "all" (default) plots every column except "timestamp" and "Time".	'all'
marker	str	Matplotlib marker style, by default "".	''
fig	Figure or None	Figure to draw on. A new figure is created when None.	None
ax	list of matplotlib.axes.Axes or None	A two-element list [data_ax, legend_ax]. Created automatically when None.	None
**kwargs		Additional keyword arguments passed to pandas.DataFrame.plot().	{}

Returns:

Type	Description
list of matplotlib.axes.Axes	[data_ax, legend_ax].

Examples:

>>> from lasertram.helpers import preprocessing, plotting
>>> import matplotlib.pyplot as plt
>>> plt.style.use("lasertram.lasertram")
>>> raw_data = preprocessing.load_test_rawdata()
>>> sample = "GSD-1G_-_1"
>>> ax = plotting.plot_timeseries_data(raw_data.loc[sample, :])
>>> ax[0].set_title(sample)

Source code in src/lasertram/helpers/plotting.py

def plot_timeseries_data(
    df: pd.DataFrame,
    analytes: str | list[str] = "all",
    marker: str = "",
    fig: matplotlib.figure.Figure | None = None,
    ax: list[matplotlib.axes.Axes] | None = None,
    **kwargs,
) -> list[matplotlib.axes.Axes]:
    """Plot time-series LA-ICP-MS data.

    The x-axis is analysis time and the y-axis is counts per second
    (or a quantity derived from it). A legend is placed in a second
    axes panel to the right.

    Parameters
    ----------
    df : pandas.DataFrame
        Data to plot. Must contain a ``"Time"`` column.
    analytes : str or list of str, optional
        Column names to plot. ``"all"`` (default) plots every column
        except ``"timestamp"`` and ``"Time"``.
    marker : str, optional
        Matplotlib marker style, by default ``""``.
    fig : matplotlib.figure.Figure or None, optional
        Figure to draw on. A new figure is created when ``None``.
    ax : list of matplotlib.axes.Axes or None, optional
        A two-element list ``[data_ax, legend_ax]``. Created
        automatically when ``None``.
    **kwargs
        Additional keyword arguments passed to
        ``pandas.DataFrame.plot()``.

    Returns
    -------
    list of matplotlib.axes.Axes
        ``[data_ax, legend_ax]``.

    Examples
    --------
    >>> from lasertram.helpers import preprocessing, plotting
    >>> import matplotlib.pyplot as plt
    >>> plt.style.use("lasertram.lasertram")
    >>> raw_data = preprocessing.load_test_rawdata()
    >>> sample = "GSD-1G_-_1"
    >>> ax = plotting.plot_timeseries_data(raw_data.loc[sample, :])
    >>> ax[0].set_title(sample)
    """

    if fig is None:
        fig = plt.figure(figsize=(8, 4))
    else:
        fig = plt.gcf()

    if ax is None:
        # setting up default axes
        rect = (0.1, 0.1, 0.8, 0.8)
        ax = [fig.add_axes(rect, label=f"{i}") for i in range(2)]

        horiz = [Size.AxesX(ax[0]), Size.Fixed(0.5), Size.AxesX(ax[1])]
        vert = [Size.AxesY(ax[0]), Size.Fixed(0.5), Size.AxesY(ax[1])]

        # divide the Axes rectangle into grid whose size is specified by horiz * vert
        divider = Divider(fig, rect, horiz, vert, aspect=False)
        ax[0].set_axes_locator(divider.new_locator(nx=0, ny=0))
        ax[1].set_axes_locator(divider.new_locator(nx=2, ny=0))

    if analytes == "all":
        analytes = [
            column
            for column in df.columns
            if ("timestamp" not in column) and ("Time" not in column)
        ]

        df.loc[:, ["Time"] + analytes].plot(
            x="Time",
            y=analytes,
            kind="line",
            marker=marker,
            ax=ax[0],
            lw=1,
            legend=False,
            **kwargs,
        )

    else:
        if isinstance(analytes, list):
            pass
        else:
            analytes = [analytes]

        df.loc[:, ["Time"] + analytes].plot(
            x="Time",
            y=analytes,
            kind="line",
            marker=marker,
            ax=ax[0],
            lw=1,
            legend=False,
            **kwargs,
        )

    ax[0].set_yscale("log")

    handles, labels = ax[0].get_legend_handles_labels()
    cols = 2
    ax[1].legend(
        handles, labels, loc="upper left", bbox_to_anchor=(0.15, 1.1), ncol=cols
    )
    ax[1].axis("off")

    return ax

plot_lasertram_uncertainties(spot, fig=None, ax=None, **kwargs)

plot a bar chart of analyte uncertainties related to the output from processing using the LaserTRAM module

Parameters:

Name	Type	Description	Default
spot	spot	the LaserTRAM.spot object to plot the uncertainties for	required
fig	Figure	The figure to apply the plot to, by default None	None
ax	Axes	the axis to apply the plot to, by default None	None

Returns:

Type	Description
ax

Source code in src/lasertram/helpers/plotting.py

def plot_lasertram_uncertainties(spot, fig=None, ax=None, **kwargs):
    """plot a bar chart of analyte uncertainties related to the output from
    processing using the `LaserTRAM` module

    Parameters
    ----------
    spot : LaserTRAM.spot
        the `LaserTRAM.spot` object to plot the uncertainties for
    fig : matplotlib.Figure, optional
        The figure to apply the plot to, by default None
    ax : matplotlib.Axes, optional
        the axis to apply the plot to, by default None

    Returns
    -------
    ax
    """

    if fig is None:
        fig = plt.figure(figsize=(12, 3))
    else:
        fig = plt.gcf()

    if ax is None:
        ax = fig.add_subplot()

    ax.bar(x=spot.analytes, height=spot.bkgd_subtract_std_err_rel, **kwargs)

    labels = [analyte for analyte in spot.analytes]
    labels = [
        "$^{{{}}}${}".format(
            re.findall(r"\d+", label)[0],
            label.replace(re.findall(r"\d+", label)[0], ""),
        )
        for label in labels
    ]
    ax.set_xticks(ax.get_xticks())
    ax.set_xticklabels(labels, rotation=90)
    ax.set_ylabel("% SE")

    return ax

Preprocessing

Functions for converting raw instrument output files into LaserTRAM-ready DataFrames, plus convenience loaders for bundled example data and the GeoReM SRM database.

lasertram.helpers.preprocessing

Preprocessing module.

Convert raw LA-ICP-MS .csv files from Agilent or ThermoFisher quadrupole instruments into a single pandas.DataFrame ready for processing with LaserTRAM.

extract_agilent_data(file)

Read raw output from an Agilent quadrupole .csv file.

Parameters:

Name	Type	Description	Default
file	str or Path	Path to the .csv file to be extracted.	required

Returns:

Type	Description
dict	Keys "timestamp", "file", "sample", "data".

Source code in src/lasertram/helpers/preprocessing.py

def extract_agilent_data(file: str | Path) -> dict:
    """Read raw output from an Agilent quadrupole ``.csv`` file.

    Parameters
    ----------
    file : str or pathlib.Path
        Path to the ``.csv`` file to be extracted.

    Returns
    -------
    dict
        Keys ``"timestamp"``, ``"file"``, ``"sample"``, ``"data"``.
    """
    # import data
    # extract sample name
    # extract timestamp
    # extract data and make headers ready for lasertram

    df = pd.read_csv(file, sep="\t", header=None)

    sample = df.iloc[0, 0].split("\\")[-1].split(".")[0].replace("_", "-")

    timestamp = parse(df.iloc[2, 0].split(" ")[7] + " " + df.iloc[2, 0].split(" ")[8])

    data = pd.DataFrame([sub.split(",") for sub in df.iloc[3:-1, 0]])

    header = data.iloc[0, :]
    data = data[1:]
    data.columns = header
    newcols = []
    for s in data.columns.tolist():
        l = re.findall(r"(\d+|[A-Za-z]+)", s)
        if "Time" in l:
            newcols.append(l[0])
        else:

            newcols.append(l[1] + l[0])
    data.columns = newcols

    return {"timestamp": timestamp, "file": file, "sample": sample, "data": data}

extract_thermo_data(file)

Read raw output from a ThermoFisher quadrupole .csv file.

Parameters:

Name	Type	Description	Default
file	str or Path	Path to the .csv file to be extracted.	required

Returns:

Type	Description
dict or None	Keys "timestamp", "file", "sample", "data". Returns None if the file is empty.

Source code in src/lasertram/helpers/preprocessing.py

def extract_thermo_data(file: str | Path) -> dict | None:
    """Read raw output from a ThermoFisher quadrupole ``.csv`` file.

    Parameters
    ----------
    file : str or pathlib.Path
        Path to the ``.csv`` file to be extracted.

    Returns
    -------
    dict or None
        Keys ``"timestamp"``, ``"file"``, ``"sample"``, ``"data"``.
        Returns ``None`` if the file is empty.
    """
    assert isinstance(file, (str, Path)), "file must be a str or pathlib.Path object"

    if isinstance(file, Path):
        pass
    else:
        file = Path(file)

    # gets the top row in your csv and turns it into a pandas series
    try:
        top = pd.read_csv(file, nrows=0)

        # since it is only 1 long it is also the column name
        # extract that as a list
        sample = list(top.columns)

        # turn that list value to a string
        sample = str(sample[0])

        # because its a string it can be split
        # split at : removes the time stamp
        sample = sample.split(":")[0]

        # .strip() removes leading and trailing spaces
        sample = sample.strip()

        # replace middle spaces with _ because spaces are bad
        nospace = sample.replace(" ", "_")

        # get the timestamp by splitting the string by the previously
        # designated sample. Also drops the colon in front of the date
        timestamp = top.columns.tolist()[0].split(sample)[1:][0][1:]

        timestamp = parse(timestamp)

        # import data
        # remove the top rows. Double check that your header is the specified
        # amount of rows to be skipped in 'skiprows' argument
        data = pd.read_csv(file, skiprows=13)
        # drop empty column at the end
        data.drop(data.columns[len(data.columns) - 1], axis=1, inplace=True)

        # remove dwell time row beneath header row
        data = data.dropna()

        return {"timestamp": timestamp, "file": file, "sample": nospace, "data": data}

    except pd.errors.EmptyDataError:
        return None

make_lt_ready_folder(folder, quad_type)

Combine a folder of raw .csv files into a LaserTRAM-ready DataFrame.

Parameters:

Name	Type	Description	Default
folder	str or Path	Path to the folder containing .csv files. The folder should contain only data files.	required
quad_type	str	"agilent" or "thermo".	required

Returns:

Type	Description
DataFrame	DataFrame ready for LaserTRAM.get_data().

Examples:

>>> from lasertram.helpers import preprocessing
>>> df = preprocessing.make_lt_ready_folder("path/to/csvs", "thermo")

Source code in src/lasertram/helpers/preprocessing.py

def make_lt_ready_folder(folder: str | Path, quad_type: str) -> pd.DataFrame:
    """Combine a folder of raw ``.csv`` files into a LaserTRAM-ready DataFrame.

    Parameters
    ----------
    folder : str or pathlib.Path
        Path to the folder containing ``.csv`` files. The folder should
        contain **only** data files.
    quad_type : str
        ``"agilent"`` or ``"thermo"``.

    Returns
    -------
    pandas.DataFrame
        DataFrame ready for ``LaserTRAM.get_data()``.

    Examples
    --------
    >>> from lasertram.helpers import preprocessing
    >>> df = preprocessing.make_lt_ready_folder("path/to/csvs", "thermo")
    """

    if isinstance(folder, Path):
        pass
    else:
        folder = Path(folder)
    assert (
        folder.is_dir()
    ), f"{folder} is not a directory, please choose a directory to your data .csv files"

    print(
        "Processing the all .csv files in the following folder for LaserTRAM input format:\n"
    )
    print(tabulate([[str(folder)]], headers=["Folder Path"], tablefmt="pipe"))
    print("\n")
    my_dict = {}
    files = [f for f in folder.glob("*.csv")]
    # GET METADATA
    # establish progress bar
    pbar = tqdm(
        files, desc="Extracting metadata from files", unit="file", total=len(files)
    )
    temp = None
    empty_files = []
    for i in pbar:
        # rename description to match file
        pbar.set_description(f"Extracting metadata from {i.name}")

        if quad_type == "thermo":
            temp = extract_thermo_data(i)

        elif quad_type == "agilent":
            temp = extract_agilent_data(i)

        # if the file is empty extract_thermo_data will be none
        if temp is None:
            empty_files.append(i)
        else:
            my_dict[temp["timestamp"]] = temp

    my_dict = dict(sorted(my_dict.items()))
    # if any empty files display them in a table
    if len(empty_files) > 0:
        print("the following files were skipped because they contained no data:\n")
        table = [[e.name] for e in empty_files]
        print(tabulate(table, headers=["Empty files"], tablefmt="pipe"))
        print("\n")
    # GET DATA FROM DICTIONARY AND CONCAT ALL THE DATA INTO ONE DF
    outdf = pd.DataFrame()
    pbar2 = tqdm(my_dict, desc="Combining individual files", unit="file")
    for timestamp in pbar2:

        pbar2.set_description(
            f"Adding data from {Path(my_dict[timestamp]['file']).name}"
        )

        samplelabel = pd.DataFrame(
            np.repeat(
                my_dict[timestamp]["sample"], my_dict[timestamp]["data"].shape[0]
            ),
            columns=["SampleLabel"],
            index=my_dict[timestamp]["data"].index,
        )
        ts = pd.DataFrame(
            np.repeat(
                my_dict[timestamp]["timestamp"], my_dict[timestamp]["data"].shape[0]
            ),
            columns=["timestamp"],
            index=my_dict[timestamp]["data"].index,
        )
        df = pd.concat([ts, samplelabel, my_dict[timestamp]["data"]], axis="columns")

        outdf = pd.concat([outdf, df])
        outdf.index = np.arange(outdf.shape[0], dtype=int)
    print("Success.\n")
    return outdf

make_lt_ready_file(file, quad_type)

Convert a single raw .csv file into a LaserTRAM-ready DataFrame.

Parameters:

Name	Type	Description	Default
file	str or Path	Path to the .csv file.	required
quad_type	str	"agilent" or "thermo".	required

Returns:

Type	Description
DataFrame	DataFrame ready for LaserTRAM.get_data().

Raises:

Type	Description
ValueError	If quad_type is not "thermo" or "agilent".

Source code in src/lasertram/helpers/preprocessing.py

def make_lt_ready_file(file: str | Path, quad_type: str) -> pd.DataFrame:
    """Convert a single raw ``.csv`` file into a LaserTRAM-ready DataFrame.

    Parameters
    ----------
    file : str or pathlib.Path
        Path to the ``.csv`` file.
    quad_type : str
        ``"agilent"`` or ``"thermo"``.

    Returns
    -------
    pandas.DataFrame
        DataFrame ready for ``LaserTRAM.get_data()``.

    Raises
    ------
    ValueError
        If ``quad_type`` is not ``"thermo"`` or ``"agilent"``.
    """

    if isinstance(file, Path):
        pass
    else:
        file = Path(file)

    assert file.name.endswith(".csv"), f"File '{file}' does not have a CSV extension."

    if quad_type == "thermo":
        temp = extract_thermo_data(file)

    elif quad_type == "agilent":
        temp = extract_agilent_data(file)
    else:
        temp = None

    if temp:
        outdf = temp["data"]
        outdf.insert(0, "SampleLabel", temp["sample"])
        outdf.insert(0, "timestamp", temp["timestamp"])

    else:
        raise ValueError("please choose either 'thermo' or 'agilent' for quad_type")

    return outdf

load_test_rawdata()

Load example raw LA-ICP-MS data.

The data accompany the following manuscript:

Lubbers, J., Kent, A., & Russo, C. (2025). lasertram: a Python
library for time resolved analysis of laser ablation inductively
coupled plasma mass spectrometry data.

Returns:

Type	Description
DataFrame	Raw counts-per-second data indexed by SampleLabel.

Examples:

>>> from lasertram.helpers import preprocessing
>>> raw_data = preprocessing.load_test_rawdata()
>>> raw_data.head()

Source code in src/lasertram/helpers/preprocessing.py

def load_test_rawdata() -> pd.DataFrame:
    """Load example raw LA-ICP-MS data.

    The data accompany the following manuscript:

        Lubbers, J., Kent, A., & Russo, C. (2025). lasertram: a Python
        library for time resolved analysis of laser ablation inductively
        coupled plasma mass spectrometry data.

    Returns
    -------
    pandas.DataFrame
        Raw counts-per-second data indexed by ``SampleLabel``.

    Examples
    --------
    >>> from lasertram.helpers import preprocessing
    >>> raw_data = preprocessing.load_test_rawdata()
    >>> raw_data.head()
    """

    # current_path = Path(__file__).parent
    # lt_ready = pd.read_excel(
    #     current_path.parents[1]
    #     / "test_data"
    #     / "computers_and_geosciences_examples"
    #     / "2022-05-10_LT_ready.xlsx"
    # ).set_index("SampleLabel")

    lt_ready = pd.read_excel(
        data_examples_dir
        / "computers_and_geosciences_examples"
        / "2022-05-10_LT_ready.xlsx"
    ).set_index("SampleLabel")

    return lt_ready

load_test_intervals()

Load example interval selections.

The data accompany the following manuscript:

Lubbers, J., Kent, A., & Russo, C. (2025). lasertram: a Python
library for time resolved analysis of laser ablation inductively
coupled plasma mass spectrometry data.

Returns:

Type	Description
DataFrame	Interval definitions indexed by Spot.

Examples:

>>> from lasertram.helpers import preprocessing
>>> intervals = preprocessing.load_test_intervals()

Source code in src/lasertram/helpers/preprocessing.py

def load_test_intervals() -> pd.DataFrame:
    """Load example interval selections.

    The data accompany the following manuscript:

        Lubbers, J., Kent, A., & Russo, C. (2025). lasertram: a Python
        library for time resolved analysis of laser ablation inductively
        coupled plasma mass spectrometry data.

    Returns
    -------
    pandas.DataFrame
        Interval definitions indexed by ``Spot``.

    Examples
    --------
    >>> from lasertram.helpers import preprocessing
    >>> intervals = preprocessing.load_test_intervals()
    """

    # current_path = Path(__file__).parent

    # intervals = pd.read_excel(
    #     current_path.parents[1]
    #     / "test_data"
    #     / "computers_and_geosciences_examples"
    #     / "example_intervals.xlsx"
    # ).set_index("Spot")

    intervals = pd.read_excel(
        data_examples_dir
        / "computers_and_geosciences_examples"
        / "example_intervals.xlsx"
    ).set_index("Spot")

    return intervals

load_test_int_std_comps()

Load example internal standard compositions.

The data accompany the following manuscript:

Lubbers, J., Kent, A., & Russo, C. (2025). lasertram: a Python
library for time resolved analysis of laser ablation inductively
coupled plasma mass spectrometry data.

Returns:

Type	Description
DataFrame	Internal standard concentrations and uncertainties.

Examples:

>>> from lasertram.helpers import preprocessing
>>> int_std_comps = preprocessing.load_test_int_std_comps()

Source code in src/lasertram/helpers/preprocessing.py

def load_test_int_std_comps() -> pd.DataFrame:
    """Load example internal standard compositions.

    The data accompany the following manuscript:

        Lubbers, J., Kent, A., & Russo, C. (2025). lasertram: a Python
        library for time resolved analysis of laser ablation inductively
        coupled plasma mass spectrometry data.

    Returns
    -------
    pandas.DataFrame
        Internal standard concentrations and uncertainties.

    Examples
    --------
    >>> from lasertram.helpers import preprocessing
    >>> int_std_comps = preprocessing.load_test_int_std_comps()
    """

    # current_path = Path(__file__).parent

    # concentrations = pd.read_excel(
    #     current_path.parents[1]
    #     / "test_data"
    #     / "computers_and_geosciences_examples"
    #     / "example_internal_std.xlsx"
    # )

    concentrations = pd.read_excel(
        data_examples_dir
        / "computers_and_geosciences_examples"
        / "example_internal_std.xlsx"
    )

    return concentrations

load_srm_database()

Load the bundled GeoReM SRM composition database.

Returns the database as a DataFrame constructed entirely from the in-memory SRM_DATABASE dict — no file I/O is performed.

Returns:

Type	Description
DataFrame	SRM compositions with columns matching CORRECT_COLS.

Source code in src/lasertram/helpers/preprocessing.py

def load_srm_database() -> pd.DataFrame:
    """Load the bundled GeoReM SRM composition database.

    Returns the database as a DataFrame constructed entirely from the
    in-memory SRM_DATABASE dict — no file I/O is performed.

    Returns
    -------
    pandas.DataFrame
        SRM compositions with columns matching CORRECT_COLS.
    """
    from ..data.srm_database import SRM_DATABASE

    df = pd.DataFrame(SRM_DATABASE)
    # Ensure numeric columns are float64 (all-None columns default to object)
    non_std_cols = [c for c in df.columns if c != "Standard"]
    df[non_std_cols] = df[non_std_cols].apply(pd.to_numeric, errors="coerce")
    return df