Reference

This part of the project documentation shows the technical implementation of the CaloCem project code.

DownSamplingParameters dataclass

Parameters for adaptive downsampling of the heat flow data.

Parameters:

Name	Type	Description	Default
apply	bool	If True, adaptive downsampling is applied to the heat flow data.	False
num_points	int	The target number of points after downsampling. Default is 1000.	1000
smoothing_factor	float	Smoothing factor used in the downsampling algorithm. Default is 1e-10.	1e-10
baseline_weight	float	Weight of the baseline in the downsampling algorithm. Default is 0.1.	0.1
section_split	bool	If True, the data is split into sections for downsampling. This can be useful if there is a narrow peak in the data early on. Default is False.	False
section_split_time_s	int	Time in seconds for splitting the data into two sections. Default is 1000 seconds.	1000

Source code in calocem/processparams.py

@dataclass
class DownSamplingParameters:
    """
    Parameters for adaptive downsampling of the heat flow data.

    Parameters
    ----------
    apply: bool
        If True, adaptive downsampling is applied to the heat flow data.
    num_points: int
        The target number of points after downsampling. Default is 1000.
    smoothing_factor: float
        Smoothing factor used in the downsampling algorithm. Default is 1e-10.
    baseline_weight: float
        Weight of the baseline in the downsampling algorithm. Default is 0.1.
    section_split: bool
        If True, the data is split into sections for downsampling. This can be useful if there is a narrow peak in the data early on. Default is False.
    section_split_time_s: int
        Time in seconds for splitting the data into two sections. Default is 1000 seconds.
    """
    apply: bool = False
    num_points: int = 1000
    smoothing_factor: float = 1e-10
    baseline_weight: float = 0.1
    section_split: bool = False
    section_split_time_s: int = 1000

GradientPeakDetectionParameters dataclass

Parameters that control the identifcation of Peaks in the first derivative (gradient) of the heat flow data. Under the hood the SciPy find_peaks() is used Link to SciPy method.

Parameters:

Name	Type	Description	Default
prominence	float	Minimum prominence of the peak	1e-09
distance	int	Minimum distance	100
width	int	Minimum width	20
rel_height	float	Relative height of the peak	0.05
height	float	Minimum height of the peak	1e-09
use_first	bool	If true, only the first peak will be used	False
use_largest_width	bool	If true the peak with the largest peak width will be used.	False

Source code in calocem/processparams.py

@dataclass
class GradientPeakDetectionParameters:
    """
    Parameters that control the identifcation of Peaks in the first derivative (gradient) of the heat flow data. Under the hood the SciPy `find_peaks()` is used [Link to SciPy method](https://docs.scipy.org/doc/scipy/reference/generated/scipy.signal.find_peaks.html).

    Parameters
    ----------
    prominence: float
        Minimum prominence of the peak
    distance: int
        Minimum distance
    width: int
        Minimum width
    rel_height: float
        Relative height of the peak
    height: float
        Minimum height of the peak
    use_first: bool
        If true, only the first peak will be used
    use_largest_width: bool
        If true the peak with the largest peak width will be used.
    """

    prominence: float = 1e-9
    distance: int = 100
    width: int = 20
    rel_height: float = 0.05
    height: float = 1e-9
    use_first: bool = False
    use_largest_width: bool = False
    use_largest_width_height: bool = False

Measurement

A base class for handling and processing isothermal heat flow calorimetry data.

Currently supported file formats are .xls and .csv files. Only TA Instruments data files are supported at the moment.

Parameters:

Name	Type	Description	Default
folder	str	path to folder containing .xls and/or .csv experimental result files. The default is None.	None
show_info	bool	whether or not to print some informative lines during code execution. The default is True.	True
regex	str	regex pattern to include only certain experimental result files during initialization. The default is None.	None
auto_clean	bool	whether or not to exclude NaN values contained in the original files and combine data from differently names temperature columns. The default is False.	False
cold_start	bool	whether or not to use "pickled" files for initialization; save time on reading	True

Examples:

>>> import CaloCem as ta
>>> from pathlib import Path
>>>
>>> calodatapath = Path(__file__).parent
>>> tam = ta.Measurement(folder=calodatapath, show_info=True)

We can use a regex pattern to only include certain files in the datafolder. Here we assume that we only want to load .csv files which contain the string "bm".

>>> tam = ta.Measurement(folder=calodatapath, regex=r".*bm.*.csv", show_info=True)

Source code in calocem/tacalorimetry.py

class Measurement:
    """
    A base class for handling and processing isothermal heat flow calorimetry data.

    Currently supported file formats are .xls and .csv files.
    Only TA Instruments data files are supported at the moment.

    Parameters
    ----------
    folder : str, optional
        path to folder containing .xls and/or .csv experimental result
        files. The default is None.
    show_info : bool, optional
        whether or not to print some informative lines during code
        execution. The default is True.
    regex : str, optional
        regex pattern to include only certain experimental result files
        during initialization. The default is None.
    auto_clean : bool, optional
        whether or not to exclude NaN values contained in the original
        files and combine data from differently names temperature columns.
        The default is False.
    cold_start : bool, optional
        whether or not to use "pickled" files for initialization; save time
        on reading

    Examples
    --------

    >>> import CaloCem as ta
    >>> from pathlib import Path
    >>>
    >>> calodatapath = Path(__file__).parent
    >>> tam = ta.Measurement(folder=calodatapath, show_info=True)

    We can use a regex pattern to only include certain files in the datafolder. Here we assume that we only want to load .csv files which contain the string "bm".

    >>> tam = ta.Measurement(folder=calodatapath, regex=r".*bm.*.csv", show_info=True)

    """

    # init
    _info = pd.DataFrame()
    _data = pd.DataFrame()
    _data_unprocessed = (
        pd.DataFrame()
    )  # helper to store experimental data as loaded from files

    # further metadata
    _metadata = pd.DataFrame()
    _metadata_id = ""

    # define pickle filenames
    _file_data_pickle = pathlib.Path().cwd() / "_data.pickle"
    _file_info_pickle = pathlib.Path().cwd() / "_info.pickle"

    #
    # init
    #
    def __init__(
        self,
        folder=None,
        show_info=True,
        regex=None,
        auto_clean=False,
        cold_start=True,
        processparams=None,
        new_code=False,
        processed=False,
    ):
        """
        intialize measurements from folder


        """
        self._new_code = new_code
        self._processed = processed

        if not isinstance(processparams, ProcessingParameters):
            self.processparams = ProcessingParameters()
        else:
            self.processparams = processparams

        # read
        if folder:
            if cold_start:
                # get data and parameters
                self._get_data_and_parameters_from_folder(
                    folder, regex=regex, show_info=show_info
                )
            else:
                # get data and parameters from pickled files
                self._get_data_and_parameters_from_pickle()
            try:
                if auto_clean:
                    # remove NaN values and merge time columns
                    self._auto_clean_data()
            except Exception as e:
                # info
                print(e)
                raise AutoCleanException
                # return
                return

        if self.processparams.downsample.apply:
            self._apply_adaptive_downsampling()
        # Message
        print(
            "================\nAre you missing some samples? Try rerunning with auto_clean=True and cold_start=True.\n================="
        )

    #
    # get_data_and_parameters_from_folder
    #
    def _get_data_and_parameters_from_folder(self, folder, regex=None, show_info=True):
        """
        get_data_and_parameters_from_folder
        """

        if not isinstance(folder, str):
            # convert
            folder = str(folder)

        # loop folder
        for f in os.listdir(folder):
            if not f.endswith((".xls", ".csv")):
                # go to next
                continue

            if regex:
                # check match
                if not re.match(regex, f):
                    # skip this file
                    continue

            # info
            if show_info:
                print(f"Reading {f}.")

            # define file
            file = folder + os.sep + f

            # check xls
            if f.endswith(".xls"):
                if self._new_code:
                    self._data = pd.concat(
                        [
                            self._data,
                            self._read_csv_data(file, show_info=show_info),
                        ]
                    )
                if self._new_code is False:
                    # collect information
                    try:
                        self._info = pd.concat(
                            [
                                self._info,
                                self._read_calo_info_xls(file, show_info=show_info),
                            ]
                        )
                    except Exception:
                        # initialize
                        if self._info.empty:
                            self._info = self._read_calo_info_xls(
                                file, show_info=show_info
                            )

                    # collect data
                    try:
                        self._data = pd.concat(
                            [
                                self._data,
                                self._read_calo_data_xls(file, show_info=show_info),
                            ]
                        )

                    except Exception:
                        # initialize
                        if self._data.empty:
                            self._data = self._read_calo_data_xls(
                                file, show_info=show_info
                            )

            # append csv
            if f.endswith(".csv"):
                # collect data
                if self._new_code:
                    self._data = pd.concat(
                        [
                            self._data,
                            self._read_csv_data(file, show_info=show_info),
                        ]
                    )
                    # self._read_csv_data(file, show_info=show_info)
                if self._new_code is False:
                    try:
                        self._data = pd.concat(
                            [
                                self._data,
                                self._read_calo_data_csv(file, show_info=show_info),
                            ]
                        )

                    except Exception:
                        # initialize
                        if self._data.empty:
                            self._data = self._read_calo_data_csv(
                                file, show_info=show_info
                            )

                # collect information
                try:
                    self._info = pd.concat(
                        [
                            self._info,
                            self._read_calo_info_csv(file, show_info=show_info),
                        ]
                    )
                except Exception:
                    # initialize
                    if self._info.empty:
                        try:
                            self._info = self._read_calo_info_csv(
                                file, show_info=show_info
                            )
                        except Exception:
                            pass

        # get "heat_j" columns if the column is not part of the source files
        if self.processparams.preprocess.infer_heat:
            try:
                self._infer_heat_j_column()
            except Exception:
                pass

        # if self.processparams.downsample.apply is not False:
        #     self._apply_adaptive_downsampling()
        # write _data and _info to pickle
        with open(self._file_data_pickle, "wb") as f:
            pickle.dump(self._data, f)
        with open(self._file_info_pickle, "wb") as f:
            pickle.dump(self._info, f)

        # store experimental data for recreating state after reading from files
        self._data_unprocessed = self._data.copy()

    #
    # get data and information from pickled files
    #
    def _get_data_and_parameters_from_pickle(self):
        """
        get data and information from pickled files

        Returns
        -------
        None.

        """

        # read from pickle
        try:
            self._data = pd.read_pickle(self._file_data_pickle)
            self._info = pd.read_pickle(self._file_info_pickle)
            # store experimental data for recreating state after reading from files
            self._data_unprocessed = self._data.copy()
        except FileNotFoundError:
            # raise custom Exception
            raise ColdStartException()

        # log
        logging.info("_data and _info loaded from pickle files.")

    #
    # determine csv data range
    #
    def _determine_data_range_csv(self, file):
        """
        determine csv data range of CSV-file.

        Parameters
        ----------
        file : str
            filepath.

        Returns
        -------
        empty_lines : TYPE
            DESCRIPTION.

        """
        # open csv file
        thefile = open(file)
        # detect empty lines which are characteristic at the beginning and
        # end of the data block
        empty_lines = [
            index for index, line in enumerate(csv.reader(thefile)) if len(line) == 0
        ]
        return empty_lines

    def _read_csv_data(self, file, show_info=True):
        """
        NEW IMPLEMENTATION
        """
        filetype = pathlib.Path(file).suffix
        if not self._processed:
            if filetype == ".csv":
                delimiter = utils.detect_delimiter(file)
                title_row = utils.find_title_row(file, delimiter)
            else:
                delimiter = None
                title_row = 0

            data = utils.load_data(file, delimiter, title_row)

            start_time = utils.find_reaction_start_time(data)

            if delimiter == "\t":
                data = utils.prepare_tab_columns(data, file)
            else:
                if filetype == ".csv":
                    data = utils.tidy_colnames(data)

            data = utils.remove_unnecessary_data(data)
            data = utils.convert_df_to_float(data)
            data = utils.correct_start_time(data, start_time)
            data = utils.add_sample_info(data, file)

        elif self._processed:
            data = pd.read_csv(file, sep=",", header=0)

        return data

    #
    # read csv data
    #
    def _read_calo_data_csv(self, file, show_info=True):
        """
        try reading calorimetry data from csv file via multiple options

        Parameters
        ----------
        file : str | pathlib.Path
            path to csv fileto be read.
        show_info : bool, optional
            flag whether or not to show information. The default is True.

        Returns
        -------
        pd.DataFrame

        """

        try:
            data = self._read_calo_data_csv_comma_sep(file, show_info=show_info)
        except Exception:
            data = self._read_calo_data_csv_tab_sep(file, show_info=show_info)

        # valid read
        if data is None:
            # log
            logging.info(f"\u2716 reading {file} FAILED.")

        # log
        logging.info(f"\u2714 reading {file} successful.")

        # return
        return data

    #
    # read csv data
    #
    def _read_calo_data_csv_comma_sep(self, file, show_info=True):
        """
        read data from csv file

        Parameters
        ----------
        file : str
            filepath.

        Returns
        -------
        data : pd.DataFrame
            experimental data contained in file.

        """

        # define Excel file
        data = pd.read_csv(
            file, header=None, sep="No meaningful separator", engine="python"
        )

        # check for tab-separation
        if "\t" in data.at[0, 0]:
            # raise Exception
            raise ValueError

        # look for potential index indicating in-situ-file
        if data[0].str.contains("Reaction start").any():
            # get target row
            helper = data[0].str.contains("Reaction start")
            # get row
            start_row = helper[helper].index.tolist()[0]
            # get offset for in-situ files
            t_offset_in_situ_s = float(data.at[start_row, 0].split(",")[0])

        data = utils.parse_rowwise_data(data)
        data = utils.tidy_colnames(data)

        data = utils.remove_unnecessary_data(data)

        # type conversion
        data = utils.convert_df_to_float(data)

        # check for "in-situ" sample --> reset
        try:
            # offset
            data["time_s"] -= t_offset_in_situ_s
            # write to log
            logging.info(
                f"\u26a0 Consider {file} as in-situ-file --> time-scale adjusted."
            )
        except Exception:
            pass

        # restrict to "time_s" > 0
        data = data.query("time_s > 0").reset_index(drop=True)

        # add sample information
        data = utils.add_sample_info(data, file)

        # if self.processparams.downsample.apply:
        #     data = self._apply_adaptive_downsampling(data)

        # return
        return data

    #
    # read csv data
    #
    def _read_calo_data_csv_tab_sep(self, file: str, show_info=True) -> pd.DataFrame:
        """
        Parameters
        ----------
        file : str | pathlib.Path
            path to tab separated csv-files from "older" versions of the device.
        show_info : bool, optional
            flag whether or not to show information. The default is True.

        Returns
        -------
        pd.DataFrame

        """

        # read
        raw = pd.read_csv(file, sep="\t", header=None)

        # process
        data = raw.copy()

        # get sample mass (if available)
        try:
            # get mass, first row in 3rd column is the title
            # the assumption is that the sample weight is one value on top of the 3rd column
            mass_index = raw.index[raw.iloc[:, 3].notna()]
            mass = float(raw.iloc[mass_index[1], 3])
        except IndexError:
            # set mass to None
            mass = None
            # go on
            pass

        # get "reaction start" time (if available)
        try:
            # get "reaction start" time in seconds
            _helper = data[data.iloc[:, 2].str.lower() == "reaction start"].head(1)
            # convert to float
            t0 = float(_helper[0].values[0])
        except Exception:
            # set t0 to None
            t0 = None
            # go on
            pass

        # remove all-Nan columns
        data = data.dropna(how="all", axis=1)

        # restrict to first two columns
        data = data.iloc[:, :2]

        # rename
        try:
            data.columns = ["time_s", "heat_flow_mw"]
        except ValueError:
            # return empty DataFrame
            return pd.DataFrame({"time_s": 0}, index=[0])

        # get data columns
        data = data.loc[3:, :].reset_index(drop=True)

        # convert data types
        data["time_s"] = data["time_s"].astype(float)
        data["heat_flow_mw"] = data["heat_flow_mw"].apply(
            lambda x: float(x.replace(",", "."))
        )

        # convert to same unit
        data["heat_flow_w"] = data["heat_flow_mw"] / 1000

        # calculate cumulative heat flow
        data["heat_j"] = integrate.cumulative_trapezoid(
            data["heat_flow_w"], x=data["time_s"], initial=0
        )

        # remove "heat_flow_w" column
        del data["heat_flow_mw"]

        # take into account time offset via "reactin start" time
        if t0:
            data["time_s"] -= t0

        # calculate normalized heat flow and heat
        if mass:
            data["normalized_heat_flow_w_g"] = data["heat_flow_w"] / mass
            data["normalized_heat_j_g"] = data["heat_j"] / mass

        # restrict to "time_s" > 0
        data = data.query("time_s >= 0").reset_index(drop=True)

        # add sample information
        data["sample"] = file
        data["sample_short"] = pathlib.Path(file).stem

        # type conversion
        data = utils.convert_df_to_float(data)

        # return
        return data

    #
    # read csv info
    #
    def _read_calo_info_csv(self, file, show_info=True):
        """
        read info from csv file

        Parameters
        ----------
        file : str
            filepath.

        Returns
        -------
        info : pd.DataFrame
            information (metadata) contained in file

        """

        try:
            # determine number of lines to skip
            empty_lines = self._determine_data_range_csv(file)
            # read info block from csv-file
            info = pd.read_csv(
                file, nrows=empty_lines[0] - 1, names=["parameter", "value"]
            ).dropna(subset=["parameter"])
            # the last block is not really meta data but summary data and
            # somewhat not necessary
        except IndexError:
            # return empty DataFrame
            info = pd.DataFrame()

        # add sample name as column
        info["sample"] = file
        info["sample_short"] = pathlib.Path(file).stem

        # return
        return info

    #
    # read excel info
    #
    def _read_calo_info_xls(self, file, show_info=True):
        """
        read information from xls-file

        Parameters
        ----------
        file : str
            filepath.
        show_info : bool, optional
            flag whether or not to show information. The default is True.

        Returns
        -------
        info : pd.DataFrame
            information (metadata) contained in file

        """
        # specify Excel
        xl = pd.ExcelFile(file)

        try:
            # get experiment info (first sheet)
            df_experiment_info = xl.parse(
                sheet_name="Experiment info", header=0, names=["parameter", "value"]
            ).dropna(subset=["parameter"])
            # use first row as header
            df_experiment_info = df_experiment_info.iloc[1:, :]

            # add sample information
            df_experiment_info["sample"] = file
            df_experiment_info["sample_short"] = pathlib.Path(file).stem

            # rename variable
            info = df_experiment_info

            # return
            return info

        except Exception as e:
            if show_info:
                print(e)
                print(f"==> ERROR in file {file}")

    #
    # read excel data
    #
    def _read_calo_data_xls(self, file, show_info=True):
        """
        read data from xls-file

        Parameters
        ----------
        file : str
            filepath.
        show_info : bool, optional
            flag whether or not to show information. The default is True.

        Returns
        -------
        data : pd.DataFrame
            data contained in file

        """

        # define Excel file
        xl = pd.ExcelFile(file)

        try:
            # parse "data" sheet
            df_data = xl.parse("Raw data", header=None)

            # replace init timestamp
            df_data.iloc[0, 0] = "time"

            # get new column names
            new_columnames = []
            for i, j in zip(df_data.iloc[0, :], df_data.iloc[1, :]):
                # build
                new_columnames.append(
                    re.sub(r"[\s\n\[\]\(\)° _]+", "_", f"{i}_{j}".lower())
                    .replace("/", "_")
                    .replace("_signal_", "_")
                )

            # set
            df_data.columns = new_columnames

            # cut out data part
            df_data = df_data.iloc[2:, :].reset_index(drop=True)

            # drop column
            try:
                df_data = df_data.drop(columns=["time_markers_nan"])
            except KeyError:
                pass

            # remove columns with too many NaNs
            df_data = df_data.dropna(axis=1, thresh=3)
            # # remove rows with NaNs
            # df_data = df_data.dropna(axis=0)

            # float conversion
            for _c in df_data.columns:
                # convert
                df_data[_c] = df_data[_c].astype(float)

            # add sample information
            df_data["sample"] = file
            df_data["sample_short"] = pathlib.Path(file).stem

            # rename
            data = df_data

            # log
            logging.info(f"\u2714 reading {file} successful.")

            # return
            return data

        except Exception as e:
            if show_info:
                print(
                    "\n\n==============================================================="
                )
                print(f"{e} in file '{pathlib.Path(file).name}'")
                print("Please, rename the data sheet to 'Raw data' (device default).")
                print(
                    "===============================================================\n\n"
                )

            # log
            logging.info(f"\u2716 reading {file} FAILED.")

            # return
            return None

    #
    # iterate samples
    #
    def _iter_samples(self, regex=None):
        """
        iterate samples and return corresponding data

        Returns
        -------
        sample (str) : name of the current sample
        data (pd.DataFrame) : data corresponding to the current sample
        """

        for sample, data in self._data.groupby(by="sample"):
            if regex:
                if not re.findall(regex, sample):
                    continue

            yield sample, data

    #
    # auto clean data
    #
    def _auto_clean_data(self):
        """
        remove NaN values from self._data and merge differently named columns
        representing the (constant) temperature set for the measurement

        Returns
        -------
        None.

        """

        # remove NaN values and reset index
        self._data = self._data.dropna(
            subset=[c for c in self._data.columns if re.match("normalized_heat", c)]
        ).reset_index(drop=True)

        # determine NaN count
        nan_count = self._data["temperature_temperature_c"].isna().astype(
            int
        ) + self._data["temperature_c"].isna().astype(int)

        # consolidate temperature columns
        if (
            "temperature_temperature_c" in self._data.columns
            and "temperature_c" in self._data.columns
        ):
            # use values from column "temperature_c" and set the values to column
            # "temperature_c"
            self._data.loc[
                (self._data["temperature_temperature_c"].isna()) & (nan_count == 1),
                "temperature_temperature_c",
            ] = self._data.loc[
                (~self._data["temperature_c"].isna()) & (nan_count == 1),
                "temperature_c",
            ]

            # remove values from column "temperature_c"
            self._data = self._data.drop(columns=["temperature_c"])

        # rename column
        self._data = self._data.rename(
            columns={"temperature_temperature_c": "temperature_c"}
        )

    #
    # plot
    #
    def plot(
        self,
        t_unit="h",
        y="normalized_heat_flow_w_g",
        y_unit_milli=True,
        regex=None,
        show_info=True,
        ax=None,
    ):
        """

        Plot the calorimetry data.

        Parameters
        ----------
        t_unit : str, optional
            time unit. The default is "h". Options are "s", "min", "h", "d".
        y : str, optional
            y-axis. The default is "normalized_heat_flow_w_g". Options are
            "normalized_heat_flow_w_g", "heat_flow_w", "normalized_heat_j_g",
            "heat_j".
        y_unit_milli : bool, optional
            whether or not to plot y-axis in Milliwatt. The default is True.
        regex : str, optional
            regex pattern to include only certain samples during plotting. The
            default is None.
        show_info : bool, optional
            whether or not to show information. The default is True.
        ax : matplotlib.axes._axes.Axes, optional
            axis to plot to. The default is None.

        Examples
        --------
        >>> import CaloCem as ta
        >>> from pathlib import Path
        >>>
        >>> calodatapath = Path(__file__).parent
        >>> tam = ta.Measurement(folder=calodatapath, show_info=True)
        >>> tam.plot(t_unit="h", y="normalized_heat_flow_w_g", y_unit_milli=False)

        """

        # y-value
        if y == "normalized_heat_flow_w_g":
            y_column = "normalized_heat_flow_w_g"
            y_label = "Normalized Heat Flow / [W/g]"
        elif y == "heat_flow_w":
            y_column = "heat_flow_w"
            y_label = "Heat Flow / [W]"
        elif y == "normalized_heat_j_g":
            y_column = "normalized_heat_j_g"
            y_label = "Normalized Heat / [J/g]"
        elif y == "heat_j":
            y_column = "heat_j"
            y_label = "Heat / [J]"

        if y_unit_milli:
            y_label = y_label.replace("[", "[m")

        # x-unit
        if t_unit == "s":
            x_factor = 1.0
        elif t_unit == "min":
            x_factor = 1 / 60
        elif t_unit == "h":
            x_factor = 1 / (60 * 60)
        elif t_unit == "d":
            x_factor = 1 / (60 * 60 * 24)

        # y-unit
        if y_unit_milli:
            y_factor = 1000
        else:
            y_factor = 1

        for sample, data in self._iter_samples():
            if regex:
                if not re.findall(rf"{regex}", os.path.basename(sample)):
                    continue
            data["time_s"] = data["time_s"] * x_factor
            # all columns containing heat
            heatcols = [s for s in data.columns if "heat" in s]
            data[heatcols] = data[heatcols] * y_factor
            ax, _ = utils.create_base_plot(data, ax, "time_s", y_column, sample)
            ax = utils.style_base_plot(
                ax,
                y_label,
                t_unit,
                sample,
            )
        return ax

    #
    # plot by category
    #
    def plot_by_category(
        self, categories, t_unit="h", y="normalized_heat_flow_w_g", y_unit_milli=True
    ):
        """
        plot by category, wherein the category is based on the information passed
        via "self._add_metadata_source". Options available as "category" are
        accessible via "self.get_metadata_grouping_options"

        Parameters
        ----------
        categories : str, list[str]
            category (from "self.get_metadata_grouping_options") to group by.
            specify a string or a list of strings here
        t_unit : TYPE, optional
            see "self.plot". The default is "h".
        y : TYPE, optional
            see "self.plot". The default is "normalized_heat_flow_w_g".
        y_unit_milli : TYPE, optional
            see "self.plot". The default is True.

        Examples
        --------
        >>> import CaloCem as ta
        >>> from pathlib import Path
        >>>
        >>> calodatapath = Path(__file__).parent
        >>> tam = ta.Measurement(folder=calodatapath, show_info=True)
        >>> tam.plot_by_category(categories="sample")


        Returns
        -------
        None.

        """

        def build_helper_string(values: list) -> str:
            """
            build a "nicely" formatted string from a supplied list
            """

            if len(values) == 2:
                # connect with "and"
                formatted = " and ".join([str(i) for i in values])
            elif len(values) > 2:
                # connect with comma and "and" for last element
                formatted = (
                    ", ".join([str(i) for i in values[:-1]]) + " and " + str(values[-1])
                )
            else:
                formatted = "---"

            # return
            return formatted

        # loop category values
        for selections, _ in self._metadata.groupby(by=categories):
            if isinstance(selections, tuple):
                # - if multiple categories to group by are specified -
                # init helper DataFrame
                target_idx = pd.DataFrame()
                # identify corresponding samples
                for selection, category in zip(selections, categories):
                    target_idx[category] = self._metadata[category] == selection
                # get relevant indices
                target_idx = target_idx.sum(axis=1) == len(categories)
                # define title
                title = f"Grouped by {build_helper_string(categories)} ({build_helper_string(selections)})"
            else:
                # - if only one(!) category to group by is specified -
                # identify corresponding samples
                target_idx = self._metadata[categories] == selections
                # define title
                title = f"Grouped by {categories} ({selections})"

            # pick relevant samples
            target_samples = self._metadata.loc[target_idx, self._metadata_id]

            # build corresponding regex
            regex = "(" + ")|(".join(target_samples) + ")"

            # plot
            ax = self.plot(regex=regex, t_unit=t_unit, y=y, y_unit_milli=y_unit_milli)

            # set title
            ax.set_title(title)

            # yield latest plot
            yield selections, ax

    @staticmethod
    def _plot_peak_positions(
        data,
        ax,
        _age_col,
        _target_col,
        peaks,
        sample,
        plt_top,
        plt_right_s,
        plot_labels,
        xmarker,
    ):
        """
        Plot detected peaks.
        """

        ax, new_ax = utils.create_base_plot(data, ax, _age_col, _target_col, sample)

        if xmarker:
            ax.plot(
                data[_age_col][peaks],
                data[_target_col][peaks],
                "x",
                color="red",
            )

        ax.vlines(
            x=data[_age_col][peaks],
            ymin=0,
            ymax=data[_target_col][peaks],
            color="red",
        )

        if plot_labels:
            for x, y in zip(data[_age_col][peaks], data[_target_col][peaks]):
                y = y + 0.0002
                ax.text(x, y, f"{round(x, 2)}", color="red")

            # ax.text(
            #     x=data[_age_col][peaks],
            #     y=data[_target_col][peaks],
            #     #s=[f"{round(i,2)}" for i in data[_age_col][peaks]],
            #     s="hallo",
            #     color="red",
            # )

        limits = {
            "left": ax.get_xlim()[0],
            "right": plt_right_s,
            "bottom": 0,
            "top": plt_top,
        }

        ax = utils.style_base_plot(ax, _target_col, _age_col, sample, limits)

        if new_ax:
            plt.show()

    @staticmethod
    def _plot_maximum_slope(
        data,
        ax,
        age_col,
        target_col,
        sample,
        characteristics,
        time_discarded_s,
        save_path=None,
        xscale="log",
        xunit="s",
    ):
        x_increment = 600
        if xunit == "h":
            data[age_col] = data[age_col] / 3600
            characteristics[age_col] = characteristics[age_col] / 3600
            time_discarded_s = time_discarded_s / 3600
            x_increment = 0.2

        ax, new_ax = utils.create_base_plot(data, ax, age_col, target_col, sample)

        ax2 = ax.twinx()
        # plot gradient
        ax2.plot(data[age_col], data["gradient"], label="Gradient", color="orange")
        ax2.set_yscale("linear")
        xmask = data[age_col] > time_discarded_s
        y_vals = data[target_col][xmask]
        ymin = y_vals.min() + y_vals.min() * 0.1
        ymax = y_vals.max() + y_vals.max() * 0.1
        # ax.set_ylim(ymin, ymax)

        y2_vals = data["gradient"][xmask]
        y2min = y2_vals.min() + y2_vals.min() * 0.1
        y2max = y2_vals.max() + y2_vals.max() * 0.1
        ax2.set_ylim(y2min, y2max)

        ax2.set_ylabel(r"Gradient [Wg$^{-1}$s$^{-1}$]")

        # add vertical lines
        for _idx, _row in characteristics.iterrows():
            # vline
            t_maxslope = _row.at[age_col]
            ax.axvline(t_maxslope, color="green", alpha=0.3)

        if xunit == "h":
            limits = {"left": 0.1, "right": ax.get_xlim()[1], "bottom": 0, "top": ymax}

        else:
            limits = {"left": 100, "right": ax.get_xlim()[1], "bottom": 0, "top": ymax}

        ax = utils.style_base_plot(
            ax,
            target_col,
            age_col,
            sample,
            limits,
            time_discarded_s=time_discarded_s,
            xunit=xunit,
        )

        ax.set_xscale(xscale)
        ax.text(
            t_maxslope + x_increment,
            0.00025,
            f"{round(t_maxslope, 2)} {xunit} ",
            color="green",
        )

        if new_ax:
            if save_path:
                sample_name = pathlib.Path(sample).stem
                plt.savefig(save_path / f"maximum_slope_detect_{sample_name}.pdf")
            else:
                plt.show()

    @staticmethod
    def _plot_intersection(
        data,
        ax,
        age_col,
        target_col,
        sample,
        # characteristics,
        time_discarded_s,
        characteristics,
        save_path=None,
        xscale="log",
        # xunit="s",
        hmax=None,
        tmax=None,
    ):
        if characteristics.xunit == "h":
            data.loc[:, age_col] = data.loc[:, age_col] / 3600
            characteristics.time_s = characteristics.time_s / 3600
            characteristics.dorm_time_s = characteristics.dorm_time_s / 3600
            characteristics.gradient = characteristics.gradient * 3600
            tmax = tmax / 3600
            characteristics.x_intersect = characteristics.x_intersect / 3600
            # characteristics[age_col] = characteristics[age_col] / 3600

        ax, new_ax = utils.create_base_plot(data, ax, age_col, target_col, sample)
        # print(new_ax)
        ax = utils.style_base_plot(
            ax,
            target_col,
            age_col,
            sample,
            time_discarded_s=time_discarded_s,
            xunit=characteristics.xunit,
        )

        ax.axline(
            (characteristics.time_s, characteristics.normalized_heat_flow_w_g),
            slope=characteristics.gradient,
            color="red",
            linestyle="--",
        )
        if characteristics.intersection == "dormant_hf":
            ax.axhline(
                y=characteristics.dorm_normalized_heat_flow_w_g,
                color="red",
                linestyle="--",
            )
            ax.text(
                x=characteristics.x_intersect,
                y=characteristics.dorm_normalized_heat_flow_w_g,
                s=rf"   $t_i=$ {characteristics.x_intersect:.1f} {characteristics.xunit}"
                + "\n",
                color="green",
            )
        elif characteristics.intersection == "abscissa":
            ax.text(
                x=characteristics.x_intersect,
                y=0,
                s=rf"   $t_i=$ {characteristics.x_intersect:.1f} {characteristics.xunit}"
                + "\n",
                color="green",
            )

        ax.axvline(
            x=characteristics.x_intersect,
            color="green",
            linestyle=":",
        )

        ax.set_xscale(xscale)
        ax.set_xlim(0, tmax)
        ax.set_ylim(0, hmax)

        if new_ax:
            if save_path:
                sample_name = pathlib.Path(sample).stem
                plt.savefig(save_path / f"intersection_detect_{sample_name}.pdf")
            else:
                plt.show()

    #
    # get the cumulated heat flow a at a certain age
    #

    def get_cumulated_heat_at_hours(self, processparams=None, target_h=4, **kwargs):
        """
        get the cumulated heat flow a at a certain age

        Parameters
        ----------
        processparams : ProcessingParameters, optional
            Processing parameters. The default is None. If None, the default
            parameters are used. The most important parameter is the cutoff time
            in minutes which describes the initial time period of the measurement
            which is not considered for the cumulated heat flow. It is defined in
            the ProcessingParameters class. The default value is 30 minutes.

        target_h : int | float
            end time in hourscv

        Returns
        -------
        A Pandas dataframe

        """
        if "cutoff_min" in kwargs:
            cutoff_min = kwargs["cutoff_min"]
            warnings.warn(
                "The cutoff_min parameter is deprecated. Please use the ProcessingParameters class instead.",
                DeprecationWarning,
                stacklevel=2,
            )
        else:
            if not processparams:
                processparams = ProcessingParameters()
            cutoff_min = processparams.cutoff.cutoff_min

        def applicable(df, target_h=4, cutoff_min=None):
            # convert target time to seconds
            target_s = 3600 * target_h
            # helper
            _helper = df.query("time_s <= @target_s").tail(1)
            # get heat at target time
            hf_at_target = float(_helper["normalized_heat_j_g"].values[0])

            # if cutoff time specified
            if cutoff_min:
                # convert target time to seconds
                target_s = 60 * cutoff_min
                try:
                    # helper
                    _helper = df.query("time_s <= @target_s").tail(1)
                    # type conversion
                    hf_at_cutoff = float(_helper["normalized_heat_j_g"].values[0])
                    # correct heatflow for heatflow at cutoff
                    hf_at_target = hf_at_target - hf_at_cutoff
                except TypeError:
                    name_wt_nan = df["sample_short"].tolist()[0]
                    print(
                        f"Found NaN in Normalized heat of sample {name_wt_nan} searching for cumulated heat at {target_h}h and a cutoff of {cutoff_min}min."
                    )
                    return np.nan

            # return
            return hf_at_target

        # in case of one specified time
        if isinstance(target_h, int) or isinstance(target_h, float):
            # groupby
            results = (
                self._data.groupby(by="sample")[["time_s", "normalized_heat_j_g"]]
                .apply(
                    lambda x: applicable(x, target_h=target_h, cutoff_min=cutoff_min),
                )
                .reset_index(level=0)
            )
            # rename
            results.columns = ["sample", "cumulated_heat_at_hours"]
            results["target_h"] = target_h
            results["cutoff_min"] = cutoff_min

        # in case of specified list of times
        elif isinstance(target_h, list):
            # init list
            list_of_results = []
            # loop
            for this_target_h in target_h:
                # groupby
                _results = (
                    self._data.groupby(by="sample")[["time_s", "normalized_heat_j_g"]]
                    .apply(
                        lambda x: applicable(
                            x, target_h=this_target_h, cutoff_min=cutoff_min
                        ),
                    )
                    .reset_index(level=0)
                )
                # rename
                _results.columns = ["sample", "cumulated_heat_at_hours"]
                _results["target_h"] = this_target_h
                _results["cutoff_min"] = cutoff_min
                # append to list
                list_of_results.append(_results)
            # build overall results DataFrame
            results = pd.concat(list_of_results)

        # return
        return results

    #
    # find peaks
    #
    def get_peaks(
        self,
        processparams,
        target_col="normalized_heat_flow_w_g",
        regex=None,
        cutoff_min=None,
        show_plot=True,
        plt_right_s=2e5,
        plt_top=1e-2,
        ax=None,
        xunit="s",
        plot_labels=None,
        xmarker=False,
    ) -> pd.DataFrame:
        """
        get DataFrame of peak characteristics.

        Parameters
        ----------
        target_col : str, optional
            measured quantity within which peaks are searched for. The default is "normalized_heat_flow_w_g"
        regex : str, optional
            regex pattern to include only certain experimental result files
            during initialization. The default is None.
        cutoff_min : int | float, optional
            Time in minutes below which collected data points are discarded for peak picking
        show_plot : bool, optional
            Flag whether or not to plot peak picking for each sample. The default is True.
        plt_right_s : int | float, optional
            Upper limit of x-axis of in seconds. The default is 2e5.
        plt_top : int | float, optional
            Upper limit of y-axis of. The default is 1e-2.
        ax : matplotlib.axes._axes.Axes | None, optional
            The default is None.

        Returns
        -------
        pd.DataFrame holding peak characterisitcs for each sample.

        """

        # list of peaks
        list_of_peaks_dfs = []

        # loop samples
        for sample, data in self._iter_samples(regex=regex):
            # cutoff
            if processparams.cutoff.cutoff_min:
                # discard points at early age
                data = data.query("time_s >= @processparams.cutoff.cutoff_min * 60")

            # reset index
            data = data.reset_index(drop=True)

            # target_columns
            _age_col = "time_s"
            _target_col = target_col

            # find peaks
            peaks, properties = signal.find_peaks(
                data[_target_col],
                prominence=processparams.peakdetection.prominence,
                distance=processparams.peakdetection.distance,
            )

            # plot?
            if show_plot:
                if xunit == "h":
                    df_copy = data.copy()
                    df_copy[_age_col] = df_copy[_age_col] / 3600
                    plt_right_s = plt_right_s / 3600
                    self._plot_peak_positions(
                        df_copy,
                        ax,
                        _age_col,
                        _target_col,
                        peaks,
                        sample,
                        plt_top,
                        plt_right_s,
                        plot_labels,
                        xmarker,
                    )
                else:
                    self._plot_peak_positions(
                        data,
                        ax,
                        _age_col,
                        _target_col,
                        peaks,
                        sample,
                        plt_top,
                        plt_right_s,
                        plot_labels,
                        xmarker,
                    )

            # compile peak characteristics
            peak_characteristics = pd.concat(
                [
                    data.iloc[peaks, :],
                    pd.DataFrame(
                        properties["prominences"], index=peaks, columns=["prominence"]
                    ),
                    pd.DataFrame({"peak_nr": np.arange((len(peaks)))}, index=peaks),
                ],
                axis=1,
            )

            # append
            list_of_peaks_dfs.append(peak_characteristics)

        # compile peak information
        peaks = pd.concat(list_of_peaks_dfs)

        if isinstance(ax, matplotlib.axes._axes.Axes):
            # return peak list and ax
            return peaks, ax
        else:  # return peak list only
            return peaks

    #
    # get peak onsets
    #
    def get_peak_onsets(
        self,
        target_col="normalized_heat_flow_w_g",
        age_col="time_s",
        time_discarded_s=900,
        rolling=1,
        gradient_threshold=0.0005,
        show_plot=False,
        exclude_discarded_time=False,
        regex=None,
        ax: plt.Axes = None,
    ):
        """
        get peak onsets based on a criterion of minimum gradient

        Parameters
        ----------
        target_col : str, optional
            measured quantity within which peak onsets are searched for. The default is "normalized_heat_flow_w_g"
        age_col : str, optional
            Time unit within which peak onsets are searched for. The default is "time_s"
        time_discarded_s : int | float, optional
            Time in seconds below which collected data points are discarded for peak onset picking. The default is 900.
        rolling : int, optional
            Width of "rolling" window within which the values of "target_col" are averaged. A higher value will introduce a stronger smoothing effect. The default is 1, i.e. no smoothing.
        gradient_threshold : float, optional
            Threshold of slope for identification of a peak onset. For a lower value, earlier peak onsets will be identified. The default is 0.0005.
        show_plot : bool, optional
            Flag whether or not to plot peak picking for each sample. The default is False.
        exclude_discarded_time : bool, optional
            Whether or not to discard the experimental values obtained before "time_discarded_s" also in the visualization. The default is False.
        regex : str, optional
            regex pattern to include only certain experimental result files during initialization. The default is None.
        ax : matplotlib.axes._axes.Axes | None, optional
            The default is None.
        Returns
        -------
        pd.DataFrame holding peak onset characterisitcs for each sample.

        """

        # init list of characteristics
        list_of_characteristics = []

        # loop samples
        for sample, data in self._iter_samples(regex=regex):
            if exclude_discarded_time:
                # exclude
                data = data.query(f"{age_col} >= {time_discarded_s}")

            # reset index
            data = data.reset_index(drop=True)

            # calculate get gradient
            data["gradient"] = pd.Series(
                np.gradient(data[target_col].rolling(rolling).mean(), data[age_col])
            )

            # get relevant points
            characteristics = data.copy()
            # discard initial time
            characteristics = characteristics.query(f"{age_col} >= {time_discarded_s}")
            # look at values with certain gradient only
            characteristics = characteristics.query("gradient > @gradient_threshold")
            # consider first entry exclusively
            characteristics = characteristics.head(1)

            # optional plotting
            if show_plot:
                # if specific axis to plot to is specified
                if isinstance(ax, matplotlib.axes._axes.Axes):
                    # plot heat flow curve
                    p = ax.plot(data[age_col], data[target_col])

                    # add vertical lines
                    for _idx, _row in characteristics.iterrows():
                        # vline
                        ax.axvline(_row.at[age_col], color=p[0].get_color(), alpha=0.3)
                        # add "slope line"
                        ax.axline(
                            (_row.at[age_col], _row.at[target_col]),
                            slope=_row.at["gradient"],
                            color=p[0].get_color(),
                            # color="k",
                            # linewidth=0.2
                            alpha=0.25,
                            linestyle="--",
                        )

                    # cosmetics
                    # ax.set_xscale("log")
                    ax.set_title("Onset for " + pathlib.Path(sample).stem)
                    ax.set_xlabel(age_col)
                    ax.set_ylabel(target_col)

                    ax.fill_between(
                        [ax.get_ylim()[0], time_discarded_s],
                        [ax.get_ylim()[0]] * 2,
                        [ax.get_ylim()[1]] * 2,
                        color="black",
                        alpha=0.35,
                    )

                    # set axis limit
                    ax.set_xlim(left=100)

                else:
                    # plot heat flow curve
                    plt.plot(data[age_col], data[target_col])

                    # add vertical lines
                    for _idx, _row in characteristics.iterrows():
                        # vline
                        plt.axvline(_row.at[age_col], color="red", alpha=0.3)

                    # cosmetics
                    # plt.xscale("log")
                    plt.title("Onset for " + pathlib.Path(sample).stem)
                    plt.xlabel(age_col)
                    plt.ylabel(target_col)

                    # get axis
                    ax = plt.gca()

                    plt.fill_between(
                        [ax.get_ylim()[0], time_discarded_s],
                        [ax.get_ylim()[0]] * 2,
                        [ax.get_ylim()[1]] * 2,
                        color="black",
                        alpha=0.35,
                    )

                    # set axis limit
                    plt.xlim(left=100)

            # append to list
            list_of_characteristics.append(characteristics)

        # build overall list
        onset_characteristics = pd.concat(list_of_characteristics)

        # return
        if isinstance(ax, matplotlib.axes._axes.Axes):
            # return onset characteristics and ax
            return onset_characteristics, ax
        else:
            # return onset characteristics exclusively
            return onset_characteristics

    #
    # get maximum slope
    #

    def get_maximum_slope(
        self,
        processparams,
        target_col="normalized_heat_flow_w_g",
        age_col="time_s",
        time_discarded_s=900,
        show_plot=False,
        show_info=True,
        exclude_discarded_time=False,
        regex=None,
        read_start_c3s=False,
        ax=None,
        save_path=None,
        xscale="log",
        xunit="s",
    ):
        """
        The method finds the point in time of the maximum slope. It also calculates the gradient at this point. The method can be controlled by passing a customized ProcessingParameters object for the `processparams` parameter. If no object is passed, the default parameters will be used.

        Parameters
        ----------
        target_col : str, optional
            measured quantity within which peak onsets are searched for. The default is "normalized_heat_flow_w_g"
        age_col : str, optional
            Time unit within which peak onsets are searched for. The default is "time_s"
        time_discarded_s : int | float, optional
            Time in seconds below which collected data points are discarded for peak onset picking. The default is 900.
        show_plot : bool, optional
            Flag whether or not to plot peak picking for each sample. The default is False.
        exclude_discarded_time : bool, optional
            Whether or not to discard the experimental values obtained before "time_discarded_s" also in the visualization. The default is False.
        regex : str, optional
            regex pattern to include only certain experimental result files during initialization. The default is None.
        Returns
        -------
        Pandas Dataframe
            A dataframe that contains the time and the gradient of the maximum slope.
        Examples
        --------
        >>> from CaloCem import tacalorimetry as ta
        >>> from pathlib import Path

        >>> thepath = Path(__file__).parent / "data"
        >>> tam = ta.Measurement(thepath)
        >>> processparams = ta.ProcessingParameters()
        >>> processparams..apply = True
        >>> max_slopes = tam.get_maximum_slope(processparams)
        """

        # init list of characteristics
        list_of_characteristics = []

        # loop samples
        for sample, data in self._iter_samples(regex=regex):
            sample_name = pathlib.Path(sample).stem
            if exclude_discarded_time:
                # exclude
                data = data.query(f"{age_col} >= {time_discarded_s}")

            # manual definition of start time to look for c3s - in case auto peak detection becomes difficult
            if read_start_c3s:
                c3s_start_time_s = self._metadata.query(
                    f"sample_number == '{sample_name}'"
                )["t_c3s_min_s"].values[0]
                c3s_end_time_s = self._metadata.query(
                    f"sample_number == '{sample_name}'"
                )["t_c3s_max_s"].values[0]
                data = data.query(
                    f"{age_col} >= {c3s_start_time_s} & {age_col} <= {c3s_end_time_s}"
                )

            if show_info:
                print(f"Determineing maximum slope of {pathlib.Path(sample).stem}")

            processor = HeatFlowProcessor(processparams)

            data = utils.make_equidistant(data)

            if processparams.rolling_mean.apply:
                data = processor.apply_rolling_mean(data)

            data["gradient"], data["curvature"] = (
                processor.calculate_heatflow_derivatives(data)
            )

            characteristics = processor.get_largest_slope(data, processparams)
            if characteristics.empty:
                continue

            # optional plotting
            if show_plot:
                self._plot_maximum_slope(
                    data,
                    ax,
                    age_col,
                    target_col,
                    sample,
                    characteristics,
                    time_discarded_s,
                    save_path=save_path,
                    xscale=xscale,
                    xunit=xunit,
                )
                # plot heat flow curve
                # plt.plot(data[age_col], data[target_col], label=target_col)
                # plt.plot(
                #     data[age_col],
                #     data["gradient"] * 1e4 + 0.001,
                #     label="gradient * 1e4 + 1mW",
                # )

                # # add vertical lines
                # for _idx, _row in characteristics.iterrows():
                #     # vline
                #     plt.axvline(_row.at[age_col], color="green", alpha=0.3)

                # # cosmetics
                # plt.xscale("log")
                # plt.title(f"Maximum slope plot for {pathlib.Path(sample).stem}")
                # plt.xlabel(age_col)
                # plt.ylabel(target_col)
                # plt.legend()

                # # get axis
                # ax = plt.gca()

                # plt.fill_between(
                #     [ax.get_ylim()[0], time_discarded_s],
                #     [ax.get_ylim()[0]] * 2,
                #     [ax.get_ylim()[1]] * 2,
                #     color="black",
                #     alpha=0.35,
                # )

                # # set axis limit
                # plt.xlim(left=100)
                # plt.ylim(bottom=0, top=0.01)

                # # show
                # plt.show()

            # append to list
            list_of_characteristics.append(characteristics)

        if not list_of_characteristics:
            print("No maximum slope found, check you processing parameters")
        # build overall list
        else:
            max_slope_characteristics = pd.concat(list_of_characteristics)
            # return
            return max_slope_characteristics


    def get_average_slope(
    self,
    processparams,
    target_col="normalized_heat_flow_w_g",
    age_col="time_s",
    regex=None,
    show_plot=False,
    ax=None,
    save_path=None,
    xscale="linear",
    xunit="s",
    ):
        """
        Calculate average slope by determining 4 additional slope values between 
        onset time and heat flow maximum, in addition to the maximum slope.

        Parameters
        ----------
        processparams : ProcessingParameters
            Processing parameters for analysis
        target_col : str, optional
            Target measurement column, by default "normalized_heat_flow_w_g"
        age_col : str, optional
            Time column name, by default "time_s"
        regex : str, optional
            Regex pattern to filter samples, by default None
        show_plot : bool, optional
            Whether to show plots, by default False
        ax : matplotlib.axes.Axes, optional
            Existing axis to plot on, by default None
        save_path : Path, optional
            Path to save plots, by default None
        xscale : str, optional
            X-axis scale, by default "log"
        xunit : str, optional
            Time unit for display, by default "s"

        Returns
        -------
        pd.DataFrame
            DataFrame containing average slope characteristics for each sample
        """

        # Get maximum slopes using existing method
        max_slopes = self.get_maximum_slope(
            processparams,
            target_col=target_col,
            age_col=age_col,
            regex=regex,
            show_plot=False,
            ax=ax,
            save_path=save_path,
            xscale=xscale,
            xunit=xunit,
        )

        if max_slopes is None or max_slopes.empty:
            print("No maximum slopes found. Cannot calculate average slopes.")
            return pd.DataFrame()

        # Get onset times using the peak onset method
        onsets = self.get_peak_onset_via_max_slope(
            processparams,
            show_plot=False,
            regex=regex,
            age_col=age_col,
            target_col=target_col,
            xunit=xunit,
        )

        if onsets.empty:
            print("No onset times found. Cannot calculate average slopes.")
            return pd.DataFrame()

        list_of_characteristics = []

        # Loop through samples
        for sample, data in self._iter_samples(regex=regex):
            sample_short = pathlib.Path(sample).stem

            # Get max slope data for this sample
            max_slope_row = max_slopes[max_slopes["sample_short"] == sample_short]
            if max_slope_row.empty:
                continue

            # Get onset data for this sample
            onset_row = onsets[onsets["sample"] == sample_short]
            if onset_row.empty:
                continue

            # Get time points
            onset_time = onset_row["onset_time_s"].iloc[0]
            max_slope_time = max_slope_row[age_col].iloc[0]

            # Find heat flow maximum after onset
            data_after_onset = data[data[age_col] >= onset_time]
            if data_after_onset.empty:
                continue

            max_hf_time = data_after_onset.loc[data_after_onset[target_col].idxmax(), age_col]

            # Create 4 intermediate time points between onset and heat flow maximum
            if max_hf_time <= onset_time:
                print(f"Warning: Heat flow maximum occurs before onset for {sample_short}")
                continue

            # Create 6 time points total (onset, 4 intermediate, max_hf)
            time_points = np.linspace(onset_time + 3600, max_hf_time - 3600, 6)

            # Calculate slopes at each interval
            slopes = []
            slope_times = []

            for i in range(len(time_points) - 1):
                t1, t2 = time_points[i], time_points[i + 1]

                # Get data points in this interval
                interval_data = data[(data[age_col] >= t1) & (data[age_col] <= t2)]

                if len(interval_data) < 2:
                    continue

                # Calculate slope using linear regression
                x_vals = interval_data[age_col].values
                y_vals = interval_data[target_col].values

                # Simple slope calculation: (y2 - y1) / (x2 - x1)
                slope = (y_vals[-1] - y_vals[0]) / (x_vals[-1] - x_vals[0])
                slopes.append(slope)
                slope_times.append((t1 + t2) / 2)  # Midpoint time

            # Include the maximum slope
            max_slope_value = max_slope_row["gradient"].iloc[0]
            slopes.append(max_slope_value)
            slope_times.append(max_slope_time)

            # Calculate average slope
            if slopes:
                avg_slope = np.mean(slopes)
                std_slope = np.std(slopes)

                # Create characteristics dictionary
                characteristics = {
                    "sample": sample,
                    "sample_short": sample_short,
                    "onset_time_s": onset_time,
                    "max_hf_time_s": max_hf_time,
                    "max_slope_time_s": max_slope_time,
                    "max_slope_value": max_slope_value,
                    "average_slope": avg_slope,
                    "slope_std": std_slope,
                    "n_slopes": len(slopes),
                    "individual_slopes": slopes,
                    "slope_times": slope_times,
                }

                # Optional plotting
                if show_plot:
                    self._plot_average_slope_analysis(
                        data, characteristics, ax, age_col, target_col, 
                        sample_short, save_path, xscale, xunit
                    )

                list_of_characteristics.append(characteristics)

        if not list_of_characteristics:
            print("No average slope characteristics calculated.")
            return pd.DataFrame()

        # Convert to DataFrame
        avg_slope_df = pd.DataFrame(list_of_characteristics)

        return avg_slope_df


    @staticmethod
    def _plot_average_slope_analysis(
        data, characteristics, ax, age_col, target_col, 
        sample_short, save_path=None, xscale="linear", xunit="s"
    ):
        """Plot average slope analysis for visualization"""

        ax, new_ax = utils.create_base_plot(data, ax, age_col, target_col, sample_short, color="gray")

        # Plot the heat flow curve
        #ax.plot(data[age_col], data[target_col], 'b-', alpha=0.7, label='Heat Flow')

        # Mark onset time
        ax.axvline(characteristics["onset_time_s"], color='green', 
                linestyle='--', alpha=0.7, label='Onset')

        # Mark max heat flow time
        ax.axvline(characteristics["max_hf_time_s"], color='orange', 
                linestyle='--', alpha=0.7, label='Max Heat Flow')

        # Mark max slope time
        ax.axvline(characteristics["max_slope_time_s"], color='red', 
                linestyle='--', alpha=0.7, label='Max Slope')

        # Plot individual slope lines
        colors = plt.cm.viridis(np.linspace(0, 1, len(characteristics["individual_slopes"])))

        for i, (slope, time, color) in enumerate(zip(
            characteristics["individual_slopes"], 
            characteristics["slope_times"], 
            colors
        )):
            # Find corresponding y-value
            y_val = np.interp(time, data[age_col], data[target_col])

            # Plot slope line (extend ±10% of time range)
            time_range = characteristics["max_hf_time_s"] - characteristics["onset_time_s"]
            dt = 0.1 * time_range

            x_line = [time - dt, time + dt]
            y_line = [y_val - slope * dt, y_val + slope * dt]

            ax.plot(x_line, y_line, color=color, alpha=0.6, linewidth=2,
                    label=f'Slope {i+1}: {slope:.2e}')

        # Add text annotation for average slope
        ax.text(0.05, 0.95, 
                f'Avg Slope: {characteristics["average_slope"]:.2e}\n'
                f'Std: {characteristics["slope_std"]:.2e}',
                transform=ax.transAxes, verticalalignment='top',
                bbox=dict(boxstyle='round', facecolor='wheat', alpha=0.8))

        ax.set_xscale(xscale)
        ax.set_xlabel(f'Time [{xunit}]')
        ax.set_ylabel(target_col.replace('_', ' ').title())
        ax.set_title(f'Average Slope Analysis - {sample_short}')
        ax.legend(bbox_to_anchor=(1.05, 1), loc='upper left')
        ax.grid(True, alpha=0.3)
        #ax.set_ylim(0,0.003)

        if new_ax:
            if save_path:
                plt.tight_layout()
                plt.show()
                #plt.savefig(save_path / f"average_slope_analysis_{sample_short}.pdf")
                plt.close()
            else:
                plt.tight_layout()
                plt.show() 


    # get reaction onset via maximum slope
    #
    def get_peak_onset_via_max_slope(
        self,
        processparams,
        show_plot=False,
        ax=None,
        regex=None,
        age_col="time_s",
        target_col="normalized_heat_flow_w_g",
        time_discarded_s=900,
        save_path=None,
        xscale="linear",
        xunit="s",
        intersection="dormant_hf",
    ):
        """
        get reaction onset based on tangent of maximum heat flow and heat flow
        during the dormant period. The characteristic time is inferred from
        the intersection of both characteristic lines

        Parameters
        ----------
        show_plot : TYPE, optional
            DESCRIPTION. The default is False.

        Returns
        -------
        None.

        """
        # get onsets
        max_slopes = self.get_maximum_slope(
            processparams,
            regex=regex,
            show_plot=False,
            ax=ax,
        )
        # % get dormant period HFs
        dorm_hfs = self.get_dormant_period_heatflow(
            processparams,  # cutoff_min=cutoff_min, prominence=prominence
            regex=regex,
            show_plot=False,
            # ax=ax,
        )

        # init list
        list_characteristics = []

        # loop samples
        for i, row in max_slopes.iterrows():
            # calculate y-offset
            t = row["normalized_heat_flow_w_g"] - row["time_s"] * row["gradient"]
            # calculate point of intersection
            # calculate x-intersect of tangent with dormant heat flow
            x_intersect_dormant = (
                    float(
                        dorm_hfs[dorm_hfs["sample_short"] == row["sample_short"]][
                            "normalized_heat_flow_w_g"
                        ]
                    )
                    - t
                ) / row["gradient"]
            # elif intersection == "abscissa":
                # calculate x-intersect of tangent with abscissa (y=0)
            x_intersect = row["time_s"] - (row["normalized_heat_flow_w_g"] / row["gradient"])

            data = self._data.query("sample_short == @row['sample_short']")
            sample = row["sample_short"]

            heat_at_intersect = np.interp(x_intersect, data["time_s"], data["normalized_heat_j_g"])

            # append to list
            list_characteristics.append(
                {
                    "sample": row["sample_short"],
                    "onset_time_s_abscissa": x_intersect,
                    "onset_time_min_abscissa": x_intersect / 60,
                    "heat_at_onset_j_g": heat_at_intersect,
                    "onset_time_s": x_intersect_dormant,
                    "onset_time_min": x_intersect_dormant / 60,
                }
            )


            dorm_hfs_sample = dorm_hfs.query("sample_short == @sample")
            # add prefix dorm to all columns
            dorm_hfs_sample.columns = ["dorm_" + s for s in dorm_hfs_sample.columns]

            characteristics = pd.concat([row, dorm_hfs_sample.squeeze()])
            characteristics.loc["xunit"] = xunit
            characteristics.loc["x_intersect"] = x_intersect
            characteristics.loc["intersection"] = intersection
            # print(characteristics.x_intersect)

            # get maximum time value
            tmax = self._data.query("sample_short == @row['sample_short']")[
                "time_s"
            ].max()
            # get maximum heat flow value
            hmax = self._data.query(
                "time_s > 3000 & sample_short == @row['sample_short']"
            )["normalized_heat_flow_w_g"].max()

            if show_plot:
                self._plot_intersection(
                    data,
                    ax,
                    age_col,
                    target_col,
                    sample,
                    # max_slopes,
                    time_discarded_s,
                    characteristics=characteristics,
                    save_path=save_path,
                    xscale=xscale,
                    # xunit=xunit,
                    hmax=hmax,
                    tmax=tmax,
                )

        # build overall dataframe to be returned
        onsets = pd.DataFrame(list_characteristics)

        # merge with dorm_hfs
        onsets = onsets.merge(
            dorm_hfs[
                ["sample_short", "normalized_heat_flow_w_g", "normalized_heat_j_g"]
            ],
            left_on="sample",
            right_on="sample_short",
            how="left",
        )

        # rename
        onsets = onsets.rename(
            columns={
                "normalized_heat_flow_w_g": "normalized_heat_flow_w_g_at_dorm_min",
                "normalized_heat_j_g": "normalized_heat_j_g_at_dorm_min",
            }
        )

        # return
        return onsets
        # if isinstance(ax, matplotlib.axes._axes.Axes):
        #     # return onset characteristics and ax
        #     return onsets, ax
        # else:
        #     # return onset characteristics exclusively
        #     return onsets

    #
    # get dormant period heatflow
    #

    def get_dormant_period_heatflow(
        self,
        processparams,
        regex: str = None,
        cutoff_min: int = 5,
        upper_dormant_thresh_w_g: float = 0.002,
        plot_right_boundary=2e5,
        prominence: float = 1e-3,
        show_plot=False,
    ) -> pd.DataFrame:
        """
        get dormant period heatflow

        Parameters
        ----------
        regex : str, optional
            Regex which can be used to filter the data, i.e., only the patterns which fit the regex will be evaluated. The default is None.
        cutoff_min : int | float, optional
            Time at the start of the experiment which will be cutoff from analysis. This can be useful for ex-situ mixed samples. The default is 5.
        upper_dormant_thresh_w_g : float, optional
            Parameter which controls the upper limit for the plotting option. The default is 0.001.
        show_plot : bool, optional
            If set to true, the data is plotted. The default is False.

        Returns
        -------
        Pandas Dataframe

        """

        # init results list
        list_dfs = []

        # loop samples
        for sample, data in self._iter_samples(regex=regex):
            # get peak as "right border"
            _peaks = self.get_peaks(
                processparams,
                # cutoff_min=cutoff_min,
                regex=pathlib.Path(sample).name,
                # prominence=processparams.gradient_peak_prominence, # prominence,
                show_plot=show_plot,
            )

            # identify "dormant period" as range between initial spike
            # and first reaction peak

            if show_plot:
                # plot
                plt.plot(
                    data["time_s"],
                    data["normalized_heat_flow_w_g"],
                    # linestyle="",
                    # marker="o",
                )

            # discard points at early age
            data = data.query("time_s >= @processparams.cutoff.cutoff_min * 60")
            if not _peaks.empty:
                # discard points after the first peak
                data = data.query('time_s <= @_peaks["time_s"].min()')

            # reset index
            data = data.reset_index(drop=True)

            # pick relevant points at minimum heat flow
            data = data.iloc[data["normalized_heat_flow_w_g"].idxmin(), :].to_frame().T

            if show_plot:
                # guide to the eye lines
                plt.axhline(float(data["normalized_heat_flow_w_g"]), color="red")
                plt.axvline(float(data["time_s"]), color="red")
                # indicate cutoff time
                plt.axvspan(0, cutoff_min * 60, color="black", alpha=0.5)
                # limits
                # plt.xlim(0, _peaks["time_s"].min())
                plt.xlim(0, plot_right_boundary)
                plt.ylim(0, upper_dormant_thresh_w_g)
                # title
                plt.title(pathlib.Path(sample).stem)
                # show
                plt.show()

            # add to list
            list_dfs.append(data)

        # convert to overall datafram
        result = pd.concat(list_dfs).reset_index(drop=True)

        # return
        return result

    #
    # get ASTM C1679 characteristics
    #

    def get_astm_c1679_characteristics(
        self,
        processparams,
        individual: bool = False,
        show_plot=False,
        ax=None,
        regex=None,
        xscale="log",
        xunit="s",
    ) -> pd.DataFrame:
        """
        get characteristics according to ASTM C1679. Compiles a list of data
        points at half-maximum "normalized heat flow", wherein the half maximum
        is either determined for each individual heat flow curve individually
        or as the mean value if the heat flow curves considered.

        Parameters
        ----------
        individual : bool, optional
            DESCRIPTION. The default is False.
        processparams: ProcessingParameters
            Dataclass containing parameters which control the processing of the calorimetry data.

        Returns
        -------
        Pandas Dataframe

        Examples
        --------
        Assuming that the calorimetry data is contained in a subfolder `data`, the time according to ASTM c1679 can be obtained by

        >>> from CaloCem import tacalorimetry as ta
        >>> from pathlib import Path
        >>>
        >>> thepath = Path(__file__).parent / "data"
        >>> tam = ta.Measurement(thepath)
        >>> astm = tam.get_astm_c1679_characteristics()
        """

        # get peaks
        peaks = self.get_peaks(processparams, plt_right_s=4e5, show_plot=False)
        # sort peaks by ascending normalized heat flow
        peaks = peaks.sort_values(by="normalized_heat_flow_w_g", ascending=True)
        # select highest peak --> ASTM C1679
        peaks = peaks.groupby(by="sample").last()

        # get data
        data = self.get_data()

        # init empty list for collecting characteristics
        astm_times = []

        # loop samples
        for sample, sample_data in self._iter_samples(regex=regex):
            # pick sample data
            helper = data[data["sample"] == sample]
            helper_df = helper.copy()

            # check if peak was found
            if peaks[peaks["sample_short"] == sample_data.sample_short.iloc[0]].empty:
                helper = helper.iloc[0:1]
                # manually set time to NaN to indicate that no peak was found
                helper["time_s"] = np.nan

            else:
                # restrict to times before the peak
                helper = helper[helper["time_s"] <= peaks.at[sample, "time_s"]]

                # restrict to relevant heatflows the peak
                if individual == True:
                    helper = helper[
                        helper["normalized_heat_flow_w_g"]
                        <= peaks.at[sample, "normalized_heat_flow_w_g"] * 0.50
                    ]
                else:
                    # use half-maximum average
                    helper = helper[
                        helper["normalized_heat_flow_w_g"]
                        <= peaks["normalized_heat_flow_w_g"].mean() * 0.50
                    ]

                # add to list of of selected points
            astm_times.append(helper.tail(1))

            if helper.tail(1)["time_s"].isna().all():
                continue

            if show_plot:
                # plot
                if xunit == "h":
                    helper_df["time_s"] = helper_df["time_s"] / 3600
                    helper["time_s"] = helper["time_s"] / 3600
                if isinstance(ax, matplotlib.axes._axes.Axes):
                    ax.plot(
                        helper_df["time_s"],
                        helper_df["normalized_heat_flow_w_g"],
                        label=sample,
                    )
                    ax.plot(
                        helper.tail(1)["time_s"],
                        helper.tail(1)["normalized_heat_flow_w_g"],
                        marker="o",
                        color="red",
                    )
                    ax.vlines(
                        x=helper.tail(1)["time_s"],
                        ymin=0,
                        ymax=helper.tail(1)["normalized_heat_flow_w_g"],
                        color="red",
                        linestyle="--",
                    )
                    ax.text(
                        x=helper.tail(1)["time_s"],
                        y=helper.tail(1)["normalized_heat_flow_w_g"] / 2,
                        s=r" $t_{ASTM}$ ="
                        + f"{helper.tail(1)['time_s'].values[0]:.1f}",
                        color="red",
                    )
                else:
                    plt.plot(
                        data["time_s"],
                        data["normalized_heat_flow_w_g"],
                        label=sample,
                    )

        # build overall DataFrame
        astm_times = pd.concat(astm_times)

        # return
        return astm_times

    #
    # get data
    #

    def get_data(self):
        """
        A convenience function which returns the Pandas Dataframe containing the read and processed calorimetry data.
        Returns
        -------
        Pandas DataFrame

        Examples
        --------
        Assuming that the calorimetry data is contained in a subfolder `data`, a conventional Pandas dataframe `df` containing the data from all calorimetry files in `data` can be obtained with the following code.

        >>> from CaloCem import tacalorimetry as ta
        >>> from pathlib import Path
        >>>
        >>> thepath = Path(__file__).parent / "data"
        >>> tam = ta.Measurement(thepath)
        >>> df = tam.get_data()

        """

        return self._data

    #
    # get information
    #

    def get_information(self):
        """
        get information

        Returns
        -------
        pd.DataFrame
            information, i.e. date of measurement, operator, comment ...

        """

        return self._info

    #
    # get added metadata
    #
    def get_metadata(self) -> tuple:
        """


         Returns
         -------
        tuple
             pd.DataFrame of metadata and string of the column used as ID (has to
             be unique).
        """

        # return
        return self._metadata, self._metadata_id

    #
    # get sample names
    #

    def get_sample_names(self):
        """
        get list of sample names

        Returns
        -------
        None.

        """

        # get list
        samples = [pathlib.Path(s).stem for s, _ in self._iter_samples()]

        # return
        return samples

    #
    # set
    #

    def normalize_sample_to_mass(
        self, sample_short: str, mass_g: float, show_info=True
    ):
        """
        normalize "heat_flow" to a certain mass

        Parameters
        ----------
        sample_short : str
            "sample_short" name of sample to be normalized.
        mass_g : float
            mass in gram to which "heat_flow_w" are normalized.

        Returns
        -------
        None.

        """

        # normalize "heat_flow_w" to sample mass
        self._data.loc[
            self._data["sample_short"] == sample_short, "normalized_heat_flow_w_g"
        ] = (
            self._data.loc[self._data["sample_short"] == sample_short, "heat_flow_w"]
            / mass_g
        )

        # normalize "heat_j" to sample mass
        try:
            self._data.loc[
                self._data["sample_short"] == sample_short, "normalized_heat_j_g"
            ] = (
                self._data.loc[self._data["sample_short"] == sample_short, "heat_j"]
                / mass_g
            )
        except Exception:
            pass

        # info
        if show_info:
            print(f"Sample {sample_short} normalized to {mass_g}g sample.")

    #
    # infer "heat_j" values
    #

    def _infer_heat_j_column(self):
        """
        helper function to calculate the "heat_j" columns from "heat_flow_w" and
        "time_s" columns

        Returns
        -------
        None.

        """

        # list of dfs
        list_of_dfs = []

        # loop samples
        for sample, roi in self._iter_samples():
            # check whether a "native" "heat_j"-column is available
            try:
                if not roi["heat_j"].isna().all():
                    # use as is
                    list_of_dfs.append(roi)
                    # go to next
                    continue
            except KeyError as e:
                # info
                print(e)

            # info
            print(f'==> Inferring "heat_j" column for {sample}')

            # get target rows
            roi = roi.dropna(subset=["heat_flow_w"]).sort_values(by="time_s")

            # inferring cumulated heat using the "trapezoidal integration method"

            # introduce helpers
            roi["_h1_y"] = 0.5 * (
                roi["heat_flow_w"] + roi["heat_flow_w"].shift(1)
            ).shift(-1)
            roi["_h2_x"] = (roi["time_s"] - roi["time_s"].shift(1)).shift(-1)

            # integrate
            roi["heat_j"] = (roi["_h1_y"] * roi["_h2_x"]).cumsum()

            # clean
            del roi["_h1_y"], roi["_h2_x"]

            # append to list
            list_of_dfs.append(roi)

        # set data including "heat_j"
        self._data = pd.concat(list_of_dfs)

    #
    # remove pickle files
    #
    def remove_pickle_files(self):
        """
        remove pickle files if re-reading of source files needed

        Returns
        -------
        None.

        """

        # remove files
        for file in [self._file_data_pickle, self._file_info_pickle]:
            # remove file
            pathlib.Path(file).unlink()

    #
    # add metadata
    #
    def add_metadata_source(self, file: str, sample_id_column: str):
        """
        add an additional source of metadata the object. The source file is of
        type "csv" or "xlsx" and holds information on one sample per row. Columns
        can be named without restrictions.

        To allow for a mapping, the values occurring in self._data["sample_short"]
        should appear in the source file. The column is declared via the keyword
        "sample_id_colum"

        Parameters
        ----------
        file : str
            path to additonal metadata source file.
        sample_id_column : str
            column name in the additional source file matching self._data["sample_short"].

        Returns
        -------
        None.

        """

        if not pathlib.Path(file).suffix.lower() in [".csv", ".xlsx"]:
            # info
            print("Please use metadata files of type csv and xlsx only.")
            # return
            return

        # read file
        try:
            # read as Excel
            self._metadata = pd.read_excel(file)
        except ValueError:
            # read as csv
            self._metadata = pd.read_csv(file)

        # save mapper column
        if sample_id_column in self._metadata.columns:
            # save mapper column
            self._metadata_id = sample_id_column
        else:
            # raise custom Exception
            raise AddMetaDataSourceException(self._metadata.columns.tolist())

    #
    # get metadata group-by options
    #
    def get_metadata_grouping_options(self) -> list:
        """
        get a list of categories to group by in in "self.plot_by_category"

        Returns
        -------
        list
            list of categories avaialble for grouping by.
        """

        # get list based on column names of "_metadata"
        return self._metadata.columns.tolist()

    #
    # average by metadata
    #
    def average_by_metadata(
        self,
        group_by: str,
        meta_id="experiment_nr",
        data_id="sample_short",
        time_average_window_s: int = None,
        time_average_log_bin_count: int = None,
        time_s_max: int = 2 * 24 * 60 * 60,
        get_time_from="left",
        resampling_s: str = "5s",
    ):
        """


        Parameters
        ----------
        group_by : str | list[str]
            DESCRIPTION.
        meta_id : TYPE, optional
            DESCRIPTION. The default is "experiment_nr".
        data_id : TYPE, optional
            DESCRIPTION. The default is "sample_short".
        time_average_window_s : TYPE, optional
            DESCRIPTION. The default is 60. The value is not(!) consindered if
            the keyword time_average_log_bin_count is specified
        get_time_from : TYPE, optional
            DESCRIPTION. The default is "left". further options: # "mid" "right"

        time_average_log_bin_count: number of bins if even spacing in logarithmic scale is applied

        Returns
        -------
        None.

        """

        # get metadata
        meta, meta_id = self.get_metadata()

        # get data
        df = self._data

        # make data equidistant grouped by sample_short
        df = (
            df.groupby(data_id)
            .apply(lambda x: utils.apply_resampling(x, resampling_s))
            .reset_index(drop=True)
        )

        # rename sample in "data" by metadata grouping options
        for value, group in meta.groupby(group_by):
            # if one grouping level is used
            if isinstance(value, str) or isinstance(value, int):
                # modify data --> replace "sample_short" with metadata group name
                _idx_to_replace = df[data_id].isin(group[meta_id])
                df.loc[_idx_to_replace, data_id] = str(value)
            # if multiple grouping levels are used
            elif isinstance(value, tuple):
                # modify data --> replace "sample_short" with metadata group name
                _idx_to_replace = df[data_id].isin(group[meta_id])
                df.loc[_idx_to_replace, data_id] = " | ".join([str(x) for x in value])
            else:
                pass

        # sort experimentally detected times to "bins"
        if time_average_log_bin_count:
            # evenly spaced bins on log scale (geometric spacing)
            df["BIN"] = pd.cut(
                df["time_s"],
                np.geomspace(1, time_s_max, num=time_average_log_bin_count),
            )
        elif time_average_window_s:
            # evenly spaced bins on linear scale with fixed width
            df["BIN"] = pd.cut(
                df["time_s"], np.arange(0, time_s_max, time_average_window_s)
            )

        if "BIN" in df.columns:
            # calculate average and std
            df = (
                df.groupby([data_id, "BIN"])
                .agg(
                    {
                        "normalized_heat_flow_w_g": ["mean", "std"],
                        "normalized_heat_j_g": ["mean", "std"],
                    }
                )
                .dropna(thresh=2)
                .reset_index()
            )
        else:
            # calculate average and std
            df = (
                df.groupby([data_id, "time_s"])
                .agg(
                    {
                        "normalized_heat_flow_w_g": ["mean", "std"],
                        "normalized_heat_j_g": ["mean", "std"],
                    }
                )
                .dropna(thresh=2)
                .reset_index()
            )

        # "flatten" column names
        df.columns = ["_".join(i).replace("mean", "_").strip("_") for i in df.columns]

        if "BIN" in df.columns:
            # regain "time_s" columns
            if get_time_from == "left":
                df["time_s"] = [i.left for i in df["BIN"]]
            elif get_time_from == "mid":
                df["time_s"] = [i.mid for i in df["BIN"]]
            elif get_time_from == "right":
                df["time_s"] = [i.right for i in df["BIN"]]

            # remove "BIN" auxiliary column
            del df["BIN"]

        # copy information to "sample" column --> needed for plotting
        df["sample"] = df[data_id]

        # overwrite data with averaged data
        self._data = df

    #
    # undo action of "average_by_metadata"
    #
    def undo_average_by_metadata(self):
        """
        undo action of "average_by_metadata"
        """

        # set "unprocessed" data as exeperimental data / "de-average"
        if not self._data_unprocessed.empty:
            # reset
            self._data = self._data_unprocessed.copy()

    #
    # apply_tian_correction
    #
    def apply_tian_correction(
        self,
        processparams,  # tau=300, window=11, polynom=3, spline_smoothing_1st: float = 1e-9, spline_smoothing_2nd: float = 1e-9
    ) -> None:
        """
        apply_tian_correction

        Parameters
        ----------

        processparams :
            ProcessingParameters object containing all processing parameters for calorimetry data.
        Returns
        -------
        None.

        """

        # apply the correction for each sample
        for s, d in self._iter_samples():
            # get y-data
            y = d["normalized_heat_flow_w_g"]
            # NaN-handling in y-data
            y = y.fillna(0)
            # get x-data
            x = d["time_s"]

            processor = HeatFlowProcessor(processparams)

            dydx, dy2dx2 = processor.calculate_heatflow_derivatives(d)

            if processparams.time_constants.tau2 == None:
                # calculate corrected heatflow
                norm_hf = (
                    dydx * processparams.time_constants.tau1
                    + self._data.loc[
                        self._data["sample"] == s, "normalized_heat_flow_w_g"
                    ]
                )
            else:
                # calculate corrected heatflow
                norm_hf = (
                    dydx
                    * (
                        processparams.time_constants.tau1
                        + processparams.time_constants.tau2
                    )
                    + dy2dx2
                    * processparams.time_constants.tau1
                    * processparams.time_constants.tau2
                    + d["normalized_heat_flow_w_g"]
                )

            self._data.loc[
                self._data["sample"] == s, "normalized_heat_flow_w_g_tian"
            ] = norm_hf

            self._data.loc[
                self._data["sample"] == s, "gradient_normalized_heat_flow_w_g"
            ] = dydx

            # calculate corresponding cumulative heat
            self._data.loc[self._data["sample"] == s, "normalized_heat_j_g_tian"] = (
                integrate.cumulative_trapezoid(norm_hf.fillna(0), x=x, initial=0)
            )

    #
    # undo Tian-correction
    #
    def undo_tian_correction(self):
        """
        undo_tian_correction; i.e. restore original data


        Returns
        -------
        None.

        """

        # call original restore function
        self.undo_average_by_metadata()

    def _apply_adaptive_downsampling(self):
        """
        apply adaptive downsampling to data
        """

        # define temporary empty DataFrame
        df = pd.DataFrame()

        # apply the correction for each sample
        for s, d in self._iter_samples():
            # print(d.sample_short[0])
            # print(len(d))
            d = d.dropna(subset=["normalized_heat_flow_w_g"])

            processor = HeatFlowProcessor(self.processparams)
            d = processor.restrict_data_range(d)
            # apply adaptive downsampling
            if not self.processparams.downsample.section_split:
                d = utils.adaptive_downsample(
                    d,
                    x_col="time_s",
                    y_col="normalized_heat_flow_w_g",
                    processparams=self.processparams,
                )
            else:
                d = utils.downsample_sections(
                    d,
                    x_col="time_s",
                    y_col="normalized_heat_flow_w_g",
                    processparams=self.processparams,
                )
            df = pd.concat([df, d])

        # set data to downsampled data
        self._data = df

    def get_ascending_flank_tangent(
        self,
        processparams,
        target_col="normalized_heat_flow_w_g",
        age_col="time_s",
        flank_fraction_start=0.2,  # Start at 20% of peak height
        flank_fraction_end=0.8,  # End at 80% of peak height
        window_size=0.1,  # Window size as fraction of flank range
        cutoff_min=None,  # Initial cutoff time in minutes to ignore
        show_plot=False,
        regex=None,
        plotpath=None,
    ):
        """
        Determine tangent to ascending flank of peak by averaging over sections.

        Parameters
        ----------
        target_col : str
            Column containing heat flow data
        age_col : str
            Column containing time data
        flank_fraction_start : float
            Start of flank section as fraction of peak height (0-1)
        flank_fraction_end : float
            End of flank section as fraction of peak height (0-1)
        window_size : float
            Size of averaging window as fraction of flank time range
        cutoff_min : float, optional
            Initial cutoff time in minutes to ignore from analysis. If None,
            uses processparams.cutoff.cutoff_min. The default is None.
        show_plot : bool
            Whether to plot the results
        regex : str
            Regex to filter samples

        Returns
        -------
        pd.DataFrame
            DataFrame with tangent characteristics for each sample
        """

        results = []

        for sample, data in self._iter_samples(regex=regex):
            # Apply cutoff if specified - use parameter cutoff_min first, then fallback to processparams
            cutoff_time_min = (
                cutoff_min
                if cutoff_min is not None
                else processparams.cutoff.cutoff_min
            )
            if cutoff_time_min:
                data = data.query(f"{age_col} >= @cutoff_time_min * 60")

            data = data.reset_index(drop=True)

            # Find the main peak
            peaks, _ = signal.find_peaks(
                data[target_col],
                prominence=processparams.peakdetection.prominence,
                distance=processparams.peakdetection.distance,
            )

            if len(peaks) == 0:
                print(f"No peak found in {sample}")
                continue

            # Use the highest peak
            peak_idx = peaks[np.argmax(data.iloc[peaks][target_col])]
            peak_time = data.iloc[peak_idx][age_col]
            peak_value = data.iloc[peak_idx][target_col]

            # Find baseline (minimum before peak)
            baseline_data = data[data[age_col] < peak_time]
            if len(baseline_data) == 0:
                baseline_value = 0
            else:
                baseline_value = baseline_data[target_col].min()

            # Define flank region
            flank_height_range = peak_value - baseline_value
            flank_start_value = (
                baseline_value + flank_fraction_start * flank_height_range
            )
            flank_end_value = baseline_value + flank_fraction_end * flank_height_range

            # Calculate gradient to ensure we only consider regions with positive slope
            data['gradient'] = np.gradient(data[target_col], data[age_col])

            # Extract ascending flank data - only include points with positive gradient
            flank_data = data[
                (data[target_col] >= flank_start_value)
                & (data[target_col] <= flank_end_value)
                & (data[age_col] <= peak_time)
                & (data['gradient'] > 0)  # Only positive gradients
            ].copy()

            # If no positive gradient data in initial range, try to find the lowest point with positive gradient
            if len(flank_data) < 3:
                # Find data points with positive gradient before peak
                positive_gradient_data = data[
                    (data[age_col] <= peak_time) & (data['gradient'] > 0)
                ]

                if len(positive_gradient_data) >= 3:
                    # Adjust flank start to the minimum value with positive gradient
                    min_positive_value = positive_gradient_data[target_col].min()
                    adjusted_flank_start = max(flank_start_value, min_positive_value)

                    flank_data = data[
                        (data[target_col] >= adjusted_flank_start)
                        & (data[target_col] <= flank_end_value)
                        & (data[age_col] <= peak_time)
                        & (data['gradient'] > 0)
                    ].copy()

                    # Update the flank_start_value for recording
                    flank_start_value = adjusted_flank_start

            if len(flank_data) < 3:
                print(f"Insufficient flank data in {sample}")
                continue

            # Calculate moving tangents over windows
            flank_time_range = flank_data[age_col].max() - flank_data[age_col].min()
            window_time = window_size * flank_time_range

            tangent_slopes = []
            tangent_times = []
            tangent_values = []

            # Slide window across flank
            start_time = flank_data[age_col].min()
            end_time = flank_data[age_col].max() - window_time

            step_size = window_time * 0.1  # 10% overlap
            current_time = start_time

            while current_time <= end_time:
                window_end = current_time + window_time
                window_data = flank_data[
                    (flank_data[age_col] >= current_time)
                    & (flank_data[age_col] <= window_end)
                ]

                if len(window_data) >= 3:
                    # Linear regression for this window
                    x = window_data[age_col].values
                    y = window_data[target_col].values

                    # Use numpy polyfit for linear regression
                    slope, intercept = np.polyfit(x, y, 1)

                    # Only consider positive gradients (ascending flank)
                    if slope > 0:
                        # Store tangent info at window center
                        center_time = (current_time + window_end) / 2
                        center_value = slope * center_time + intercept

                        tangent_slopes.append(slope)
                        tangent_times.append(center_time)
                        tangent_values.append(center_value)

                current_time += step_size

            if not tangent_slopes:
                print(
                    f"No valid tangent windows with positive gradients found in {sample}"
                )
                continue

            # Calculate representative tangent (median to avoid outliers)
            representative_slope = np.median(tangent_slopes)
            representative_time = np.median(tangent_times)
            representative_value = np.median(tangent_values)

            # Calculate tangent line parameters
            # y = mx + b, so b = y - mx
            tangent_intercept = (
                representative_value - representative_slope * representative_time
            )
            # calculate x intection
            # y=0, so x = -b/m
            x_intersection = (
                -tangent_intercept / representative_slope if representative_slope != 0 else np.nan
            )

            # Calculate intersection with horizontal line at minimum before tangent_time_s
            data_before_tangent = data[data[age_col] <= representative_time]
            if len(data_before_tangent) > 0:
                min_value_before_tangent = data_before_tangent[target_col].min()
                # Intersection: y = min_value = slope * x + intercept
                # x = (y - intercept) / slope
                x_intersection_min = (
                    (min_value_before_tangent - tangent_intercept) / representative_slope 
                    if representative_slope != 0 else np.nan
                )
            else:
                min_value_before_tangent = np.nan
                x_intersection_min = np.nan

            result = {
                "sample": sample,
                "sample_short": pathlib.Path(sample).stem,
                "peak_time_s": peak_time,
                "peak_value": peak_value,
                "tangent_slope": representative_slope,
                "tangent_time_s": representative_time,
                "tangent_value": representative_value,
                "tangent_intercept": tangent_intercept,
                "flank_start_value": flank_start_value,
                "flank_end_value": flank_end_value,
                "n_windows": len(tangent_slopes),
                "slope_std": np.std(tangent_slopes),
                "x_intersection": x_intersection,
                "min_value_before_tangent": min_value_before_tangent,
                "x_intersection_min": x_intersection_min,
            }

            results.append(result)

            # Optional plotting
            if show_plot:
                fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(10, 8))

                # Plot 1: Full curve with peak and flank region
                ax1.plot(data[age_col], data[target_col], "b-", alpha=0.7, label="Data")
                ax1.axvline(
                    peak_time, color="red", linestyle="--", alpha=0.7, label="Peak"
                )
                ax1.axhline(flank_start_value, color="green", linestyle=":", alpha=0.7)
                ax1.axhline(flank_end_value, color="green", linestyle=":", alpha=0.7)
                ax1.fill_between(
                    flank_data[age_col],
                    flank_start_value,
                    flank_end_value,
                    alpha=0.2,
                    color="green",
                    label="Flank region",
                )

                # Plot tangent line
                x_tangent = np.linspace(x_intersection, peak_time, 10)
                y_tangent = representative_slope * x_tangent + tangent_intercept
                ax1.plot(
                    x_tangent, y_tangent, "r-", linewidth=2, label="Average tangent"
                )

                # Add text label for tangent slope
                mid_x = (x_intersection + peak_time) / 2
                mid_y = representative_slope * mid_x + tangent_intercept
                ax1.annotate(
                    f"Slope: {representative_slope:.2e}",
                    xy=(mid_x, mid_y),
                    xytext=(10, 10),
                    textcoords='offset points',
                    fontsize=10,
                    color='red',
                    bbox=dict(boxstyle='round,pad=0.3', facecolor='white', alpha=0.8)
                )

                # Add vertical line and label for x_intersection
                if not np.isnan(x_intersection) and x_intersection > data[age_col].min():
                    ax1.axvline(x_intersection, color='orange', linestyle=':', alpha=0.8,) 
                            #    label=f'X-intersection: {x_intersection:.0f}s')
                    ax1.annotate(
                        f"{x_intersection:.0f}s",
                        xy=(x_intersection, baseline_value),
                        xytext=(-50, 20),
                        textcoords='offset points',
                        fontsize=10,
                        color='orange',
                        bbox=dict(boxstyle='round,pad=0.3', facecolor='white', alpha=0.8),
                        arrowprops=dict(arrowstyle='->', color='orange', alpha=0.6)
                    )

                # Add horizontal line at minimum and its intersection with tangent
                if not np.isnan(min_value_before_tangent) and not np.isnan(x_intersection_min):
                    # Draw horizontal line at minimum value
                    ax1.axhline(min_value_before_tangent, color='purple', linestyle='--', 
                               alpha=0.7, label=f'Min before tangent: {min_value_before_tangent:.4f}')

                    # Add vertical line at intersection with minimum
                    if x_intersection_min > data[age_col].min() and x_intersection_min < peak_time:
                        ax1.axvline(x_intersection_min, color='purple', linestyle=':', alpha=0.8)
                        ax1.annotate(
                            f"Min-int: {x_intersection_min:.0f}s",
                            xy=(x_intersection_min, min_value_before_tangent),
                            xytext=(10, -30),
                            textcoords='offset points',
                            fontsize=10,
                            color='purple',
                            bbox=dict(boxstyle='round,pad=0.3', facecolor='white', alpha=0.8),
                            arrowprops=dict(arrowstyle='->', color='purple', alpha=0.6)
                        )

                ax1.set_xlabel(age_col)
                ax1.set_ylabel(target_col)
                ax1.set_title(f"Peak Analysis: {pathlib.Path(sample).stem}")
                ax1.legend()
                # ax1.grid(True, alpha=0.3)
                ax1.set_ylim(0,)

                # Plot 2: Slope variation across windows
                ax2.plot(tangent_times, tangent_slopes, "bo-", alpha=0.7)
                ax2.axhline(
                    representative_slope,
                    color="red",
                    linestyle="--",
                    label=f"Median slope: {representative_slope:.2e}",
                )
                ax2.set_xlabel("Window center time (s)")
                ax2.set_ylabel("Local slope")
                ax2.set_title("Slope variation across flank windows")
                ax2.legend()
                # ax2.grid(True, alpha=0.3)
                # ax2.set_ylim(0,)

                plt.tight_layout()
                if plotpath is not None:
                    filename = pathlib.Path(sample).stem
                    plt.savefig(plotpath / f"{filename}.png")
                else:
                    plt.show()

        return pd.DataFrame(results)

__init__(folder=None, show_info=True, regex=None, auto_clean=False, cold_start=True, processparams=None, new_code=False, processed=False)

intialize measurements from folder

Source code in calocem/tacalorimetry.py

def __init__(
    self,
    folder=None,
    show_info=True,
    regex=None,
    auto_clean=False,
    cold_start=True,
    processparams=None,
    new_code=False,
    processed=False,
):
    """
    intialize measurements from folder


    """
    self._new_code = new_code
    self._processed = processed

    if not isinstance(processparams, ProcessingParameters):
        self.processparams = ProcessingParameters()
    else:
        self.processparams = processparams

    # read
    if folder:
        if cold_start:
            # get data and parameters
            self._get_data_and_parameters_from_folder(
                folder, regex=regex, show_info=show_info
            )
        else:
            # get data and parameters from pickled files
            self._get_data_and_parameters_from_pickle()
        try:
            if auto_clean:
                # remove NaN values and merge time columns
                self._auto_clean_data()
        except Exception as e:
            # info
            print(e)
            raise AutoCleanException
            # return
            return

    if self.processparams.downsample.apply:
        self._apply_adaptive_downsampling()
    # Message
    print(
        "================\nAre you missing some samples? Try rerunning with auto_clean=True and cold_start=True.\n================="
    )

add_metadata_source(file, sample_id_column)

add an additional source of metadata the object. The source file is of type "csv" or "xlsx" and holds information on one sample per row. Columns can be named without restrictions.

To allow for a mapping, the values occurring in self._data["sample_short"] should appear in the source file. The column is declared via the keyword "sample_id_colum"

Parameters:

Name	Type	Description	Default
file	str	path to additonal metadata source file.	required
sample_id_column	str	column name in the additional source file matching self._data["sample_short"].	required

Returns:

Type	Description
None.

Source code in calocem/tacalorimetry.py

def add_metadata_source(self, file: str, sample_id_column: str):
    """
    add an additional source of metadata the object. The source file is of
    type "csv" or "xlsx" and holds information on one sample per row. Columns
    can be named without restrictions.

    To allow for a mapping, the values occurring in self._data["sample_short"]
    should appear in the source file. The column is declared via the keyword
    "sample_id_colum"

    Parameters
    ----------
    file : str
        path to additonal metadata source file.
    sample_id_column : str
        column name in the additional source file matching self._data["sample_short"].

    Returns
    -------
    None.

    """

    if not pathlib.Path(file).suffix.lower() in [".csv", ".xlsx"]:
        # info
        print("Please use metadata files of type csv and xlsx only.")
        # return
        return

    # read file
    try:
        # read as Excel
        self._metadata = pd.read_excel(file)
    except ValueError:
        # read as csv
        self._metadata = pd.read_csv(file)

    # save mapper column
    if sample_id_column in self._metadata.columns:
        # save mapper column
        self._metadata_id = sample_id_column
    else:
        # raise custom Exception
        raise AddMetaDataSourceException(self._metadata.columns.tolist())

apply_tian_correction(processparams)

apply_tian_correction

Parameters:

Name	Type	Description	Default
processparams		ProcessingParameters object containing all processing parameters for calorimetry data.	required

Returns:

Type	Description
None.

Source code in calocem/tacalorimetry.py

def apply_tian_correction(
    self,
    processparams,  # tau=300, window=11, polynom=3, spline_smoothing_1st: float = 1e-9, spline_smoothing_2nd: float = 1e-9
) -> None:
    """
    apply_tian_correction

    Parameters
    ----------

    processparams :
        ProcessingParameters object containing all processing parameters for calorimetry data.
    Returns
    -------
    None.

    """

    # apply the correction for each sample
    for s, d in self._iter_samples():
        # get y-data
        y = d["normalized_heat_flow_w_g"]
        # NaN-handling in y-data
        y = y.fillna(0)
        # get x-data
        x = d["time_s"]

        processor = HeatFlowProcessor(processparams)

        dydx, dy2dx2 = processor.calculate_heatflow_derivatives(d)

        if processparams.time_constants.tau2 == None:
            # calculate corrected heatflow
            norm_hf = (
                dydx * processparams.time_constants.tau1
                + self._data.loc[
                    self._data["sample"] == s, "normalized_heat_flow_w_g"
                ]
            )
        else:
            # calculate corrected heatflow
            norm_hf = (
                dydx
                * (
                    processparams.time_constants.tau1
                    + processparams.time_constants.tau2
                )
                + dy2dx2
                * processparams.time_constants.tau1
                * processparams.time_constants.tau2
                + d["normalized_heat_flow_w_g"]
            )

        self._data.loc[
            self._data["sample"] == s, "normalized_heat_flow_w_g_tian"
        ] = norm_hf

        self._data.loc[
            self._data["sample"] == s, "gradient_normalized_heat_flow_w_g"
        ] = dydx

        # calculate corresponding cumulative heat
        self._data.loc[self._data["sample"] == s, "normalized_heat_j_g_tian"] = (
            integrate.cumulative_trapezoid(norm_hf.fillna(0), x=x, initial=0)
        )

average_by_metadata(group_by, meta_id='experiment_nr', data_id='sample_short', time_average_window_s=None, time_average_log_bin_count=None, time_s_max=2 * 24 * 60 * 60, get_time_from='left', resampling_s='5s')

Parameters:

Name	Type	Description	Default
group_by	str | list[str]	DESCRIPTION.	required
meta_id	TYPE	DESCRIPTION. The default is "experiment_nr".	'experiment_nr'
data_id	TYPE	DESCRIPTION. The default is "sample_short".	'sample_short'
time_average_window_s	TYPE	DESCRIPTION. The default is 60. The value is not(!) consindered if the keyword time_average_log_bin_count is specified	None
get_time_from	TYPE	DESCRIPTION. The default is "left". further options: # "mid" "right"	'left'
time_average_log_bin_count	int		None

Returns:

Type	Description
None.

Source code in calocem/tacalorimetry.py

def average_by_metadata(
    self,
    group_by: str,
    meta_id="experiment_nr",
    data_id="sample_short",
    time_average_window_s: int = None,
    time_average_log_bin_count: int = None,
    time_s_max: int = 2 * 24 * 60 * 60,
    get_time_from="left",
    resampling_s: str = "5s",
):
    """


    Parameters
    ----------
    group_by : str | list[str]
        DESCRIPTION.
    meta_id : TYPE, optional
        DESCRIPTION. The default is "experiment_nr".
    data_id : TYPE, optional
        DESCRIPTION. The default is "sample_short".
    time_average_window_s : TYPE, optional
        DESCRIPTION. The default is 60. The value is not(!) consindered if
        the keyword time_average_log_bin_count is specified
    get_time_from : TYPE, optional
        DESCRIPTION. The default is "left". further options: # "mid" "right"

    time_average_log_bin_count: number of bins if even spacing in logarithmic scale is applied

    Returns
    -------
    None.

    """

    # get metadata
    meta, meta_id = self.get_metadata()

    # get data
    df = self._data

    # make data equidistant grouped by sample_short
    df = (
        df.groupby(data_id)
        .apply(lambda x: utils.apply_resampling(x, resampling_s))
        .reset_index(drop=True)
    )

    # rename sample in "data" by metadata grouping options
    for value, group in meta.groupby(group_by):
        # if one grouping level is used
        if isinstance(value, str) or isinstance(value, int):
            # modify data --> replace "sample_short" with metadata group name
            _idx_to_replace = df[data_id].isin(group[meta_id])
            df.loc[_idx_to_replace, data_id] = str(value)
        # if multiple grouping levels are used
        elif isinstance(value, tuple):
            # modify data --> replace "sample_short" with metadata group name
            _idx_to_replace = df[data_id].isin(group[meta_id])
            df.loc[_idx_to_replace, data_id] = " | ".join([str(x) for x in value])
        else:
            pass

    # sort experimentally detected times to "bins"
    if time_average_log_bin_count:
        # evenly spaced bins on log scale (geometric spacing)
        df["BIN"] = pd.cut(
            df["time_s"],
            np.geomspace(1, time_s_max, num=time_average_log_bin_count),
        )
    elif time_average_window_s:
        # evenly spaced bins on linear scale with fixed width
        df["BIN"] = pd.cut(
            df["time_s"], np.arange(0, time_s_max, time_average_window_s)
        )

    if "BIN" in df.columns:
        # calculate average and std
        df = (
            df.groupby([data_id, "BIN"])
            .agg(
                {
                    "normalized_heat_flow_w_g": ["mean", "std"],
                    "normalized_heat_j_g": ["mean", "std"],
                }
            )
            .dropna(thresh=2)
            .reset_index()
        )
    else:
        # calculate average and std
        df = (
            df.groupby([data_id, "time_s"])
            .agg(
                {
                    "normalized_heat_flow_w_g": ["mean", "std"],
                    "normalized_heat_j_g": ["mean", "std"],
                }
            )
            .dropna(thresh=2)
            .reset_index()
        )

    # "flatten" column names
    df.columns = ["_".join(i).replace("mean", "_").strip("_") for i in df.columns]

    if "BIN" in df.columns:
        # regain "time_s" columns
        if get_time_from == "left":
            df["time_s"] = [i.left for i in df["BIN"]]
        elif get_time_from == "mid":
            df["time_s"] = [i.mid for i in df["BIN"]]
        elif get_time_from == "right":
            df["time_s"] = [i.right for i in df["BIN"]]

        # remove "BIN" auxiliary column
        del df["BIN"]

    # copy information to "sample" column --> needed for plotting
    df["sample"] = df[data_id]

    # overwrite data with averaged data
    self._data = df

get_ascending_flank_tangent(processparams, target_col='normalized_heat_flow_w_g', age_col='time_s', flank_fraction_start=0.2, flank_fraction_end=0.8, window_size=0.1, cutoff_min=None, show_plot=False, regex=None, plotpath=None)

Determine tangent to ascending flank of peak by averaging over sections.

Parameters:

Name	Type	Description	Default
target_col	str	Column containing heat flow data	'normalized_heat_flow_w_g'
age_col	str	Column containing time data	'time_s'
flank_fraction_start	float	Start of flank section as fraction of peak height (0-1)	0.2
flank_fraction_end	float	End of flank section as fraction of peak height (0-1)	0.8
window_size	float	Size of averaging window as fraction of flank time range	0.1
cutoff_min	float	Initial cutoff time in minutes to ignore from analysis. If None, uses processparams.cutoff.cutoff_min. The default is None.	None
show_plot	bool	Whether to plot the results	False
regex	str	Regex to filter samples	None

Returns:

Type	Description
DataFrame	DataFrame with tangent characteristics for each sample

Source code in calocem/tacalorimetry.py

def get_ascending_flank_tangent(
    self,
    processparams,
    target_col="normalized_heat_flow_w_g",
    age_col="time_s",
    flank_fraction_start=0.2,  # Start at 20% of peak height
    flank_fraction_end=0.8,  # End at 80% of peak height
    window_size=0.1,  # Window size as fraction of flank range
    cutoff_min=None,  # Initial cutoff time in minutes to ignore
    show_plot=False,
    regex=None,
    plotpath=None,
):
    """
    Determine tangent to ascending flank of peak by averaging over sections.

    Parameters
    ----------
    target_col : str
        Column containing heat flow data
    age_col : str
        Column containing time data
    flank_fraction_start : float
        Start of flank section as fraction of peak height (0-1)
    flank_fraction_end : float
        End of flank section as fraction of peak height (0-1)
    window_size : float
        Size of averaging window as fraction of flank time range
    cutoff_min : float, optional
        Initial cutoff time in minutes to ignore from analysis. If None,
        uses processparams.cutoff.cutoff_min. The default is None.
    show_plot : bool
        Whether to plot the results
    regex : str
        Regex to filter samples

    Returns
    -------
    pd.DataFrame
        DataFrame with tangent characteristics for each sample
    """

    results = []

    for sample, data in self._iter_samples(regex=regex):
        # Apply cutoff if specified - use parameter cutoff_min first, then fallback to processparams
        cutoff_time_min = (
            cutoff_min
            if cutoff_min is not None
            else processparams.cutoff.cutoff_min
        )
        if cutoff_time_min:
            data = data.query(f"{age_col} >= @cutoff_time_min * 60")

        data = data.reset_index(drop=True)

        # Find the main peak
        peaks, _ = signal.find_peaks(
            data[target_col],
            prominence=processparams.peakdetection.prominence,
            distance=processparams.peakdetection.distance,
        )

        if len(peaks) == 0:
            print(f"No peak found in {sample}")
            continue

        # Use the highest peak
        peak_idx = peaks[np.argmax(data.iloc[peaks][target_col])]
        peak_time = data.iloc[peak_idx][age_col]
        peak_value = data.iloc[peak_idx][target_col]

        # Find baseline (minimum before peak)
        baseline_data = data[data[age_col] < peak_time]
        if len(baseline_data) == 0:
            baseline_value = 0
        else:
            baseline_value = baseline_data[target_col].min()

        # Define flank region
        flank_height_range = peak_value - baseline_value
        flank_start_value = (
            baseline_value + flank_fraction_start * flank_height_range
        )
        flank_end_value = baseline_value + flank_fraction_end * flank_height_range

        # Calculate gradient to ensure we only consider regions with positive slope
        data['gradient'] = np.gradient(data[target_col], data[age_col])

        # Extract ascending flank data - only include points with positive gradient
        flank_data = data[
            (data[target_col] >= flank_start_value)
            & (data[target_col] <= flank_end_value)
            & (data[age_col] <= peak_time)
            & (data['gradient'] > 0)  # Only positive gradients
        ].copy()

        # If no positive gradient data in initial range, try to find the lowest point with positive gradient
        if len(flank_data) < 3:
            # Find data points with positive gradient before peak
            positive_gradient_data = data[
                (data[age_col] <= peak_time) & (data['gradient'] > 0)
            ]

            if len(positive_gradient_data) >= 3:
                # Adjust flank start to the minimum value with positive gradient
                min_positive_value = positive_gradient_data[target_col].min()
                adjusted_flank_start = max(flank_start_value, min_positive_value)

                flank_data = data[
                    (data[target_col] >= adjusted_flank_start)
                    & (data[target_col] <= flank_end_value)
                    & (data[age_col] <= peak_time)
                    & (data['gradient'] > 0)
                ].copy()

                # Update the flank_start_value for recording
                flank_start_value = adjusted_flank_start

        if len(flank_data) < 3:
            print(f"Insufficient flank data in {sample}")
            continue

        # Calculate moving tangents over windows
        flank_time_range = flank_data[age_col].max() - flank_data[age_col].min()
        window_time = window_size * flank_time_range

        tangent_slopes = []
        tangent_times = []
        tangent_values = []

        # Slide window across flank
        start_time = flank_data[age_col].min()
        end_time = flank_data[age_col].max() - window_time

        step_size = window_time * 0.1  # 10% overlap
        current_time = start_time

        while current_time <= end_time:
            window_end = current_time + window_time
            window_data = flank_data[
                (flank_data[age_col] >= current_time)
                & (flank_data[age_col] <= window_end)
            ]

            if len(window_data) >= 3:
                # Linear regression for this window
                x = window_data[age_col].values
                y = window_data[target_col].values

                # Use numpy polyfit for linear regression
                slope, intercept = np.polyfit(x, y, 1)

                # Only consider positive gradients (ascending flank)
                if slope > 0:
                    # Store tangent info at window center
                    center_time = (current_time + window_end) / 2
                    center_value = slope * center_time + intercept

                    tangent_slopes.append(slope)
                    tangent_times.append(center_time)
                    tangent_values.append(center_value)

            current_time += step_size

        if not tangent_slopes:
            print(
                f"No valid tangent windows with positive gradients found in {sample}"
            )
            continue

        # Calculate representative tangent (median to avoid outliers)
        representative_slope = np.median(tangent_slopes)
        representative_time = np.median(tangent_times)
        representative_value = np.median(tangent_values)

        # Calculate tangent line parameters
        # y = mx + b, so b = y - mx
        tangent_intercept = (
            representative_value - representative_slope * representative_time
        )
        # calculate x intection
        # y=0, so x = -b/m
        x_intersection = (
            -tangent_intercept / representative_slope if representative_slope != 0 else np.nan
        )

        # Calculate intersection with horizontal line at minimum before tangent_time_s
        data_before_tangent = data[data[age_col] <= representative_time]
        if len(data_before_tangent) > 0:
            min_value_before_tangent = data_before_tangent[target_col].min()
            # Intersection: y = min_value = slope * x + intercept
            # x = (y - intercept) / slope
            x_intersection_min = (
                (min_value_before_tangent - tangent_intercept) / representative_slope 
                if representative_slope != 0 else np.nan
            )
        else:
            min_value_before_tangent = np.nan
            x_intersection_min = np.nan

        result = {
            "sample": sample,
            "sample_short": pathlib.Path(sample).stem,
            "peak_time_s": peak_time,
            "peak_value": peak_value,
            "tangent_slope": representative_slope,
            "tangent_time_s": representative_time,
            "tangent_value": representative_value,
            "tangent_intercept": tangent_intercept,
            "flank_start_value": flank_start_value,
            "flank_end_value": flank_end_value,
            "n_windows": len(tangent_slopes),
            "slope_std": np.std(tangent_slopes),
            "x_intersection": x_intersection,
            "min_value_before_tangent": min_value_before_tangent,
            "x_intersection_min": x_intersection_min,
        }

        results.append(result)

        # Optional plotting
        if show_plot:
            fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(10, 8))

            # Plot 1: Full curve with peak and flank region
            ax1.plot(data[age_col], data[target_col], "b-", alpha=0.7, label="Data")
            ax1.axvline(
                peak_time, color="red", linestyle="--", alpha=0.7, label="Peak"
            )
            ax1.axhline(flank_start_value, color="green", linestyle=":", alpha=0.7)
            ax1.axhline(flank_end_value, color="green", linestyle=":", alpha=0.7)
            ax1.fill_between(
                flank_data[age_col],
                flank_start_value,
                flank_end_value,
                alpha=0.2,
                color="green",
                label="Flank region",
            )

            # Plot tangent line
            x_tangent = np.linspace(x_intersection, peak_time, 10)
            y_tangent = representative_slope * x_tangent + tangent_intercept
            ax1.plot(
                x_tangent, y_tangent, "r-", linewidth=2, label="Average tangent"
            )

            # Add text label for tangent slope
            mid_x = (x_intersection + peak_time) / 2
            mid_y = representative_slope * mid_x + tangent_intercept
            ax1.annotate(
                f"Slope: {representative_slope:.2e}",
                xy=(mid_x, mid_y),
                xytext=(10, 10),
                textcoords='offset points',
                fontsize=10,
                color='red',
                bbox=dict(boxstyle='round,pad=0.3', facecolor='white', alpha=0.8)
            )

            # Add vertical line and label for x_intersection
            if not np.isnan(x_intersection) and x_intersection > data[age_col].min():
                ax1.axvline(x_intersection, color='orange', linestyle=':', alpha=0.8,) 
                        #    label=f'X-intersection: {x_intersection:.0f}s')
                ax1.annotate(
                    f"{x_intersection:.0f}s",
                    xy=(x_intersection, baseline_value),
                    xytext=(-50, 20),
                    textcoords='offset points',
                    fontsize=10,
                    color='orange',
                    bbox=dict(boxstyle='round,pad=0.3', facecolor='white', alpha=0.8),
                    arrowprops=dict(arrowstyle='->', color='orange', alpha=0.6)
                )

            # Add horizontal line at minimum and its intersection with tangent
            if not np.isnan(min_value_before_tangent) and not np.isnan(x_intersection_min):
                # Draw horizontal line at minimum value
                ax1.axhline(min_value_before_tangent, color='purple', linestyle='--', 
                           alpha=0.7, label=f'Min before tangent: {min_value_before_tangent:.4f}')

                # Add vertical line at intersection with minimum
                if x_intersection_min > data[age_col].min() and x_intersection_min < peak_time:
                    ax1.axvline(x_intersection_min, color='purple', linestyle=':', alpha=0.8)
                    ax1.annotate(
                        f"Min-int: {x_intersection_min:.0f}s",
                        xy=(x_intersection_min, min_value_before_tangent),
                        xytext=(10, -30),
                        textcoords='offset points',
                        fontsize=10,
                        color='purple',
                        bbox=dict(boxstyle='round,pad=0.3', facecolor='white', alpha=0.8),
                        arrowprops=dict(arrowstyle='->', color='purple', alpha=0.6)
                    )

            ax1.set_xlabel(age_col)
            ax1.set_ylabel(target_col)
            ax1.set_title(f"Peak Analysis: {pathlib.Path(sample).stem}")
            ax1.legend()
            # ax1.grid(True, alpha=0.3)
            ax1.set_ylim(0,)

            # Plot 2: Slope variation across windows
            ax2.plot(tangent_times, tangent_slopes, "bo-", alpha=0.7)
            ax2.axhline(
                representative_slope,
                color="red",
                linestyle="--",
                label=f"Median slope: {representative_slope:.2e}",
            )
            ax2.set_xlabel("Window center time (s)")
            ax2.set_ylabel("Local slope")
            ax2.set_title("Slope variation across flank windows")
            ax2.legend()
            # ax2.grid(True, alpha=0.3)
            # ax2.set_ylim(0,)

            plt.tight_layout()
            if plotpath is not None:
                filename = pathlib.Path(sample).stem
                plt.savefig(plotpath / f"{filename}.png")
            else:
                plt.show()

    return pd.DataFrame(results)

get_astm_c1679_characteristics(processparams, individual=False, show_plot=False, ax=None, regex=None, xscale='log', xunit='s')

get characteristics according to ASTM C1679. Compiles a list of data points at half-maximum "normalized heat flow", wherein the half maximum is either determined for each individual heat flow curve individually or as the mean value if the heat flow curves considered.

Parameters:

Name	Type	Description	Default
individual	bool	DESCRIPTION. The default is False.	False
processparams		Dataclass containing parameters which control the processing of the calorimetry data.	required

Returns:

Type	Description
Pandas Dataframe

Examples:

Assuming that the calorimetry data is contained in a subfolder data, the time according to ASTM c1679 can be obtained by

>>> from CaloCem import tacalorimetry as ta
>>> from pathlib import Path
>>>
>>> thepath = Path(__file__).parent / "data"
>>> tam = ta.Measurement(thepath)
>>> astm = tam.get_astm_c1679_characteristics()

Source code in calocem/tacalorimetry.py

def get_astm_c1679_characteristics(
    self,
    processparams,
    individual: bool = False,
    show_plot=False,
    ax=None,
    regex=None,
    xscale="log",
    xunit="s",
) -> pd.DataFrame:
    """
    get characteristics according to ASTM C1679. Compiles a list of data
    points at half-maximum "normalized heat flow", wherein the half maximum
    is either determined for each individual heat flow curve individually
    or as the mean value if the heat flow curves considered.

    Parameters
    ----------
    individual : bool, optional
        DESCRIPTION. The default is False.
    processparams: ProcessingParameters
        Dataclass containing parameters which control the processing of the calorimetry data.

    Returns
    -------
    Pandas Dataframe

    Examples
    --------
    Assuming that the calorimetry data is contained in a subfolder `data`, the time according to ASTM c1679 can be obtained by

    >>> from CaloCem import tacalorimetry as ta
    >>> from pathlib import Path
    >>>
    >>> thepath = Path(__file__).parent / "data"
    >>> tam = ta.Measurement(thepath)
    >>> astm = tam.get_astm_c1679_characteristics()
    """

    # get peaks
    peaks = self.get_peaks(processparams, plt_right_s=4e5, show_plot=False)
    # sort peaks by ascending normalized heat flow
    peaks = peaks.sort_values(by="normalized_heat_flow_w_g", ascending=True)
    # select highest peak --> ASTM C1679
    peaks = peaks.groupby(by="sample").last()

    # get data
    data = self.get_data()

    # init empty list for collecting characteristics
    astm_times = []

    # loop samples
    for sample, sample_data in self._iter_samples(regex=regex):
        # pick sample data
        helper = data[data["sample"] == sample]
        helper_df = helper.copy()

        # check if peak was found
        if peaks[peaks["sample_short"] == sample_data.sample_short.iloc[0]].empty:
            helper = helper.iloc[0:1]
            # manually set time to NaN to indicate that no peak was found
            helper["time_s"] = np.nan

        else:
            # restrict to times before the peak
            helper = helper[helper["time_s"] <= peaks.at[sample, "time_s"]]

            # restrict to relevant heatflows the peak
            if individual == True:
                helper = helper[
                    helper["normalized_heat_flow_w_g"]
                    <= peaks.at[sample, "normalized_heat_flow_w_g"] * 0.50
                ]
            else:
                # use half-maximum average
                helper = helper[
                    helper["normalized_heat_flow_w_g"]
                    <= peaks["normalized_heat_flow_w_g"].mean() * 0.50
                ]

            # add to list of of selected points
        astm_times.append(helper.tail(1))

        if helper.tail(1)["time_s"].isna().all():
            continue

        if show_plot:
            # plot
            if xunit == "h":
                helper_df["time_s"] = helper_df["time_s"] / 3600
                helper["time_s"] = helper["time_s"] / 3600
            if isinstance(ax, matplotlib.axes._axes.Axes):
                ax.plot(
                    helper_df["time_s"],
                    helper_df["normalized_heat_flow_w_g"],
                    label=sample,
                )
                ax.plot(
                    helper.tail(1)["time_s"],
                    helper.tail(1)["normalized_heat_flow_w_g"],
                    marker="o",
                    color="red",
                )
                ax.vlines(
                    x=helper.tail(1)["time_s"],
                    ymin=0,
                    ymax=helper.tail(1)["normalized_heat_flow_w_g"],
                    color="red",
                    linestyle="--",
                )
                ax.text(
                    x=helper.tail(1)["time_s"],
                    y=helper.tail(1)["normalized_heat_flow_w_g"] / 2,
                    s=r" $t_{ASTM}$ ="
                    + f"{helper.tail(1)['time_s'].values[0]:.1f}",
                    color="red",
                )
            else:
                plt.plot(
                    data["time_s"],
                    data["normalized_heat_flow_w_g"],
                    label=sample,
                )

    # build overall DataFrame
    astm_times = pd.concat(astm_times)

    # return
    return astm_times

get_average_slope(processparams, target_col='normalized_heat_flow_w_g', age_col='time_s', regex=None, show_plot=False, ax=None, save_path=None, xscale='linear', xunit='s')

Calculate average slope by determining 4 additional slope values between onset time and heat flow maximum, in addition to the maximum slope.

Parameters:

Name	Type	Description	Default
processparams	ProcessingParameters	Processing parameters for analysis	required
target_col	str	Target measurement column, by default "normalized_heat_flow_w_g"	'normalized_heat_flow_w_g'
age_col	str	Time column name, by default "time_s"	'time_s'
regex	str	Regex pattern to filter samples, by default None	None
show_plot	bool	Whether to show plots, by default False	False
ax	Axes	Existing axis to plot on, by default None	None
save_path	Path	Path to save plots, by default None	None
xscale	str	X-axis scale, by default "log"	'linear'
xunit	str	Time unit for display, by default "s"	's'

Returns:

Type	Description
DataFrame	DataFrame containing average slope characteristics for each sample

Source code in calocem/tacalorimetry.py

def get_average_slope(
self,
processparams,
target_col="normalized_heat_flow_w_g",
age_col="time_s",
regex=None,
show_plot=False,
ax=None,
save_path=None,
xscale="linear",
xunit="s",
):
    """
    Calculate average slope by determining 4 additional slope values between 
    onset time and heat flow maximum, in addition to the maximum slope.

    Parameters
    ----------
    processparams : ProcessingParameters
        Processing parameters for analysis
    target_col : str, optional
        Target measurement column, by default "normalized_heat_flow_w_g"
    age_col : str, optional
        Time column name, by default "time_s"
    regex : str, optional
        Regex pattern to filter samples, by default None
    show_plot : bool, optional
        Whether to show plots, by default False
    ax : matplotlib.axes.Axes, optional
        Existing axis to plot on, by default None
    save_path : Path, optional
        Path to save plots, by default None
    xscale : str, optional
        X-axis scale, by default "log"
    xunit : str, optional
        Time unit for display, by default "s"

    Returns
    -------
    pd.DataFrame
        DataFrame containing average slope characteristics for each sample
    """

    # Get maximum slopes using existing method
    max_slopes = self.get_maximum_slope(
        processparams,
        target_col=target_col,
        age_col=age_col,
        regex=regex,
        show_plot=False,
        ax=ax,
        save_path=save_path,
        xscale=xscale,
        xunit=xunit,
    )

    if max_slopes is None or max_slopes.empty:
        print("No maximum slopes found. Cannot calculate average slopes.")
        return pd.DataFrame()

    # Get onset times using the peak onset method
    onsets = self.get_peak_onset_via_max_slope(
        processparams,
        show_plot=False,
        regex=regex,
        age_col=age_col,
        target_col=target_col,
        xunit=xunit,
    )

    if onsets.empty:
        print("No onset times found. Cannot calculate average slopes.")
        return pd.DataFrame()

    list_of_characteristics = []

    # Loop through samples
    for sample, data in self._iter_samples(regex=regex):
        sample_short = pathlib.Path(sample).stem

        # Get max slope data for this sample
        max_slope_row = max_slopes[max_slopes["sample_short"] == sample_short]
        if max_slope_row.empty:
            continue

        # Get onset data for this sample
        onset_row = onsets[onsets["sample"] == sample_short]
        if onset_row.empty:
            continue

        # Get time points
        onset_time = onset_row["onset_time_s"].iloc[0]
        max_slope_time = max_slope_row[age_col].iloc[0]

        # Find heat flow maximum after onset
        data_after_onset = data[data[age_col] >= onset_time]
        if data_after_onset.empty:
            continue

        max_hf_time = data_after_onset.loc[data_after_onset[target_col].idxmax(), age_col]

        # Create 4 intermediate time points between onset and heat flow maximum
        if max_hf_time <= onset_time:
            print(f"Warning: Heat flow maximum occurs before onset for {sample_short}")
            continue

        # Create 6 time points total (onset, 4 intermediate, max_hf)
        time_points = np.linspace(onset_time + 3600, max_hf_time - 3600, 6)

        # Calculate slopes at each interval
        slopes = []
        slope_times = []

        for i in range(len(time_points) - 1):
            t1, t2 = time_points[i], time_points[i + 1]

            # Get data points in this interval
            interval_data = data[(data[age_col] >= t1) & (data[age_col] <= t2)]

            if len(interval_data) < 2:
                continue

            # Calculate slope using linear regression
            x_vals = interval_data[age_col].values
            y_vals = interval_data[target_col].values

            # Simple slope calculation: (y2 - y1) / (x2 - x1)
            slope = (y_vals[-1] - y_vals[0]) / (x_vals[-1] - x_vals[0])
            slopes.append(slope)
            slope_times.append((t1 + t2) / 2)  # Midpoint time

        # Include the maximum slope
        max_slope_value = max_slope_row["gradient"].iloc[0]
        slopes.append(max_slope_value)
        slope_times.append(max_slope_time)

        # Calculate average slope
        if slopes:
            avg_slope = np.mean(slopes)
            std_slope = np.std(slopes)

            # Create characteristics dictionary
            characteristics = {
                "sample": sample,
                "sample_short": sample_short,
                "onset_time_s": onset_time,
                "max_hf_time_s": max_hf_time,
                "max_slope_time_s": max_slope_time,
                "max_slope_value": max_slope_value,
                "average_slope": avg_slope,
                "slope_std": std_slope,
                "n_slopes": len(slopes),
                "individual_slopes": slopes,
                "slope_times": slope_times,
            }

            # Optional plotting
            if show_plot:
                self._plot_average_slope_analysis(
                    data, characteristics, ax, age_col, target_col, 
                    sample_short, save_path, xscale, xunit
                )

            list_of_characteristics.append(characteristics)

    if not list_of_characteristics:
        print("No average slope characteristics calculated.")
        return pd.DataFrame()

    # Convert to DataFrame
    avg_slope_df = pd.DataFrame(list_of_characteristics)

    return avg_slope_df

get_cumulated_heat_at_hours(processparams=None, target_h=4, **kwargs)

get the cumulated heat flow a at a certain age

Parameters:

Name	Type	Description	Default
processparams	ProcessingParameters	Processing parameters. The default is None. If None, the default parameters are used. The most important parameter is the cutoff time in minutes which describes the initial time period of the measurement which is not considered for the cumulated heat flow. It is defined in the ProcessingParameters class. The default value is 30 minutes.	None
target_h	int | float	end time in hourscv	4

Returns:

Type	Description
A Pandas dataframe

Source code in calocem/tacalorimetry.py

def get_cumulated_heat_at_hours(self, processparams=None, target_h=4, **kwargs):
    """
    get the cumulated heat flow a at a certain age

    Parameters
    ----------
    processparams : ProcessingParameters, optional
        Processing parameters. The default is None. If None, the default
        parameters are used. The most important parameter is the cutoff time
        in minutes which describes the initial time period of the measurement
        which is not considered for the cumulated heat flow. It is defined in
        the ProcessingParameters class. The default value is 30 minutes.

    target_h : int | float
        end time in hourscv

    Returns
    -------
    A Pandas dataframe

    """
    if "cutoff_min" in kwargs:
        cutoff_min = kwargs["cutoff_min"]
        warnings.warn(
            "The cutoff_min parameter is deprecated. Please use the ProcessingParameters class instead.",
            DeprecationWarning,
            stacklevel=2,
        )
    else:
        if not processparams:
            processparams = ProcessingParameters()
        cutoff_min = processparams.cutoff.cutoff_min

    def applicable(df, target_h=4, cutoff_min=None):
        # convert target time to seconds
        target_s = 3600 * target_h
        # helper
        _helper = df.query("time_s <= @target_s").tail(1)
        # get heat at target time
        hf_at_target = float(_helper["normalized_heat_j_g"].values[0])

        # if cutoff time specified
        if cutoff_min:
            # convert target time to seconds
            target_s = 60 * cutoff_min
            try:
                # helper
                _helper = df.query("time_s <= @target_s").tail(1)
                # type conversion
                hf_at_cutoff = float(_helper["normalized_heat_j_g"].values[0])
                # correct heatflow for heatflow at cutoff
                hf_at_target = hf_at_target - hf_at_cutoff
            except TypeError:
                name_wt_nan = df["sample_short"].tolist()[0]
                print(
                    f"Found NaN in Normalized heat of sample {name_wt_nan} searching for cumulated heat at {target_h}h and a cutoff of {cutoff_min}min."
                )
                return np.nan

        # return
        return hf_at_target

    # in case of one specified time
    if isinstance(target_h, int) or isinstance(target_h, float):
        # groupby
        results = (
            self._data.groupby(by="sample")[["time_s", "normalized_heat_j_g"]]
            .apply(
                lambda x: applicable(x, target_h=target_h, cutoff_min=cutoff_min),
            )
            .reset_index(level=0)
        )
        # rename
        results.columns = ["sample", "cumulated_heat_at_hours"]
        results["target_h"] = target_h
        results["cutoff_min"] = cutoff_min

    # in case of specified list of times
    elif isinstance(target_h, list):
        # init list
        list_of_results = []
        # loop
        for this_target_h in target_h:
            # groupby
            _results = (
                self._data.groupby(by="sample")[["time_s", "normalized_heat_j_g"]]
                .apply(
                    lambda x: applicable(
                        x, target_h=this_target_h, cutoff_min=cutoff_min
                    ),
                )
                .reset_index(level=0)
            )
            # rename
            _results.columns = ["sample", "cumulated_heat_at_hours"]
            _results["target_h"] = this_target_h
            _results["cutoff_min"] = cutoff_min
            # append to list
            list_of_results.append(_results)
        # build overall results DataFrame
        results = pd.concat(list_of_results)

    # return
    return results

get_data()

A convenience function which returns the Pandas Dataframe containing the read and processed calorimetry data.

Returns:

Type	Description
Pandas DataFrame

Examples:

Assuming that the calorimetry data is contained in a subfolder data, a conventional Pandas dataframe df containing the data from all calorimetry files in data can be obtained with the following code.

>>> from CaloCem import tacalorimetry as ta
>>> from pathlib import Path
>>>
>>> thepath = Path(__file__).parent / "data"
>>> tam = ta.Measurement(thepath)
>>> df = tam.get_data()

Source code in calocem/tacalorimetry.py

def get_data(self):
    """
    A convenience function which returns the Pandas Dataframe containing the read and processed calorimetry data.
    Returns
    -------
    Pandas DataFrame

    Examples
    --------
    Assuming that the calorimetry data is contained in a subfolder `data`, a conventional Pandas dataframe `df` containing the data from all calorimetry files in `data` can be obtained with the following code.

    >>> from CaloCem import tacalorimetry as ta
    >>> from pathlib import Path
    >>>
    >>> thepath = Path(__file__).parent / "data"
    >>> tam = ta.Measurement(thepath)
    >>> df = tam.get_data()

    """

    return self._data

get_dormant_period_heatflow(processparams, regex=None, cutoff_min=5, upper_dormant_thresh_w_g=0.002, plot_right_boundary=200000.0, prominence=0.001, show_plot=False)

get dormant period heatflow

Parameters:

Name	Type	Description	Default
regex	str	Regex which can be used to filter the data, i.e., only the patterns which fit the regex will be evaluated. The default is None.	None
cutoff_min	int | float	Time at the start of the experiment which will be cutoff from analysis. This can be useful for ex-situ mixed samples. The default is 5.	5
upper_dormant_thresh_w_g	float	Parameter which controls the upper limit for the plotting option. The default is 0.001.	0.002
show_plot	bool	If set to true, the data is plotted. The default is False.	False

Returns:

Type	Description
Pandas Dataframe

Source code in calocem/tacalorimetry.py

def get_dormant_period_heatflow(
    self,
    processparams,
    regex: str = None,
    cutoff_min: int = 5,
    upper_dormant_thresh_w_g: float = 0.002,
    plot_right_boundary=2e5,
    prominence: float = 1e-3,
    show_plot=False,
) -> pd.DataFrame:
    """
    get dormant period heatflow

    Parameters
    ----------
    regex : str, optional
        Regex which can be used to filter the data, i.e., only the patterns which fit the regex will be evaluated. The default is None.
    cutoff_min : int | float, optional
        Time at the start of the experiment which will be cutoff from analysis. This can be useful for ex-situ mixed samples. The default is 5.
    upper_dormant_thresh_w_g : float, optional
        Parameter which controls the upper limit for the plotting option. The default is 0.001.
    show_plot : bool, optional
        If set to true, the data is plotted. The default is False.

    Returns
    -------
    Pandas Dataframe

    """

    # init results list
    list_dfs = []

    # loop samples
    for sample, data in self._iter_samples(regex=regex):
        # get peak as "right border"
        _peaks = self.get_peaks(
            processparams,
            # cutoff_min=cutoff_min,
            regex=pathlib.Path(sample).name,
            # prominence=processparams.gradient_peak_prominence, # prominence,
            show_plot=show_plot,
        )

        # identify "dormant period" as range between initial spike
        # and first reaction peak

        if show_plot:
            # plot
            plt.plot(
                data["time_s"],
                data["normalized_heat_flow_w_g"],
                # linestyle="",
                # marker="o",
            )

        # discard points at early age
        data = data.query("time_s >= @processparams.cutoff.cutoff_min * 60")
        if not _peaks.empty:
            # discard points after the first peak
            data = data.query('time_s <= @_peaks["time_s"].min()')

        # reset index
        data = data.reset_index(drop=True)

        # pick relevant points at minimum heat flow
        data = data.iloc[data["normalized_heat_flow_w_g"].idxmin(), :].to_frame().T

        if show_plot:
            # guide to the eye lines
            plt.axhline(float(data["normalized_heat_flow_w_g"]), color="red")
            plt.axvline(float(data["time_s"]), color="red")
            # indicate cutoff time
            plt.axvspan(0, cutoff_min * 60, color="black", alpha=0.5)
            # limits
            # plt.xlim(0, _peaks["time_s"].min())
            plt.xlim(0, plot_right_boundary)
            plt.ylim(0, upper_dormant_thresh_w_g)
            # title
            plt.title(pathlib.Path(sample).stem)
            # show
            plt.show()

        # add to list
        list_dfs.append(data)

    # convert to overall datafram
    result = pd.concat(list_dfs).reset_index(drop=True)

    # return
    return result

get_information()

get information

Returns:

Type	Description
DataFrame	information, i.e. date of measurement, operator, comment ...

Source code in calocem/tacalorimetry.py

def get_information(self):
    """
    get information

    Returns
    -------
    pd.DataFrame
        information, i.e. date of measurement, operator, comment ...

    """

    return self._info

get_maximum_slope(processparams, target_col='normalized_heat_flow_w_g', age_col='time_s', time_discarded_s=900, show_plot=False, show_info=True, exclude_discarded_time=False, regex=None, read_start_c3s=False, ax=None, save_path=None, xscale='log', xunit='s')

The method finds the point in time of the maximum slope. It also calculates the gradient at this point. The method can be controlled by passing a customized ProcessingParameters object for the processparams parameter. If no object is passed, the default parameters will be used.

Parameters:

Name	Type	Description	Default
target_col	str	measured quantity within which peak onsets are searched for. The default is "normalized_heat_flow_w_g"	'normalized_heat_flow_w_g'
age_col	str	Time unit within which peak onsets are searched for. The default is "time_s"	'time_s'
time_discarded_s	int | float	Time in seconds below which collected data points are discarded for peak onset picking. The default is 900.	900
show_plot	bool	Flag whether or not to plot peak picking for each sample. The default is False.	False
exclude_discarded_time	bool	Whether or not to discard the experimental values obtained before "time_discarded_s" also in the visualization. The default is False.	False
regex	str	regex pattern to include only certain experimental result files during initialization. The default is None.	None

Returns:

Type	Description
Pandas Dataframe	A dataframe that contains the time and the gradient of the maximum slope.

Examples:

>>> from CaloCem import tacalorimetry as ta
>>> from pathlib import Path

>>> thepath = Path(__file__).parent / "data"
>>> tam = ta.Measurement(thepath)
>>> processparams = ta.ProcessingParameters()
>>> processparams..apply = True
>>> max_slopes = tam.get_maximum_slope(processparams)

Source code in calocem/tacalorimetry.py

def get_maximum_slope(
    self,
    processparams,
    target_col="normalized_heat_flow_w_g",
    age_col="time_s",
    time_discarded_s=900,
    show_plot=False,
    show_info=True,
    exclude_discarded_time=False,
    regex=None,
    read_start_c3s=False,
    ax=None,
    save_path=None,
    xscale="log",
    xunit="s",
):
    """
    The method finds the point in time of the maximum slope. It also calculates the gradient at this point. The method can be controlled by passing a customized ProcessingParameters object for the `processparams` parameter. If no object is passed, the default parameters will be used.

    Parameters
    ----------
    target_col : str, optional
        measured quantity within which peak onsets are searched for. The default is "normalized_heat_flow_w_g"
    age_col : str, optional
        Time unit within which peak onsets are searched for. The default is "time_s"
    time_discarded_s : int | float, optional
        Time in seconds below which collected data points are discarded for peak onset picking. The default is 900.
    show_plot : bool, optional
        Flag whether or not to plot peak picking for each sample. The default is False.
    exclude_discarded_time : bool, optional
        Whether or not to discard the experimental values obtained before "time_discarded_s" also in the visualization. The default is False.
    regex : str, optional
        regex pattern to include only certain experimental result files during initialization. The default is None.
    Returns
    -------
    Pandas Dataframe
        A dataframe that contains the time and the gradient of the maximum slope.
    Examples
    --------
    >>> from CaloCem import tacalorimetry as ta
    >>> from pathlib import Path

    >>> thepath = Path(__file__).parent / "data"
    >>> tam = ta.Measurement(thepath)
    >>> processparams = ta.ProcessingParameters()
    >>> processparams..apply = True
    >>> max_slopes = tam.get_maximum_slope(processparams)
    """

    # init list of characteristics
    list_of_characteristics = []

    # loop samples
    for sample, data in self._iter_samples(regex=regex):
        sample_name = pathlib.Path(sample).stem
        if exclude_discarded_time:
            # exclude
            data = data.query(f"{age_col} >= {time_discarded_s}")

        # manual definition of start time to look for c3s - in case auto peak detection becomes difficult
        if read_start_c3s:
            c3s_start_time_s = self._metadata.query(
                f"sample_number == '{sample_name}'"
            )["t_c3s_min_s"].values[0]
            c3s_end_time_s = self._metadata.query(
                f"sample_number == '{sample_name}'"
            )["t_c3s_max_s"].values[0]
            data = data.query(
                f"{age_col} >= {c3s_start_time_s} & {age_col} <= {c3s_end_time_s}"
            )

        if show_info:
            print(f"Determineing maximum slope of {pathlib.Path(sample).stem}")

        processor = HeatFlowProcessor(processparams)

        data = utils.make_equidistant(data)

        if processparams.rolling_mean.apply:
            data = processor.apply_rolling_mean(data)

        data["gradient"], data["curvature"] = (
            processor.calculate_heatflow_derivatives(data)
        )

        characteristics = processor.get_largest_slope(data, processparams)
        if characteristics.empty:
            continue

        # optional plotting
        if show_plot:
            self._plot_maximum_slope(
                data,
                ax,
                age_col,
                target_col,
                sample,
                characteristics,
                time_discarded_s,
                save_path=save_path,
                xscale=xscale,
                xunit=xunit,
            )
            # plot heat flow curve
            # plt.plot(data[age_col], data[target_col], label=target_col)
            # plt.plot(
            #     data[age_col],
            #     data["gradient"] * 1e4 + 0.001,
            #     label="gradient * 1e4 + 1mW",
            # )

            # # add vertical lines
            # for _idx, _row in characteristics.iterrows():
            #     # vline
            #     plt.axvline(_row.at[age_col], color="green", alpha=0.3)

            # # cosmetics
            # plt.xscale("log")
            # plt.title(f"Maximum slope plot for {pathlib.Path(sample).stem}")
            # plt.xlabel(age_col)
            # plt.ylabel(target_col)
            # plt.legend()

            # # get axis
            # ax = plt.gca()

            # plt.fill_between(
            #     [ax.get_ylim()[0], time_discarded_s],
            #     [ax.get_ylim()[0]] * 2,
            #     [ax.get_ylim()[1]] * 2,
            #     color="black",
            #     alpha=0.35,
            # )

            # # set axis limit
            # plt.xlim(left=100)
            # plt.ylim(bottom=0, top=0.01)

            # # show
            # plt.show()

        # append to list
        list_of_characteristics.append(characteristics)

    if not list_of_characteristics:
        print("No maximum slope found, check you processing parameters")
    # build overall list
    else:
        max_slope_characteristics = pd.concat(list_of_characteristics)
        # return
        return max_slope_characteristics

get_metadata()

Returns

tuple pd.DataFrame of metadata and string of the column used as ID (has to be unique).

Source code in calocem/tacalorimetry.py

def get_metadata(self) -> tuple:
    """


     Returns
     -------
    tuple
         pd.DataFrame of metadata and string of the column used as ID (has to
         be unique).
    """

    # return
    return self._metadata, self._metadata_id

get_metadata_grouping_options()

get a list of categories to group by in in "self.plot_by_category"

Returns:

Type	Description
list	list of categories avaialble for grouping by.

Source code in calocem/tacalorimetry.py

def get_metadata_grouping_options(self) -> list:
    """
    get a list of categories to group by in in "self.plot_by_category"

    Returns
    -------
    list
        list of categories avaialble for grouping by.
    """

    # get list based on column names of "_metadata"
    return self._metadata.columns.tolist()

get_peak_onset_via_max_slope(processparams, show_plot=False, ax=None, regex=None, age_col='time_s', target_col='normalized_heat_flow_w_g', time_discarded_s=900, save_path=None, xscale='linear', xunit='s', intersection='dormant_hf')

get reaction onset based on tangent of maximum heat flow and heat flow during the dormant period. The characteristic time is inferred from the intersection of both characteristic lines

Parameters:

Name	Type	Description	Default
show_plot	TYPE	DESCRIPTION. The default is False.	False

Returns:

Type	Description
None.

Source code in calocem/tacalorimetry.py

def get_peak_onset_via_max_slope(
    self,
    processparams,
    show_plot=False,
    ax=None,
    regex=None,
    age_col="time_s",
    target_col="normalized_heat_flow_w_g",
    time_discarded_s=900,
    save_path=None,
    xscale="linear",
    xunit="s",
    intersection="dormant_hf",
):
    """
    get reaction onset based on tangent of maximum heat flow and heat flow
    during the dormant period. The characteristic time is inferred from
    the intersection of both characteristic lines

    Parameters
    ----------
    show_plot : TYPE, optional
        DESCRIPTION. The default is False.

    Returns
    -------
    None.

    """
    # get onsets
    max_slopes = self.get_maximum_slope(
        processparams,
        regex=regex,
        show_plot=False,
        ax=ax,
    )
    # % get dormant period HFs
    dorm_hfs = self.get_dormant_period_heatflow(
        processparams,  # cutoff_min=cutoff_min, prominence=prominence
        regex=regex,
        show_plot=False,
        # ax=ax,
    )

    # init list
    list_characteristics = []

    # loop samples
    for i, row in max_slopes.iterrows():
        # calculate y-offset
        t = row["normalized_heat_flow_w_g"] - row["time_s"] * row["gradient"]
        # calculate point of intersection
        # calculate x-intersect of tangent with dormant heat flow
        x_intersect_dormant = (
                float(
                    dorm_hfs[dorm_hfs["sample_short"] == row["sample_short"]][
                        "normalized_heat_flow_w_g"
                    ]
                )
                - t
            ) / row["gradient"]
        # elif intersection == "abscissa":
            # calculate x-intersect of tangent with abscissa (y=0)
        x_intersect = row["time_s"] - (row["normalized_heat_flow_w_g"] / row["gradient"])

        data = self._data.query("sample_short == @row['sample_short']")
        sample = row["sample_short"]

        heat_at_intersect = np.interp(x_intersect, data["time_s"], data["normalized_heat_j_g"])

        # append to list
        list_characteristics.append(
            {
                "sample": row["sample_short"],
                "onset_time_s_abscissa": x_intersect,
                "onset_time_min_abscissa": x_intersect / 60,
                "heat_at_onset_j_g": heat_at_intersect,
                "onset_time_s": x_intersect_dormant,
                "onset_time_min": x_intersect_dormant / 60,
            }
        )


        dorm_hfs_sample = dorm_hfs.query("sample_short == @sample")
        # add prefix dorm to all columns
        dorm_hfs_sample.columns = ["dorm_" + s for s in dorm_hfs_sample.columns]

        characteristics = pd.concat([row, dorm_hfs_sample.squeeze()])
        characteristics.loc["xunit"] = xunit
        characteristics.loc["x_intersect"] = x_intersect
        characteristics.loc["intersection"] = intersection
        # print(characteristics.x_intersect)

        # get maximum time value
        tmax = self._data.query("sample_short == @row['sample_short']")[
            "time_s"
        ].max()
        # get maximum heat flow value
        hmax = self._data.query(
            "time_s > 3000 & sample_short == @row['sample_short']"
        )["normalized_heat_flow_w_g"].max()

        if show_plot:
            self._plot_intersection(
                data,
                ax,
                age_col,
                target_col,
                sample,
                # max_slopes,
                time_discarded_s,
                characteristics=characteristics,
                save_path=save_path,
                xscale=xscale,
                # xunit=xunit,
                hmax=hmax,
                tmax=tmax,
            )

    # build overall dataframe to be returned
    onsets = pd.DataFrame(list_characteristics)

    # merge with dorm_hfs
    onsets = onsets.merge(
        dorm_hfs[
            ["sample_short", "normalized_heat_flow_w_g", "normalized_heat_j_g"]
        ],
        left_on="sample",
        right_on="sample_short",
        how="left",
    )

    # rename
    onsets = onsets.rename(
        columns={
            "normalized_heat_flow_w_g": "normalized_heat_flow_w_g_at_dorm_min",
            "normalized_heat_j_g": "normalized_heat_j_g_at_dorm_min",
        }
    )

    # return
    return onsets

get_peak_onsets(target_col='normalized_heat_flow_w_g', age_col='time_s', time_discarded_s=900, rolling=1, gradient_threshold=0.0005, show_plot=False, exclude_discarded_time=False, regex=None, ax=None)

get peak onsets based on a criterion of minimum gradient

Parameters:

Name	Type	Description	Default
target_col	str	measured quantity within which peak onsets are searched for. The default is "normalized_heat_flow_w_g"	'normalized_heat_flow_w_g'
age_col	str	Time unit within which peak onsets are searched for. The default is "time_s"	'time_s'
time_discarded_s	int | float	Time in seconds below which collected data points are discarded for peak onset picking. The default is 900.	900
rolling	int	Width of "rolling" window within which the values of "target_col" are averaged. A higher value will introduce a stronger smoothing effect. The default is 1, i.e. no smoothing.	1
gradient_threshold	float	Threshold of slope for identification of a peak onset. For a lower value, earlier peak onsets will be identified. The default is 0.0005.	0.0005
show_plot	bool	Flag whether or not to plot peak picking for each sample. The default is False.	False
exclude_discarded_time	bool	Whether or not to discard the experimental values obtained before "time_discarded_s" also in the visualization. The default is False.	False
regex	str	regex pattern to include only certain experimental result files during initialization. The default is None.	None
ax	Axes | None	The default is None.	None

Returns:

Type	Description
pd.DataFrame holding peak onset characterisitcs for each sample.

Source code in calocem/tacalorimetry.py

def get_peak_onsets(
    self,
    target_col="normalized_heat_flow_w_g",
    age_col="time_s",
    time_discarded_s=900,
    rolling=1,
    gradient_threshold=0.0005,
    show_plot=False,
    exclude_discarded_time=False,
    regex=None,
    ax: plt.Axes = None,
):
    """
    get peak onsets based on a criterion of minimum gradient

    Parameters
    ----------
    target_col : str, optional
        measured quantity within which peak onsets are searched for. The default is "normalized_heat_flow_w_g"
    age_col : str, optional
        Time unit within which peak onsets are searched for. The default is "time_s"
    time_discarded_s : int | float, optional
        Time in seconds below which collected data points are discarded for peak onset picking. The default is 900.
    rolling : int, optional
        Width of "rolling" window within which the values of "target_col" are averaged. A higher value will introduce a stronger smoothing effect. The default is 1, i.e. no smoothing.
    gradient_threshold : float, optional
        Threshold of slope for identification of a peak onset. For a lower value, earlier peak onsets will be identified. The default is 0.0005.
    show_plot : bool, optional
        Flag whether or not to plot peak picking for each sample. The default is False.
    exclude_discarded_time : bool, optional
        Whether or not to discard the experimental values obtained before "time_discarded_s" also in the visualization. The default is False.
    regex : str, optional
        regex pattern to include only certain experimental result files during initialization. The default is None.
    ax : matplotlib.axes._axes.Axes | None, optional
        The default is None.
    Returns
    -------
    pd.DataFrame holding peak onset characterisitcs for each sample.

    """

    # init list of characteristics
    list_of_characteristics = []

    # loop samples
    for sample, data in self._iter_samples(regex=regex):
        if exclude_discarded_time:
            # exclude
            data = data.query(f"{age_col} >= {time_discarded_s}")

        # reset index
        data = data.reset_index(drop=True)

        # calculate get gradient
        data["gradient"] = pd.Series(
            np.gradient(data[target_col].rolling(rolling).mean(), data[age_col])
        )

        # get relevant points
        characteristics = data.copy()
        # discard initial time
        characteristics = characteristics.query(f"{age_col} >= {time_discarded_s}")
        # look at values with certain gradient only
        characteristics = characteristics.query("gradient > @gradient_threshold")
        # consider first entry exclusively
        characteristics = characteristics.head(1)

        # optional plotting
        if show_plot:
            # if specific axis to plot to is specified
            if isinstance(ax, matplotlib.axes._axes.Axes):
                # plot heat flow curve
                p = ax.plot(data[age_col], data[target_col])

                # add vertical lines
                for _idx, _row in characteristics.iterrows():
                    # vline
                    ax.axvline(_row.at[age_col], color=p[0].get_color(), alpha=0.3)
                    # add "slope line"
                    ax.axline(
                        (_row.at[age_col], _row.at[target_col]),
                        slope=_row.at["gradient"],
                        color=p[0].get_color(),
                        # color="k",
                        # linewidth=0.2
                        alpha=0.25,
                        linestyle="--",
                    )

                # cosmetics
                # ax.set_xscale("log")
                ax.set_title("Onset for " + pathlib.Path(sample).stem)
                ax.set_xlabel(age_col)
                ax.set_ylabel(target_col)

                ax.fill_between(
                    [ax.get_ylim()[0], time_discarded_s],
                    [ax.get_ylim()[0]] * 2,
                    [ax.get_ylim()[1]] * 2,
                    color="black",
                    alpha=0.35,
                )

                # set axis limit
                ax.set_xlim(left=100)

            else:
                # plot heat flow curve
                plt.plot(data[age_col], data[target_col])

                # add vertical lines
                for _idx, _row in characteristics.iterrows():
                    # vline
                    plt.axvline(_row.at[age_col], color="red", alpha=0.3)

                # cosmetics
                # plt.xscale("log")
                plt.title("Onset for " + pathlib.Path(sample).stem)
                plt.xlabel(age_col)
                plt.ylabel(target_col)

                # get axis
                ax = plt.gca()

                plt.fill_between(
                    [ax.get_ylim()[0], time_discarded_s],
                    [ax.get_ylim()[0]] * 2,
                    [ax.get_ylim()[1]] * 2,
                    color="black",
                    alpha=0.35,
                )

                # set axis limit
                plt.xlim(left=100)

        # append to list
        list_of_characteristics.append(characteristics)

    # build overall list
    onset_characteristics = pd.concat(list_of_characteristics)

    # return
    if isinstance(ax, matplotlib.axes._axes.Axes):
        # return onset characteristics and ax
        return onset_characteristics, ax
    else:
        # return onset characteristics exclusively
        return onset_characteristics

get_peaks(processparams, target_col='normalized_heat_flow_w_g', regex=None, cutoff_min=None, show_plot=True, plt_right_s=200000.0, plt_top=0.01, ax=None, xunit='s', plot_labels=None, xmarker=False)

get DataFrame of peak characteristics.

Parameters:

Name	Type	Description	Default
target_col	str	measured quantity within which peaks are searched for. The default is "normalized_heat_flow_w_g"	'normalized_heat_flow_w_g'
regex	str	regex pattern to include only certain experimental result files during initialization. The default is None.	None
cutoff_min	int | float	Time in minutes below which collected data points are discarded for peak picking	None
show_plot	bool	Flag whether or not to plot peak picking for each sample. The default is True.	True
plt_right_s	int | float	Upper limit of x-axis of in seconds. The default is 2e5.	200000.0
plt_top	int | float	Upper limit of y-axis of. The default is 1e-2.	0.01
ax	Axes | None	The default is None.	None

Returns:

Type	Description
pd.DataFrame holding peak characterisitcs for each sample.

Source code in calocem/tacalorimetry.py

def get_peaks(
    self,
    processparams,
    target_col="normalized_heat_flow_w_g",
    regex=None,
    cutoff_min=None,
    show_plot=True,
    plt_right_s=2e5,
    plt_top=1e-2,
    ax=None,
    xunit="s",
    plot_labels=None,
    xmarker=False,
) -> pd.DataFrame:
    """
    get DataFrame of peak characteristics.

    Parameters
    ----------
    target_col : str, optional
        measured quantity within which peaks are searched for. The default is "normalized_heat_flow_w_g"
    regex : str, optional
        regex pattern to include only certain experimental result files
        during initialization. The default is None.
    cutoff_min : int | float, optional
        Time in minutes below which collected data points are discarded for peak picking
    show_plot : bool, optional
        Flag whether or not to plot peak picking for each sample. The default is True.
    plt_right_s : int | float, optional
        Upper limit of x-axis of in seconds. The default is 2e5.
    plt_top : int | float, optional
        Upper limit of y-axis of. The default is 1e-2.
    ax : matplotlib.axes._axes.Axes | None, optional
        The default is None.

    Returns
    -------
    pd.DataFrame holding peak characterisitcs for each sample.

    """

    # list of peaks
    list_of_peaks_dfs = []

    # loop samples
    for sample, data in self._iter_samples(regex=regex):
        # cutoff
        if processparams.cutoff.cutoff_min:
            # discard points at early age
            data = data.query("time_s >= @processparams.cutoff.cutoff_min * 60")

        # reset index
        data = data.reset_index(drop=True)

        # target_columns
        _age_col = "time_s"
        _target_col = target_col

        # find peaks
        peaks, properties = signal.find_peaks(
            data[_target_col],
            prominence=processparams.peakdetection.prominence,
            distance=processparams.peakdetection.distance,
        )

        # plot?
        if show_plot:
            if xunit == "h":
                df_copy = data.copy()
                df_copy[_age_col] = df_copy[_age_col] / 3600
                plt_right_s = plt_right_s / 3600
                self._plot_peak_positions(
                    df_copy,
                    ax,
                    _age_col,
                    _target_col,
                    peaks,
                    sample,
                    plt_top,
                    plt_right_s,
                    plot_labels,
                    xmarker,
                )
            else:
                self._plot_peak_positions(
                    data,
                    ax,
                    _age_col,
                    _target_col,
                    peaks,
                    sample,
                    plt_top,
                    plt_right_s,
                    plot_labels,
                    xmarker,
                )

        # compile peak characteristics
        peak_characteristics = pd.concat(
            [
                data.iloc[peaks, :],
                pd.DataFrame(
                    properties["prominences"], index=peaks, columns=["prominence"]
                ),
                pd.DataFrame({"peak_nr": np.arange((len(peaks)))}, index=peaks),
            ],
            axis=1,
        )

        # append
        list_of_peaks_dfs.append(peak_characteristics)

    # compile peak information
    peaks = pd.concat(list_of_peaks_dfs)

    if isinstance(ax, matplotlib.axes._axes.Axes):
        # return peak list and ax
        return peaks, ax
    else:  # return peak list only
        return peaks

get_sample_names()

get list of sample names

Returns:

Type	Description
None.

Source code in calocem/tacalorimetry.py

def get_sample_names(self):
    """
    get list of sample names

    Returns
    -------
    None.

    """

    # get list
    samples = [pathlib.Path(s).stem for s, _ in self._iter_samples()]

    # return
    return samples

normalize_sample_to_mass(sample_short, mass_g, show_info=True)

normalize "heat_flow" to a certain mass

Parameters:

Name	Type	Description	Default
sample_short	str	"sample_short" name of sample to be normalized.	required
mass_g	float	mass in gram to which "heat_flow_w" are normalized.	required

Returns:

Type	Description
None.

Source code in calocem/tacalorimetry.py

def normalize_sample_to_mass(
    self, sample_short: str, mass_g: float, show_info=True
):
    """
    normalize "heat_flow" to a certain mass

    Parameters
    ----------
    sample_short : str
        "sample_short" name of sample to be normalized.
    mass_g : float
        mass in gram to which "heat_flow_w" are normalized.

    Returns
    -------
    None.

    """

    # normalize "heat_flow_w" to sample mass
    self._data.loc[
        self._data["sample_short"] == sample_short, "normalized_heat_flow_w_g"
    ] = (
        self._data.loc[self._data["sample_short"] == sample_short, "heat_flow_w"]
        / mass_g
    )

    # normalize "heat_j" to sample mass
    try:
        self._data.loc[
            self._data["sample_short"] == sample_short, "normalized_heat_j_g"
        ] = (
            self._data.loc[self._data["sample_short"] == sample_short, "heat_j"]
            / mass_g
        )
    except Exception:
        pass

    # info
    if show_info:
        print(f"Sample {sample_short} normalized to {mass_g}g sample.")

plot(t_unit='h', y='normalized_heat_flow_w_g', y_unit_milli=True, regex=None, show_info=True, ax=None)

Plot the calorimetry data.

Parameters:

Name	Type	Description	Default
t_unit	str	time unit. The default is "h". Options are "s", "min", "h", "d".	'h'
y	str	y-axis. The default is "normalized_heat_flow_w_g". Options are "normalized_heat_flow_w_g", "heat_flow_w", "normalized_heat_j_g", "heat_j".	'normalized_heat_flow_w_g'
y_unit_milli	bool	whether or not to plot y-axis in Milliwatt. The default is True.	True
regex	str	regex pattern to include only certain samples during plotting. The default is None.	None
show_info	bool	whether or not to show information. The default is True.	True
ax	Axes	axis to plot to. The default is None.	None

Examples:

>>> import CaloCem as ta
>>> from pathlib import Path
>>>
>>> calodatapath = Path(__file__).parent
>>> tam = ta.Measurement(folder=calodatapath, show_info=True)
>>> tam.plot(t_unit="h", y="normalized_heat_flow_w_g", y_unit_milli=False)

Source code in calocem/tacalorimetry.py

def plot(
    self,
    t_unit="h",
    y="normalized_heat_flow_w_g",
    y_unit_milli=True,
    regex=None,
    show_info=True,
    ax=None,
):
    """

    Plot the calorimetry data.

    Parameters
    ----------
    t_unit : str, optional
        time unit. The default is "h". Options are "s", "min", "h", "d".
    y : str, optional
        y-axis. The default is "normalized_heat_flow_w_g". Options are
        "normalized_heat_flow_w_g", "heat_flow_w", "normalized_heat_j_g",
        "heat_j".
    y_unit_milli : bool, optional
        whether or not to plot y-axis in Milliwatt. The default is True.
    regex : str, optional
        regex pattern to include only certain samples during plotting. The
        default is None.
    show_info : bool, optional
        whether or not to show information. The default is True.
    ax : matplotlib.axes._axes.Axes, optional
        axis to plot to. The default is None.

    Examples
    --------
    >>> import CaloCem as ta
    >>> from pathlib import Path
    >>>
    >>> calodatapath = Path(__file__).parent
    >>> tam = ta.Measurement(folder=calodatapath, show_info=True)
    >>> tam.plot(t_unit="h", y="normalized_heat_flow_w_g", y_unit_milli=False)

    """

    # y-value
    if y == "normalized_heat_flow_w_g":
        y_column = "normalized_heat_flow_w_g"
        y_label = "Normalized Heat Flow / [W/g]"
    elif y == "heat_flow_w":
        y_column = "heat_flow_w"
        y_label = "Heat Flow / [W]"
    elif y == "normalized_heat_j_g":
        y_column = "normalized_heat_j_g"
        y_label = "Normalized Heat / [J/g]"
    elif y == "heat_j":
        y_column = "heat_j"
        y_label = "Heat / [J]"

    if y_unit_milli:
        y_label = y_label.replace("[", "[m")

    # x-unit
    if t_unit == "s":
        x_factor = 1.0
    elif t_unit == "min":
        x_factor = 1 / 60
    elif t_unit == "h":
        x_factor = 1 / (60 * 60)
    elif t_unit == "d":
        x_factor = 1 / (60 * 60 * 24)

    # y-unit
    if y_unit_milli:
        y_factor = 1000
    else:
        y_factor = 1

    for sample, data in self._iter_samples():
        if regex:
            if not re.findall(rf"{regex}", os.path.basename(sample)):
                continue
        data["time_s"] = data["time_s"] * x_factor
        # all columns containing heat
        heatcols = [s for s in data.columns if "heat" in s]
        data[heatcols] = data[heatcols] * y_factor
        ax, _ = utils.create_base_plot(data, ax, "time_s", y_column, sample)
        ax = utils.style_base_plot(
            ax,
            y_label,
            t_unit,
            sample,
        )
    return ax

plot_by_category(categories, t_unit='h', y='normalized_heat_flow_w_g', y_unit_milli=True)

plot by category, wherein the category is based on the information passed via "self._add_metadata_source". Options available as "category" are accessible via "self.get_metadata_grouping_options"

Parameters:

Name	Type	Description	Default
categories	(str, list[str])	category (from "self.get_metadata_grouping_options") to group by. specify a string or a list of strings here	required
t_unit	TYPE	see "self.plot". The default is "h".	'h'
y	TYPE	see "self.plot". The default is "normalized_heat_flow_w_g".	'normalized_heat_flow_w_g'
y_unit_milli	TYPE	see "self.plot". The default is True.	True

Examples:

>>> import CaloCem as ta
>>> from pathlib import Path
>>>
>>> calodatapath = Path(__file__).parent
>>> tam = ta.Measurement(folder=calodatapath, show_info=True)
>>> tam.plot_by_category(categories="sample")

Returns:

Type	Description
None.

Source code in calocem/tacalorimetry.py

def plot_by_category(
    self, categories, t_unit="h", y="normalized_heat_flow_w_g", y_unit_milli=True
):
    """
    plot by category, wherein the category is based on the information passed
    via "self._add_metadata_source". Options available as "category" are
    accessible via "self.get_metadata_grouping_options"

    Parameters
    ----------
    categories : str, list[str]
        category (from "self.get_metadata_grouping_options") to group by.
        specify a string or a list of strings here
    t_unit : TYPE, optional
        see "self.plot". The default is "h".
    y : TYPE, optional
        see "self.plot". The default is "normalized_heat_flow_w_g".
    y_unit_milli : TYPE, optional
        see "self.plot". The default is True.

    Examples
    --------
    >>> import CaloCem as ta
    >>> from pathlib import Path
    >>>
    >>> calodatapath = Path(__file__).parent
    >>> tam = ta.Measurement(folder=calodatapath, show_info=True)
    >>> tam.plot_by_category(categories="sample")


    Returns
    -------
    None.

    """

    def build_helper_string(values: list) -> str:
        """
        build a "nicely" formatted string from a supplied list
        """

        if len(values) == 2:
            # connect with "and"
            formatted = " and ".join([str(i) for i in values])
        elif len(values) > 2:
            # connect with comma and "and" for last element
            formatted = (
                ", ".join([str(i) for i in values[:-1]]) + " and " + str(values[-1])
            )
        else:
            formatted = "---"

        # return
        return formatted

    # loop category values
    for selections, _ in self._metadata.groupby(by=categories):
        if isinstance(selections, tuple):
            # - if multiple categories to group by are specified -
            # init helper DataFrame
            target_idx = pd.DataFrame()
            # identify corresponding samples
            for selection, category in zip(selections, categories):
                target_idx[category] = self._metadata[category] == selection
            # get relevant indices
            target_idx = target_idx.sum(axis=1) == len(categories)
            # define title
            title = f"Grouped by {build_helper_string(categories)} ({build_helper_string(selections)})"
        else:
            # - if only one(!) category to group by is specified -
            # identify corresponding samples
            target_idx = self._metadata[categories] == selections
            # define title
            title = f"Grouped by {categories} ({selections})"

        # pick relevant samples
        target_samples = self._metadata.loc[target_idx, self._metadata_id]

        # build corresponding regex
        regex = "(" + ")|(".join(target_samples) + ")"

        # plot
        ax = self.plot(regex=regex, t_unit=t_unit, y=y, y_unit_milli=y_unit_milli)

        # set title
        ax.set_title(title)

        # yield latest plot
        yield selections, ax

remove_pickle_files()

remove pickle files if re-reading of source files needed

Returns:

Type	Description
None.

Source code in calocem/tacalorimetry.py

def remove_pickle_files(self):
    """
    remove pickle files if re-reading of source files needed

    Returns
    -------
    None.

    """

    # remove files
    for file in [self._file_data_pickle, self._file_info_pickle]:
        # remove file
        pathlib.Path(file).unlink()

undo_average_by_metadata()

undo action of "average_by_metadata"

Source code in calocem/tacalorimetry.py

def undo_average_by_metadata(self):
    """
    undo action of "average_by_metadata"
    """

    # set "unprocessed" data as exeperimental data / "de-average"
    if not self._data_unprocessed.empty:
        # reset
        self._data = self._data_unprocessed.copy()

undo_tian_correction()

undo_tian_correction; i.e. restore original data

Returns:

Type	Description
None.

Source code in calocem/tacalorimetry.py

def undo_tian_correction(self):
    """
    undo_tian_correction; i.e. restore original data


    Returns
    -------
    None.

    """

    # call original restore function
    self.undo_average_by_metadata()

MedianFilterParameters dataclass

Parameters for the application of a Median filter to the data. The SciPy method median_filter is applied. Link to method

Parameters:

Name	Type	Description	Default
apply	bool	default is false. If True, a median filter is applied. The Scipy function median_filter is applied.	False
size	int	The size of the median filter (see the SciPy documentation)	7

Source code in calocem/processparams.py

@dataclass
class MedianFilterParameters:
    """Parameters for the application of a Median filter to the data. The SciPy method `median_filter` is applied. [Link to method](https://docs.scipy.org/doc/scipy/reference/generated/scipy.ndimage.median_filter.html)


    Parameters
    ----------
    apply: bool
        default is false. If `True`, a median filter is applied. The Scipy function `median_filter` is  applied.
    size: int
        The size of the median filter (see the SciPy documentation)

    """

    apply: bool = False
    size: int = 7

PeakDetectionParameters dataclass

Parameters that control the identication of peaks during peak detection.

Parameters:

Name	Type	Description	Default
prominence	float	The minimum prominence of the peak	1e-05
distance	int	The minimum distance of the peak.	100

Source code in calocem/processparams.py

@dataclass
class PeakDetectionParameters:
    """
    Parameters that control the identication of peaks during peak detection.

    Parameters
    ----------

    prominence: float
        The minimum prominence of the peak
    distance: int
        The minimum distance of the peak.
    """

    prominence: float = 1e-5
    distance: int = 100

PreProcessParameters dataclass

Parameters for preprocessing the data before analysis.

Attributes:

Name	Type	Description
infer_heat	bool	If True, the heat flow is inferred from the data. This is useful for data that does not have a heat flow column.

Source code in calocem/processparams.py

@dataclass
class PreProcessParameters:
    """
    Parameters for preprocessing the data before analysis.

    Attributes
    ----------
    infer_heat: bool
        If True, the heat flow is inferred from the data. This is useful for data that does not have a heat flow column.
    """

    infer_heat: bool = False

ProcessingParameters dataclass

A data class for storing all processing parameters for calorimetry data.

This class aggregates various processing parameters, including cutoff criteria, time constants for the Tian correction, and parameters for peak detection and gradient peak detection.

Attributes:

Name	Type	Description
cutoff	CutOffParameters	Parameters defining the cutoff criteria for the analysis. Currently only cutoff_min is implemented, which defines the minimum time in minutes for the analysis. The default value is defined in the CutOffParameters class.
time_constants	TianCorrectionParameters	Parameters related to time constants used in Tian's correction method for thermal analysis. he default values are defined in the TianCorrectionParameters class.
peakdetection	PeakDetectionParameters	Parameters for detecting peaks in the thermal analysis data. This includes settings such as the minimum prominence and distance between peaks. The default values are defined in the PeakDetectionParameters class.
gradient_peakdetection	GradientPeakDetectionParameters	Parameters for detecting peaks based on the gradient of the thermal analysis data. This includes more nuanced settings such as prominence, distance, width, relative height, and the criteria for selecting peaks (e.g., first peak, largest width). The default values are defined in the GradientPeakDetectionParameters class.
downsample	DownSamplingParameters	Parameters for adaptive downsampling of the thermal analysis data. This includes settings such as the number of points, smoothing factor, and baseline weight. The default values are defined in the DownSamplingParameters class.
spline_interpolation	SplineInterpolationParameters	Parameters which control the interpolation of the first and second derivative of the data. If no smoothing is applied the derivatives often become very noisy.
slope_analysis	SlopeAnalysisParameters	Parameters for slope analysis of the heat flow data. This includes settings which control the identification of the mean slope of the main hydration peak.

Examples:

Define a set of processing parameters for thermal analysis data.

>>> processparams = ProcessingParameters()
>>> processparams.cutoff.cutoff_min = 30
>>> processparams.spline_interpolation.apply = True

Source code in calocem/processparams.py

@dataclass
class ProcessingParameters:
    """
    A data class for storing all processing parameters for calorimetry data.

    This class aggregates various processing parameters, including cutoff criteria, time constants for the Tian correction, and parameters for peak detection and gradient peak detection.

    Attributes
    ----------

    cutoff :
        Parameters defining the cutoff criteria for the analysis.
        Currently only cutoff_min is implemented, which defines the minimum time in minutes for the analysis. The default value is defined in the CutOffParameters class.

    time_constants : TianCorrectionParameters
        Parameters related to time constants used in Tian's correction method for thermal analysis. he default values are defined in the
        TianCorrectionParameters class.

    peakdetection : PeakDetectionParameters
        Parameters for detecting peaks in the thermal analysis data. This includes settings such as the minimum
        prominence and distance between peaks. The default values are defined in the PeakDetectionParameters class.

    gradient_peakdetection : GradientPeakDetectionParameters
        Parameters for detecting peaks based on the gradient of the thermal analysis data. This includes more
        nuanced settings such as prominence, distance, width, relative height, and the criteria for selecting peaks
        (e.g., first peak, largest width). The default values are defined in the GradientPeakDetectionParameters class.

    downsample : DownSamplingParameters
        Parameters for adaptive downsampling of the thermal analysis data. This includes settings such as the number of points,
        smoothing factor, and baseline weight. The default values are defined in the DownSamplingParameters class.

    spline_interpolation: SplineInterpolationParameters
        Parameters which control the interpolation of the first and second derivative of the data. If no smoothing is applied the derivatives often become very noisy.

    slope_analysis: SlopeAnalysisParameters
        Parameters for slope analysis of the heat flow data. This includes settings which control the identification of the mean slope of the main hydration peak.


    Examples
    --------

    Define a set of processing parameters for thermal analysis data.

    >>> processparams = ProcessingParameters()
    >>> processparams.cutoff.cutoff_min = 30
    >>> processparams.spline_interpolation.apply = True
    """

    cutoff: CutOffParameters = field(default_factory=CutOffParameters)
    time_constants: TianCorrectionParameters = field(
        default_factory=TianCorrectionParameters
    )

    # peak detection params
    peakdetection: PeakDetectionParameters = field(
        default_factory=PeakDetectionParameters
    )
    gradient_peakdetection: GradientPeakDetectionParameters = field(
        default_factory=GradientPeakDetectionParameters
    )

    # smoothing params
    rolling_mean: RollingMeanParameters = field(default_factory=RollingMeanParameters)
    median_filter: MedianFilterParameters = field(
        default_factory=MedianFilterParameters
    )
    nonlin_savgol: NonLinSavGolParameters = field(
        default_factory=NonLinSavGolParameters
    )
    spline_interpolation: SplineInterpolationParameters = field(
        default_factory=SplineInterpolationParameters
    )
    # preprocessing params
    downsample: DownSamplingParameters = field(default_factory=DownSamplingParameters)
    preprocess: PreProcessParameters = field(default_factory=PreProcessParameters)
    # slope analysis params
    slope_analysis: SlopeAnalysisParameters = field(
        default_factory=SlopeAnalysisParameters
    )

SlopeAnalysisParameters dataclass

Parameters for slope analysis of heat flow data.

Attributes:

Name	Type	Description
flank_fraction_start	float	The start fraction of the window for averaging the slope of the main hydration peak. Example: 0.35 (the default value) means that the slope is calculated starting from 35% of the peak height measured relative to the minimum of the dormant period heat flow.
flank_fraction_end	float	The end fraction of the window for averaging the slope of the main hydration peak. Example: 0.55 (the default value) means that the slope is calculated up to 55% of the peak height measured relative to the minimum of the dormant period heat flow.
window_size	float	The size of the window for averaging the slope, given as a fraction of the total number of data points. Example: 0.1 (the default value) means that the slope is averaged over a window that is 10% of the total number of data points.

Source code in calocem/processparams.py

@dataclass
class SlopeAnalysisParameters:
    """
    Parameters for slope analysis of heat flow data.

    Attributes
    ----------
    flank_fraction_start: float
        The start fraction of the window for averaging the slope of the main hydration peak. Example: 0.35 (the default value) means that the slope is calculated starting from 35% of the peak height measured relative to the minimum of the dormant period heat flow.
    flank_fraction_end: float
        The end fraction of the window for averaging the slope of the main hydration peak. Example: 0.55 (the default value) means that the slope is calculated up to 55% of the peak height measured relative to the minimum of the dormant period heat flow.
    window_size: float
        The size of the window for averaging the slope, given as a fraction of the total number of data points. Example: 0.1 (the default value) means that the slope is averaged over a window that is 10% of the total number of data points.
    """

    flank_fraction_start: float = 0.35
    flank_fraction_end: float = 0.55
    window_size: float = 0.1  # as fraction of total data points

SplineInterpolationParameters dataclass

Parameters for spline interpolation of heat flow data.

Parameters:

Name	Type	Description	Default
apply	bool	Flag indicating whether spline interpolation should be applied to the heat flow data. The default value is False.	False
smoothing_1st_deriv	float	Smoothing parameter for the first derivative of the heat flow data. The default value is 1e-9.	1e-09
smoothing_2nd_deriv	float	Smoothing parameter for the second derivative of the heat flow data. The default value is 1e-9.	1e-09

Source code in calocem/processparams.py

@dataclass
class SplineInterpolationParameters:
    """Parameters for spline interpolation of heat flow data.

    Parameters
    ----------

    apply :
        Flag indicating whether spline interpolation should be applied to the heat flow data. The default value is False.

    smoothing_1st_deriv :
        Smoothing parameter for the first derivative of the heat flow data. The default value is 1e-9.

    smoothing_2nd_deriv :
        Smoothing parameter for the second derivative of the heat flow data. The default value is 1e-9.

    """

    apply: bool = False
    smoothing_1st_deriv: float = 1e-9
    smoothing_2nd_deriv: float = 1e-9

TianCorrectionParameters dataclass

Parameters related to time constants used in Tian's correction method for thermal analysis. The default values are defined in the TianCorrectionParameters class.

Parameters:

Name	Type	Description	Default
tau1	int	Time constant for the first correction step in Tian's method. The default value is 300.	300
tau2	int	Time constant for the second correction step in Tian's method. The default value is 100.	100

Source code in calocem/processparams.py

@dataclass
class TianCorrectionParameters:
    """
    Parameters related to time constants used in Tian's correction method for thermal analysis. The default values are defined in the TianCorrectionParameters class.

    Parameters
    ----------
    tau1 : int
        Time constant for the first correction step in Tian's method. The default value is 300.

    tau2 : int
        Time constant for the second correction step in Tian's method. The default value is 100.
    """

    tau1: int = 300
    tau2: int = 100

Refactored main measurement class for calorimetry data handling.

Measurement

Class for handling and processing isothermal heat flow calorimetry data.

This class coordinates file I/O, data processing, analysis, and visualization operations while maintaining the same API as the original implementation.

Source code in calocem/measurement.py

class Measurement:
    """
    Class for handling and processing isothermal heat flow calorimetry data.

    This class coordinates file I/O, data processing, analysis, and visualization
    operations while maintaining the same API as the original implementation.
    """

    def __init__(
        self,
        folder: Optional[Union[str, pathlib.Path]] = None,
        show_info: bool = True,
        regex: Optional[str] = None,
        auto_clean: bool = False,
        cold_start: bool = True,
        processparams: Optional[ProcessingParameters] = None,
        new_code: bool = False,
        processed: bool = False,
    ):
        """
        Initialize measurements from folder or existing data.

        Parameters
        ----------
        folder : str or pathlib.Path, optional
            Path to folder containing experimental files
        show_info : bool, optional
            Whether to print informative messages, by default True
        regex : str, optional
            Regex pattern to filter files, by default None
        auto_clean : bool, optional
            Whether to clean data automatically, by default False
        cold_start : bool, optional
            Whether to read from files or use cached data, by default True
        processparams : ProcessingParameters, optional
            Processing parameters, by default None. If None, the default parameters will be used
        new_code : bool, optional
            Flag for new code features, by default False
        processed : bool, optional
            Whether data is already processed, i.e., if a .csv file is used which was processed  by Calocem. By default False
        """
        # Initialize attributes
        self._data = pd.DataFrame()
        self._info = pd.DataFrame()
        self._data_unprocessed = pd.DataFrame()
        self._metadata = pd.DataFrame()
        self._metadata_id = ""

        # Store configuration
        self._new_code = new_code
        self._processed = processed

        # Setup processing parameters
        if not isinstance(processparams, ProcessingParameters):
            self.processparams = ProcessingParameters()
        else:
            self.processparams = processparams

        # Initialize components
        self._folder_loader = FolderDataLoader(processed=processed)
        self._data_persistence = DataPersistence()
        self._data_cleaner = DataCleaner()
        self._plotter = SimplePlotter()

        # Load data if folder provided
        if folder:
            try:
                if cold_start:
                    self._load_from_folder(folder, regex, show_info)
                else:
                    self._load_from_cache()

                if auto_clean:
                    self._auto_clean_data()

            except Exception as e:
                if show_info:
                    print(f"Error during initialization: {e}")
                if auto_clean:
                    raise AutoCleanException()
                if not cold_start:
                    raise ColdStartException()
                raise

        # Apply downsampling if requested
        if self.processparams.downsample.apply:
            self._apply_adaptive_downsampling()

        # Information message
        if show_info:
            print("================")
            print(
                "Are you missing some samples? Try rerunning with auto_clean=True and cold_start=True."
            )
            print("================")

    def _load_from_folder(
        self, folder: Union[str, pathlib.Path], regex: Optional[str], show_info: bool
    ):
        """Load data from folder using file loader."""
        try:
            self._data, self._info = self._folder_loader.load_from_folder(
                folder, regex, show_info
            )
            self._data_unprocessed = self._data.copy()

            # Save to cache
            self._data_persistence.save_data(self._data, self._info)

        except Exception as e:
            raise DataProcessingException("load_from_folder", e)

    def _load_from_cache(self):
        """Load data from cached pickle files."""
        try:
            if not self._data_persistence.pickle_files_exist():
                raise FileNotFoundError("No pickle files found for cold start")

            self._data, self._info = self._data_persistence.load_data()
            self._data_unprocessed = self._data.copy()

        except Exception as e:
            raise ColdStartException() from e

    def _auto_clean_data(self):
        """Apply automatic data cleaning."""
        try:
            self._data = self._data_cleaner.auto_clean_data(self._data)
        except Exception as e:
            raise AutoCleanException() from e

    def _apply_adaptive_downsampling(self):
        """Apply adaptive downsampling if configured."""
        # TODO: Implement downsampling logic
        logger.info("Downsampling requested but not yet implemented")

    # Data access methods
    def get_data(self) -> pd.DataFrame:
        """Get the processed calorimetry data."""
        return self._data

    def get_information(self) -> pd.DataFrame:
        """Get the measurement information/metadata."""
        return self._info

    def get_metadata(self) -> tuple:
        """Get added metadata and the ID column name."""
        return self._metadata, self._metadata_id

    def get_sample_names(self) -> list:
        """Get list of sample names."""
        return [
            pathlib.Path(str(sample)).stem
            for sample, _ in SampleIterator.iter_samples(self._data)
        ]

    # Plotting methods
    def plot(
        self,
        t_unit: str = "h",
        y: str = "normalized_heat_flow_w_g",
        y_unit_milli: bool = True,
        regex: Optional[str] = None,
        show_info: bool = True,
        ax=None,
    ):
        """Plot the calorimetry data."""
        return self._plotter.plot_data(
            self._data, t_unit, y, y_unit_milli, regex, show_info, ax
        )

    def plot_by_category(
        self,
        categories: str,
        t_unit: str = "h",
        y: str = "normalized_heat_flow_w_g",
        y_unit_milli: bool = True,
    ):
        """Plot data by metadata categories."""
        # Simplified implementation - would need full metadata integration
        logger.warning(
            "plot_by_category requires metadata integration - not fully implemented"
        )
        yield from []

    # Analysis methods
    def get_peaks(
        self,
        processparams: Optional[ProcessingParameters] = None,
        target_col: str = "normalized_heat_flow_w_g",
        regex: Optional[str] = None,
        cutoff_min: Optional[float] = None,  # Deprecated parameter
        show_plot: bool = True,
        plt_right_s: float = 2e5,
        plt_top: float = 1e-2,
        ax=None,
        xunit: str = "s",
        plot_labels: Optional[bool] = None,
        xmarker: bool = False,
    ) -> pd.DataFrame:
        """Get DataFrame of peak characteristics."""
        if cutoff_min is not None:
            warnings.warn(
                "The cutoff_min parameter is deprecated. Use ProcessingParameters instead.",
                DeprecationWarning,
                stacklevel=2,
            )

        params = processparams or self.processparams
        analyzer = PeakAnalyzer(params)
        peaks_df = analyzer.get_peaks(self._data, target_col, regex)

        if show_plot and not peaks_df.empty:
            # Simple plotting implementation
            for sample, sample_data in SampleIterator.iter_samples(self._data, regex):
                sample_peaks = peaks_df[
                    peaks_df["sample_short"] == pathlib.Path(str(sample)).stem
                ]
                if not sample_peaks.empty:
                    # Get peak indices relative to sample data
                    import numpy as np

                    peak_indices = np.array(sample_peaks.index.tolist())
                    self._plotter.plot_peaks(
                        sample_data, peak_indices, str(sample), ax, "time_s", target_col
                    )

        return peaks_df

    def get_peak_onsets(
        self,
        target_col: str = "normalized_heat_flow_w_g",
        age_col: str = "time_s",
        time_discarded_s: float = 900,
        rolling: int = 1,
        gradient_threshold: float = 0.0005,
        show_plot: bool = False,
        exclude_discarded_time: bool = False,
        regex: Optional[str] = None,
        ax=None,
    ):
        """Get peak onsets based on gradient threshold."""
        analyzer = OnsetAnalyzer(self.processparams)
        return analyzer.get_peak_onsets(
            self._data,
            target_col,
            age_col,
            time_discarded_s,
            rolling,
            gradient_threshold,
            exclude_discarded_time,
            regex,
        )

    def get_maximum_slope(
        self,
        processparams: Optional[ProcessingParameters] = None,
        target_col: str = "normalized_heat_flow_w_g",
        age_col: str = "time_s",
        time_discarded_s: float = 900,
        show_plot: bool = False,
        show_info: bool = True,
        exclude_discarded_time: bool = False,
        regex: Optional[str] = None,
        read_start_c3s: bool = False,
        ax=None,
        save_path: Optional[pathlib.Path] = None,
        xscale: str = "linear",
        xunit: str = "s",
    ):
        """Find the point in time of the maximum slope."""
        params = processparams or self.processparams

        time_discarded_s = (
            params.cutoff.cutoff_min * 60 if params.cutoff.cutoff_min else 0
        )
        analyzer = SlopeAnalyzer(params)

        result = analyzer.get_maximum_slope(
            self._data,
            target_col,
            age_col,
            time_discarded_s,
            exclude_discarded_time,
            regex,
            # read_start_c3s,
            # self._metadata,
        )

        if show_plot and not result.empty:
            for sample, sample_data in SampleIterator.iter_samples(self._data, regex):
                sample_short = pathlib.Path(str(sample)).stem
                sample_result = result[result["sample_short"] == sample_short]
                sample_result = sample_result[
                    sample_result[age_col] >= time_discarded_s
                ]
                if not sample_result.empty:
                    self._plotter.plot_slopes(
                        sample_data,
                        sample_result,
                        str(sample_short),
                        ax,
                        age_col,
                        target_col,
                    )

        return result

    def get_mainpeak_params(
        self,
        processparams: Optional[ProcessingParameters] = None,
        target_col: str = "normalized_heat_flow_w_g",
        age_col: str = "time_s",
        show_plot: bool = False,
        plot_type: str = "mean",
        regex: Optional[str] = None,
        plotpath: Optional[pathlib.Path] = None,
        ax=None,
    ) -> pd.DataFrame:
        """
        Unified method that calculates BOTH maximum and mean slope onset analyses.

        This method performs both slope-based analysis approaches simultaneously:
        - Maximum slope: Uses single point with maximum gradient for onset determination
        - Mean slope: Uses averaged slope over flank windows for onset determination

        Both results are returned in a single DataFrame with all slope values and onsets.

        Parameters
        ----------
        processparams : ProcessingParameters, optional
            Processing parameters, by default None
        target_col : str
            Column containing heat flow data. The default is 'normalized_heat_flow_w_g'.
        age_col : str
            Column containing time data. The default is 'time_s'.
        show_plot : bool
            Whether to plot the results
        plot_type : str
            Type of plot to show: 'max', 'mean',
            - 'max': Shows only maximum slope analysis plot
            - 'mean': Shows only mean slope (flank tangent) analysis plot
        regex : str, optional
            Regex to filter samples
        plotpath : pathlib.Path, optional
            Path to save plots
        ax : matplotlib.axes.Axes, optional
            Matplotlib axes to plot on

        Returns
        -------
        pd.DataFrame
            Comprehensive DataFrame with both max and mean slope results including:
            - Gradients and curvatures at max slope
            - Gradients of mean slope
            - Onset times from both methods
            - Normalized heat flow and heat values at key points
            - Dormant period heat flow values
            - ASTM C1679 characteristic values

        Examples
        --------
        >>> measurement = Measurement(folder="data/")
        >>> mainpeak_params = measurement.get_mainpeak_params(
        ...     processparams=ProcessingParameters(),
        ...     show_plot=False,
        ...     plot_type="mean",
        ... )
        """
        params = processparams or self.processparams

        max_slope_results = self._calculate_max_slope_analysis(
            params,
            target_col,
            age_col,
            regex,
        )

        mean_slope_results = self._calculate_mean_slope_analysis(
            params,
            target_col,
            age_col,
            regex,
        )

        dormant_minimum_heatflow = self.get_dormant_period_heatflow(
            params, regex, show_plot=False
        )

        astm_values = self.get_astm_c1679_characteristics(params, individual=True, show_plot=False, regex=regex)

        # Merge results into comprehensive DataFrame
        combined_results = self._merge_slope_results(
            max_slope_results, mean_slope_results, dormant_minimum_heatflow, astm_values
        )

        # Plot if requested
        if show_plot and not (mean_slope_results.empty or max_slope_results.empty):
            self._plot_combined_slope_analysis(
                combined_results,
                params,
                target_col,
                age_col,
                plot_type,
                regex,
                plotpath,
                ax,
            )
            if not ax:
                plt.show()
        elif mean_slope_results.empty:
            # logger.warning("No slope analysis results to plot.")
            print("No mean slope analysis obtained - check the processing parameters.")

        elif max_slope_results.empty:
            print(
                "No maximum slope analysis obtained - check the processing parameters."
            )

        return combined_results

    def _calculate_max_slope_analysis(
        self,
        params: ProcessingParameters,
        target_col: str,
        age_col: str,
        regex: Optional[str],
    ) -> pd.DataFrame:
        """Calculate maximum slope analysis and return structured results."""
        # Get required data
        max_slope_analyzer = SlopeAnalyzer(params)
        max_slopes = max_slope_analyzer.get_maximum_slope(
            self._data,
            target_col,
            age_col,
            regex,
        )

        if max_slopes.empty:
            logger.warning("No maximum slopes found. Check processing parameters.")
            return pd.DataFrame()

        dormant_hfs = self.get_dormant_period_heatflow(params, regex, show_plot=False)
        if dormant_hfs.empty:
            logger.warning("No dormant period heat flows found.")
            return pd.DataFrame()

        # Calculate onsets
        analyzer = OnsetAnalyzer(params)
        onsets = analyzer.get_peak_onset_via_max_slope(
            self._data,
            max_slopes,
            dormant_hfs,  # intersection, xunit
        )

        # Structure results with consistent naming
        results = []
        for _, slope_row in max_slopes.iterrows():
            sample = slope_row.get("sample", slope_row.get("sample_short", ""))
            sample_short = slope_row.get("sample_short", slope_row.get("sample", ""))

            onset_row = (
                onsets[onsets["sample_short"] == sample_short]
                if not onsets.empty
                else pd.DataFrame()
            )
            onset_time = (
                onset_row.iloc[0]["onset_time_s"] if not onset_row.empty else None
            )

            # get normalized_heat_j_g at onset_time
            if onset_time and not pd.isna(onset_time):
                onset_j_g = np.interp(
                    onset_time,
                    self._data[age_col],
                    self._data["normalized_heat_j_g"],
                )
            else:
                onset_j_g = None

            result_data = {
                "sample": sample,
                "sample_short": sample_short,
                "gradient_from_max_slope": slope_row.get("gradient", 0),
                "curvature_at_max_slope": slope_row.get("curvature", 0),
                "max_slope_time_s": slope_row.get("time_s", 0),
                "normalized_heat_flow_w_g_at_max_slope": slope_row.get(
                    "normalized_heat_flow_w_g", 0
                ),
                "normalized_heat_j_g_at_max_slope": slope_row.get("normalized_heat_j_g", 0),
                "normalized_heat_j_g_at_onset_time_max_slope": onset_j_g,
                "onset_time_s_from_max_slope": onset_time,
                "onset_time_min_max_slope": onset_time / 60 if onset_time else None,
                "onset_time_s_max_slope_abscissa": (
                    onset_row.iloc[0]["onset_time_s_abscissa"]
                    if not onset_row.empty
                    else None
                ),
            }
            results.append(result_data)

        return pd.DataFrame(results)

    def _calculate_mean_slope_analysis(
        self,
        params: ProcessingParameters,
        target_col: str,
        age_col: str,
        regex: Optional[str],
    ) -> pd.DataFrame:
        """Calculate mean slope (flank tangent) analysis and return structured results."""
        analyzer = FlankTangentAnalyzer(params)

        # Get flank tangent results
        tangent_results = analyzer.get_ascending_flank_tangent(
            self._data,
            target_col,
            age_col,
            regex,
        )

        if tangent_results.empty:
            logger.warning("No flank tangent results found.")
            return pd.DataFrame()

        results = []
        for _, row in tangent_results.iterrows():
            sample = row.get("sample", row.get("sample_short", ""))
            sample_short = row.get("sample_short", row.get("sample", ""))

            # onset by intersection with tangent to dormant period
            onset_time = row.get("x_intersection_dormant", row.get("tangent_time_s", 0))

            result_data = {
                "sample": sample,
                "sample_short": sample_short,
                "gradient_of_mean_slope": row.get("tangent_slope", 0),
                "mean_slope_time_s": row.get("tangent_time_s", 0),
                "normalized_heat_flow_w_g_at_mean_slope": row.get("tangent_value", 0),
                "normalized_heat_j_g_at_mean_slope": row.get("tangent_j_g", 0),
                "onset_time_s_from_mean_slope": onset_time,
                "onset_time_min_from_mean_slope": onset_time / 60 if onset_time else None,
                "onset_time_s_from_mean_slope_abscissa": row.get("x_intersection", 0),
                "normalized_heat_at_onset_time_mean_slope_abscissa_j_g": row.get("x_intersection_j_g", 0),
                "normalized_heat_at_onset_time_mean_slope_dormant_j_g": row.get("x_intersection_dormant_j_g", 0),
                "flank_start_value": row.get("flank_start_value", 0),
                "flank_end_value": row.get("flank_end_value", 0),
                "peak_time_s": row.get("peak_time_s", 0),
                "normalized_heat_flow_w_g_at_peak": row.get("peak_value", 0),
                "normalized_heat_j_g_at_peak": row.get("peak_j_g", 0),
            }
            results.append(result_data)

        return pd.DataFrame(results)

    def _merge_slope_results(
        self,
        max_slope_results: pd.DataFrame,
        mean_slope_results: pd.DataFrame,
        dormant_hf_results: pd.DataFrame,
        astm_results: pd.DataFrame,
    ) -> pd.DataFrame:
        """Merge max slope and mean slope results into comprehensive DataFrame."""
        if (
            max_slope_results.empty
            and mean_slope_results.empty
            and dormant_hf_results.empty
            and astm_results.empty
        ):
            return pd.DataFrame()

        # Use outer join to combine results by sample
        if max_slope_results.empty:
            return mean_slope_results
        if mean_slope_results.empty:
            return max_slope_results

        combined = pd.merge(
            max_slope_results,
            mean_slope_results,
            on=["sample", "sample_short"],
            how="outer",
            suffixes=("", "_duplicate"),
        )

        combined = pd.merge(
            combined,
            dormant_hf_results,
            on=["sample", "sample_short"],
            how="outer",
            suffixes=("", "_duplicate"),
        )

        combined = pd.merge(
            combined,
            astm_results,
            on=["sample", "sample_short"],
            how="outer",
            suffixes=("", "_duplicate"),
        )

        duplicate_cols = [col for col in combined.columns if col.endswith("_duplicate")]
        combined = combined.drop(columns=duplicate_cols)

        return combined

    def _plot_combined_slope_analysis(
        self,
        results: pd.DataFrame,
        params: ProcessingParameters,
        target_col: str,
        age_col: str,
        plot_type: str,
        regex: Optional[str],
        plotpath: Optional[pathlib.Path],
        ax,
    ):
        """
        Plot combined slope analysis results based on plot_type parameter.

        Parameters
        ----------
        results : pd.DataFrame
            Combined results containing both max and mean slope data
        target_col : str
            Column name for heat flow data
        age_col : str
            Column name for time data
        plot_type : str
            Type of plot to show: 'max', 'mean', or 'both'
            - 'max': Shows only maximum slope analysis plot
            - 'mean': Shows only mean slope (flank tangent) analysis plot
            - 'both': Shows both analysis types (separate plots for each)
        regex : str, optional
            Regex to filter samples
        plotpath : pathlib.Path, optional
            Path to save plots
        cutoff_min : float, optional
            Cutoff time in minutes
        ax : matplotlib.axes.Axes, optional
            Matplotlib axes to plot on
        """
        # Validate plot_type parameter
        valid_plot_types = ["max", "mean", "both"]
        cutoff_min = params.cutoff.cutoff_min

        if plot_type not in valid_plot_types:
            raise ValueError(
                f"plot_type must be one of {valid_plot_types}, got '{plot_type}'"
            )

        # For now, plot using the existing unified plotting approach
        # This could be enhanced to show both slope methods simultaneously
        for _, result_row in results.iterrows():
            sample = result_row["sample"]
            sample_short = result_row["sample_short"]

            # Get sample data
            sample_data = self._get_filtered_sample_data(
                sample, age_col, cutoff_time_min=cutoff_min
            )
            if sample_data.empty:
                continue

            if not pd.isna(
                result_row.onset_time_s_from_mean_slope or result_row.onset_time_s_from_max_slope
            ):
                self._plotter.plot_tangent_analysis(
                    sample_data,
                    sample_short,
                    params,
                    ax=ax,
                    age_col=age_col,
                    target_col=target_col,
                    cutoff_time_min=cutoff_min,
                    analysis_type=plot_type,  # Use correct analysis type
                    results=result_row.to_frame().T,
                    figsize=(7, 5),
                )
            self._save_and_show_plot(
                plotpath, f"{plot_type}_slope_{sample_short}.png", ax
            )


    # Backward compatibility methods
    def get_peak_onset_via_max_slope(
        self,
        processparams: Optional[ProcessingParameters] = None,
        show_plot: bool = False,
        ax=None,
        regex: Optional[str] = None,
        age_col: str = "time_s",
        target_col: str = "normalized_heat_flow_w_g",
        time_discarded_s: float = 900,
        save_path: Optional[pathlib.Path] = None,
        xscale: str = "linear",
        xunit: str = "s",
        intersection: str = "dormant_hf",
    ):
        """
        Get reaction onset via maximum slope intersection method.

        This is a wrapper around get_peak_onset_via_slope for backward compatibility.
        Returns only the max slope related columns for compatibility.
        """
        full_results = self.get_mainpeak_params(
            processparams=processparams,
            target_col=target_col,
            age_col=age_col,
            show_plot=show_plot,
            regex=regex,
            ax=ax,
            plot_type="max",

            #time_discarded_s=time_discarded_s,
            #intersection=intersection,
            #xunit=xunit,
        )

        if full_results.empty:
            return full_results

        # Extract only max slope related columns for backward compatibility
        # max_slope_cols = [
        #     col
        #     for col in full_results.columns
        #     if col.startswith("max_slope_") or col in ["sample", "sample_short"]
        # ]

        # result = full_results[max_slope_cols].copy()

        # Rename columns to match old API
        # column_mapping = {
        #     "onset_time_s_from_max_slope": "onset_time_s",
        #     "max_slope_onset_time_min": "onset_time_min",
        #     "max_slope_value": "maximum_slope",
        #     "max_slope_time_s": "maximum_slope_time_s",
        # }

        # for old_name, new_name in column_mapping.items():
        #     if old_name in result.columns:
        #         result = result.rename(columns={old_name: new_name})

        return full_results


    def get_ascending_flank_tangent(
        self,
        processparams: Optional[ProcessingParameters] = None,
        target_col: str = "normalized_heat_flow_w_g",
        age_col: str = "time_s",
        flank_fraction_start: float = 0.2,
        flank_fraction_end: float = 0.8,
        window_size: float = 0.1,
        cutoff_min: Optional[float] = None,
        show_plot: bool = False,
        regex: Optional[str] = None,
        plotpath: Optional[pathlib.Path] = None,
        ax=None,
    ) -> pd.DataFrame:
        """
        Determine tangent to ascending flank of peak by averaging over sections.

        This is a wrapper around get_peak_onset_via_slope for backward compatibility.
        Returns only the mean slope related columns for compatibility.
        """
        full_results = self.get_peak_onset_via_slope(
            processparams=processparams,
            target_col=target_col,
            age_col=age_col,
            cutoff_min=cutoff_min,
            show_plot=show_plot,
            regex=regex,
            plotpath=plotpath,
            ax=ax,
            flank_fraction_start=flank_fraction_start,
            flank_fraction_end=flank_fraction_end,
            window_size=window_size,
        )

        if full_results.empty:
            return full_results

        # Extract only mean slope related columns for backward compatibility
        mean_slope_cols = [
            col
            for col in full_results.columns
            if col.startswith("mean_slope_")
            or col in ["sample", "sample_short", "peak_time_s", "peak_value"]
        ]

        result = full_results[mean_slope_cols].copy()

        # Rename columns to match old API
        column_mapping = {
            "mean_slope_onset_time_s": "x_intersection",
            "mean_slope_value": "tangent_slope",
            "mean_slope_time_s": "tangent_time_s",
        }

        for old_name, new_name in column_mapping.items():
            if old_name in result.columns:
                result = result.rename(columns={old_name: new_name})

        return result

    def get_dormant_period_heatflow(
        self,
        processparams: Optional[ProcessingParameters] = None,
        regex: Optional[str] = None,
        cutoff_min: int = 5,
        upper_dormant_thresh_w_g: float = 0.002,
        plot_right_boundary: float = 2e5,
        prominence: float = 1e-3,
        show_plot: bool = False,
    ) -> pd.DataFrame:
        """Get dormant period heat flow characteristics."""
        params = processparams or self.processparams

        # Get peaks first
        peaks = self.get_peaks(params, regex=regex, show_plot=False)

        # Analyze dormant period
        analyzer = DormantPeriodAnalyzer(params)
        dorm_hf = analyzer.get_dormant_period_heatflow(
            self._data, peaks, regex, upper_dormant_thresh_w_g
        )

        if not dorm_hf.empty:
            return dorm_hf
        else:
            return pd.DataFrame()

    def get_astm_c1679_characteristics(
        self,
        processparams: Optional[ProcessingParameters] = None,
        individual: bool = True,
        show_plot: bool = False,
        ax=None,
        regex: Optional[str] = None,
        xscale: str = "log",
        xunit: str = "s",
    ) -> pd.DataFrame:
        """Get characteristics according to ASTM C1679."""
        params = processparams or self.processparams

        # Get peaks first
        peaks = self.get_peaks(params, regex=regex, show_plot=False)

        # Analyze ASTM characteristics
        analyzer = ASTMC1679Analyzer(params)
        df = analyzer.get_astm_c1679_characteristics(
            self._data, peaks, individual, regex
        )
        return df

    def get_cumulated_heat_at_hours(
        self,
        processparams: Optional[ProcessingParameters] = None,
        target_h: float = 4,
        **kwargs,
    ) -> pd.DataFrame:
        """Get cumulated heat flow at specific age."""
        if "cutoff_min" in kwargs:
            cutoff_min = kwargs["cutoff_min"]
            warnings.warn(
                "The cutoff_min parameter is deprecated. Use ProcessingParameters instead.",
                DeprecationWarning,
                stacklevel=2,
            )
        else:
            params = processparams or self.processparams
            cutoff_min = params.cutoff.cutoff_min

        return HeatCalculator.get_cumulated_heat_at_hours(
            self._data, target_h, cutoff_min
        )

    def get_average_slope(
        self,
        processparams: Optional[ProcessingParameters] = None,
        target_col: str = "normalized_heat_flow_w_g",
        age_col: str = "time_s",
        regex: Optional[str] = None,
        show_plot: bool = False,
        ax=None,
        save_path: Optional[pathlib.Path] = None,
        xscale: str = "log",
        xunit: str = "s",
    ) -> pd.DataFrame:
        """Calculate average slope between onset and heat flow maximum."""
        params = processparams or self.processparams

        # Get required data
        max_slopes = self.get_maximum_slope(
            params, target_col, age_col, regex=regex, show_plot=False
        )
        onsets = self.get_peak_onset_via_max_slope(params, regex=regex, show_plot=False)

        if max_slopes.empty or onsets.empty:
            logger.warning("Cannot calculate average slopes - missing required data")
            return pd.DataFrame()

        analyzer = AverageSlopeAnalyzer(params)
        result = analyzer.get_average_slope(
            self._data, max_slopes, onsets, target_col, age_col, regex
        )
        return result

    def _plot_tangent_analysis_unified(
        self,
        results: pd.DataFrame,
        analysis_type: str,
        target_col: str,
        age_col: str,
        regex: Optional[str] = None,
        plotpath: Optional[pathlib.Path] = None,
        cutoff_time_min: Optional[float] = None,
        intersection: str = "dormant_hf",
        xunit: str = "s",
        time_discarded_s: float = 900,
        ax=None,
        # Additional data for onset intersection analysis
        max_slopes: Optional[pd.DataFrame] = None,
        dormant_hfs: Optional[pd.DataFrame] = None,
        onsets: Optional[pd.DataFrame] = None,
    ):
        """
        Unified plotting method for tangent-based analysis results.

        This method handles both flank tangent and onset intersection analysis,
        with the main difference being how the slope is determined:
        - Flank tangent: Uses averaged slope over a window
        - Max slope: Uses single point with maximum gradient

        Parameters
        ----------
        results : pd.DataFrame
            Results from the analysis (tangent results for flank, onsets for max slope)
        analysis_type : str
            Either 'flank_tangent' or 'max_slope_onset'
        target_col : str
            Column name for heat flow data
        age_col : str
            Column name for time data
        regex : str, optional
            Regex to filter samples
        plotpath : pathlib.Path, optional
            Path to save plots
        cutoff_time_min : float, optional
            Cutoff time in minutes
        intersection : str
            Type of intersection for onset analysis ('dormant_hf' or 'abscissa')
        xunit : str
            Time unit for plotting
        time_discarded_s : float
            Time to discard for onset analysis
        ax : matplotlib.axes.Axes, optional
            Matplotlib axes to plot on
        max_slopes : pd.DataFrame, optional
            Required for onset intersection analysis
        dormant_hfs : pd.DataFrame, optional
            Required for onset intersection analysis with dormant_hf
        onsets : pd.DataFrame, optional
            Required for onset intersection analysis
        """
        try:
            if analysis_type == "flank_tangent":
                self._plot_flank_tangent_unified(
                    results, target_col, age_col, regex, plotpath, cutoff_time_min, ax
                )
            elif analysis_type == "max_slope_onset":
                self._plot_onset_intersection_unified(
                    results,
                    max_slopes,
                    dormant_hfs,
                    target_col,
                    age_col,
                    regex,
                    intersection,
                    xunit,
                    time_discarded_s,
                    ax,
                )
            else:
                raise ValueError(f"Unknown analysis_type: {analysis_type}")

        except Exception as e:
            logger.error(f"Error plotting tangent analysis results: {e}")
            print(f"Plotting failed: {e}")

    def _plot_flank_tangent_unified(
        self,
        results: pd.DataFrame,
        target_col: str,
        age_col: str,
        regex: Optional[str] = None,
        plotpath: Optional[pathlib.Path] = None,
        cutoff_time_min: Optional[float] = None,
        ax=None,
    ):
        """Plot flank tangent analysis results using unified SimplePlotter."""
        for _, result_row in results.iterrows():
            sample = result_row["sample"]
            sample_short = result_row["sample_short"]

            # Get sample data
            sample_data = self._get_filtered_sample_data(
                sample, age_col, cutoff_time_min=cutoff_time_min
            )
            if sample_data.empty:
                continue

            # Create a DataFrame with just this result for plotting
            single_result = pd.DataFrame([result_row])

            # Use unified plotting method
            self._plotter.plot_tangent_analysis(
                sample_data,
                sample_short,
                ax=ax,
                age_col=age_col,
                target_col=target_col,
                cutoff_time_min=cutoff_time_min,
                analysis_type="flank_tangent",
                tangent_results=single_result,
                figsize=(7, 5),
            )

            self._save_and_show_plot(plotpath, f"flank_tangent_{sample_short}.png", ax)

    def _plot_onset_intersection_unified(
        self,
        onsets: pd.DataFrame,
        max_slopes: Optional[pd.DataFrame],
        dormant_hfs: Optional[pd.DataFrame],
        target_col: str,
        age_col: str,
        regex: Optional[str] = None,
        intersection: str = "dormant_hf",
        xunit: str = "s",
        time_discarded_s: float = 900,
        ax=None,
    ):
        """Plot onset intersection analysis results using unified SimplePlotter."""
        if max_slopes is None:
            raise ValueError("max_slopes required for onset intersection analysis")

        for _, onset_row in onsets.iterrows():
            sample = onset_row["sample"]

            # Get sample data
            sample_data = self._get_filtered_sample_data(
                sample, age_col, time_discarded_s=time_discarded_s
            )
            if sample_data.empty:
                continue

            # Use unified plotting method
            self._plotter.plot_tangent_analysis(
                sample_data,
                sample,
                ax=ax,
                age_col=age_col,
                target_col=target_col,
                analysis_type="onset_intersection",
                max_slopes=max_slopes,
                dormant_hfs=dormant_hfs,
                onsets=onsets,
                intersection=intersection,
                xunit=xunit,
                figsize=(12, 8),
            )

            # Note: plotpath not available in this context, only show plot
            self._save_and_show_plot(None, f"onset_intersection_{sample}.png", ax)

    def _get_filtered_sample_data(
        self,
        sample: str,
        age_col: str,
        cutoff_time_min: Optional[float] = None,
        time_discarded_s: Optional[float] = None,
    ) -> pd.DataFrame:
        """
        Get sample data with appropriate filtering applied.

        This consolidates the common data filtering logic used in both analysis types.
        """
        # Get sample data - handle both 'sample' and 'sample_short' columns
        sample_data = self._data[
            (self._data["sample"] == sample)
            | (self._data.get("sample_short", "") == sample)
        ]

        if sample_data.empty:
            return sample_data

        # Apply cutoff time filtering
        if cutoff_time_min is not None:
            cutoff_seconds = cutoff_time_min * 60
            sample_data = sample_data[sample_data[age_col] >= cutoff_seconds]

        # Apply time discarded filtering (for onset analysis)
        if time_discarded_s is not None and time_discarded_s > 0:
            sample_data = sample_data[sample_data[age_col] >= time_discarded_s]

        return sample_data

    def _save_and_show_plot(self, plotpath: Optional[pathlib.Path], filename: str, ax):
        """Handle plot saving and showing - common logic for both analysis types."""
        if plotpath:
            plot_file = plotpath / filename
            import matplotlib.pyplot as plt

            plt.savefig(plot_file, dpi=300, bbox_inches="tight")

        import matplotlib.pyplot as plt

        if not ax:
            plt.show()

    def _plot_flank_tangent_results(
        self,
        results: pd.DataFrame,
        target_col: str,
        age_col: str,
        regex: Optional[str] = None,
        plotpath: Optional[pathlib.Path] = None,
        cutoff_time_min: Optional[float] = None,
        ax=None,
    ):
        """
        Plot flank tangent analysis results using SimplePlotter.

        This is a wrapper around the unified plotting method for backward compatibility.
        """
        return self._plot_tangent_analysis_unified(
            results=results,
            analysis_type="flank_tangent",
            target_col=target_col,
            age_col=age_col,
            regex=regex,
            plotpath=plotpath,
            cutoff_time_min=cutoff_time_min,
            ax=ax,
        )

    def _plot_onset_intersections(
        self,
        onsets: pd.DataFrame,
        max_slopes: pd.DataFrame,
        dormant_hfs: pd.DataFrame,
        target_col: str,
        age_col: str,
        regex: Optional[str] = None,
        intersection: str = "dormant_hf",
        xunit: str = "s",
        time_discarded_s: float = 900,
        ax=None,
    ):
        """
        Plot onset intersection analysis results using SimplePlotter.

        This is a wrapper around the unified plotting method for backward compatibility.
        """
        return self._plot_tangent_analysis_unified(
            results=onsets,
            analysis_type="max_slope_onset",
            target_col=target_col,
            age_col=age_col,
            regex=regex,
            intersection=intersection,
            xunit=xunit,
            time_discarded_s=time_discarded_s,
            ax=ax,
            max_slopes=max_slopes,
            dormant_hfs=dormant_hfs,
            onsets=onsets,
        )

    # Data manipulation methods
    def normalize_sample_to_mass(
        self, sample_short: str, mass_g: float, show_info: bool = True
    ):
        """Normalize heat flow values to a specific mass."""
        self._data = DataNormalizer.normalize_sample_to_mass(
            self._data, sample_short, mass_g, show_info
        )

    def add_metadata_source(self, file: str, sample_id_column: str):
        """Add metadata from external source."""
        # TODO: Implement metadata loading
        logger.warning("add_metadata_source not yet implemented in refactored version")

    def remove_pickle_files(self):
        """Remove pickle cache files."""
        self._data_persistence.remove_pickle_files()

    # Private utility methods
    def _iter_samples(self, regex: Optional[str] = None):
        """Iterate over samples - compatibility method."""
        return SampleIterator.iter_samples(self._data, regex)

__init__(folder=None, show_info=True, regex=None, auto_clean=False, cold_start=True, processparams=None, new_code=False, processed=False)

Initialize measurements from folder or existing data.

Parameters:

Name	Type	Description	Default
folder	str or Path	Path to folder containing experimental files	None
show_info	bool	Whether to print informative messages, by default True	True
regex	str	Regex pattern to filter files, by default None	None
auto_clean	bool	Whether to clean data automatically, by default False	False
cold_start	bool	Whether to read from files or use cached data, by default True	True
processparams	ProcessingParameters	Processing parameters, by default None. If None, the default parameters will be used	None
new_code	bool	Flag for new code features, by default False	False
processed	bool	Whether data is already processed, i.e., if a .csv file is used which was processed by Calocem. By default False	False

Source code in calocem/measurement.py

def __init__(
    self,
    folder: Optional[Union[str, pathlib.Path]] = None,
    show_info: bool = True,
    regex: Optional[str] = None,
    auto_clean: bool = False,
    cold_start: bool = True,
    processparams: Optional[ProcessingParameters] = None,
    new_code: bool = False,
    processed: bool = False,
):
    """
    Initialize measurements from folder or existing data.

    Parameters
    ----------
    folder : str or pathlib.Path, optional
        Path to folder containing experimental files
    show_info : bool, optional
        Whether to print informative messages, by default True
    regex : str, optional
        Regex pattern to filter files, by default None
    auto_clean : bool, optional
        Whether to clean data automatically, by default False
    cold_start : bool, optional
        Whether to read from files or use cached data, by default True
    processparams : ProcessingParameters, optional
        Processing parameters, by default None. If None, the default parameters will be used
    new_code : bool, optional
        Flag for new code features, by default False
    processed : bool, optional
        Whether data is already processed, i.e., if a .csv file is used which was processed  by Calocem. By default False
    """
    # Initialize attributes
    self._data = pd.DataFrame()
    self._info = pd.DataFrame()
    self._data_unprocessed = pd.DataFrame()
    self._metadata = pd.DataFrame()
    self._metadata_id = ""

    # Store configuration
    self._new_code = new_code
    self._processed = processed

    # Setup processing parameters
    if not isinstance(processparams, ProcessingParameters):
        self.processparams = ProcessingParameters()
    else:
        self.processparams = processparams

    # Initialize components
    self._folder_loader = FolderDataLoader(processed=processed)
    self._data_persistence = DataPersistence()
    self._data_cleaner = DataCleaner()
    self._plotter = SimplePlotter()

    # Load data if folder provided
    if folder:
        try:
            if cold_start:
                self._load_from_folder(folder, regex, show_info)
            else:
                self._load_from_cache()

            if auto_clean:
                self._auto_clean_data()

        except Exception as e:
            if show_info:
                print(f"Error during initialization: {e}")
            if auto_clean:
                raise AutoCleanException()
            if not cold_start:
                raise ColdStartException()
            raise

    # Apply downsampling if requested
    if self.processparams.downsample.apply:
        self._apply_adaptive_downsampling()

    # Information message
    if show_info:
        print("================")
        print(
            "Are you missing some samples? Try rerunning with auto_clean=True and cold_start=True."
        )
        print("================")

add_metadata_source(file, sample_id_column)

Add metadata from external source.

Source code in calocem/measurement.py

def add_metadata_source(self, file: str, sample_id_column: str):
    """Add metadata from external source."""
    # TODO: Implement metadata loading
    logger.warning("add_metadata_source not yet implemented in refactored version")

get_ascending_flank_tangent(processparams=None, target_col='normalized_heat_flow_w_g', age_col='time_s', flank_fraction_start=0.2, flank_fraction_end=0.8, window_size=0.1, cutoff_min=None, show_plot=False, regex=None, plotpath=None, ax=None)

Determine tangent to ascending flank of peak by averaging over sections.

This is a wrapper around get_peak_onset_via_slope for backward compatibility. Returns only the mean slope related columns for compatibility.

Source code in calocem/measurement.py

def get_ascending_flank_tangent(
    self,
    processparams: Optional[ProcessingParameters] = None,
    target_col: str = "normalized_heat_flow_w_g",
    age_col: str = "time_s",
    flank_fraction_start: float = 0.2,
    flank_fraction_end: float = 0.8,
    window_size: float = 0.1,
    cutoff_min: Optional[float] = None,
    show_plot: bool = False,
    regex: Optional[str] = None,
    plotpath: Optional[pathlib.Path] = None,
    ax=None,
) -> pd.DataFrame:
    """
    Determine tangent to ascending flank of peak by averaging over sections.

    This is a wrapper around get_peak_onset_via_slope for backward compatibility.
    Returns only the mean slope related columns for compatibility.
    """
    full_results = self.get_peak_onset_via_slope(
        processparams=processparams,
        target_col=target_col,
        age_col=age_col,
        cutoff_min=cutoff_min,
        show_plot=show_plot,
        regex=regex,
        plotpath=plotpath,
        ax=ax,
        flank_fraction_start=flank_fraction_start,
        flank_fraction_end=flank_fraction_end,
        window_size=window_size,
    )

    if full_results.empty:
        return full_results

    # Extract only mean slope related columns for backward compatibility
    mean_slope_cols = [
        col
        for col in full_results.columns
        if col.startswith("mean_slope_")
        or col in ["sample", "sample_short", "peak_time_s", "peak_value"]
    ]

    result = full_results[mean_slope_cols].copy()

    # Rename columns to match old API
    column_mapping = {
        "mean_slope_onset_time_s": "x_intersection",
        "mean_slope_value": "tangent_slope",
        "mean_slope_time_s": "tangent_time_s",
    }

    for old_name, new_name in column_mapping.items():
        if old_name in result.columns:
            result = result.rename(columns={old_name: new_name})

    return result

get_astm_c1679_characteristics(processparams=None, individual=True, show_plot=False, ax=None, regex=None, xscale='log', xunit='s')

Get characteristics according to ASTM C1679.

Source code in calocem/measurement.py

def get_astm_c1679_characteristics(
    self,
    processparams: Optional[ProcessingParameters] = None,
    individual: bool = True,
    show_plot: bool = False,
    ax=None,
    regex: Optional[str] = None,
    xscale: str = "log",
    xunit: str = "s",
) -> pd.DataFrame:
    """Get characteristics according to ASTM C1679."""
    params = processparams or self.processparams

    # Get peaks first
    peaks = self.get_peaks(params, regex=regex, show_plot=False)

    # Analyze ASTM characteristics
    analyzer = ASTMC1679Analyzer(params)
    df = analyzer.get_astm_c1679_characteristics(
        self._data, peaks, individual, regex
    )
    return df

get_average_slope(processparams=None, target_col='normalized_heat_flow_w_g', age_col='time_s', regex=None, show_plot=False, ax=None, save_path=None, xscale='log', xunit='s')

Calculate average slope between onset and heat flow maximum.

Source code in calocem/measurement.py

def get_average_slope(
    self,
    processparams: Optional[ProcessingParameters] = None,
    target_col: str = "normalized_heat_flow_w_g",
    age_col: str = "time_s",
    regex: Optional[str] = None,
    show_plot: bool = False,
    ax=None,
    save_path: Optional[pathlib.Path] = None,
    xscale: str = "log",
    xunit: str = "s",
) -> pd.DataFrame:
    """Calculate average slope between onset and heat flow maximum."""
    params = processparams or self.processparams

    # Get required data
    max_slopes = self.get_maximum_slope(
        params, target_col, age_col, regex=regex, show_plot=False
    )
    onsets = self.get_peak_onset_via_max_slope(params, regex=regex, show_plot=False)

    if max_slopes.empty or onsets.empty:
        logger.warning("Cannot calculate average slopes - missing required data")
        return pd.DataFrame()

    analyzer = AverageSlopeAnalyzer(params)
    result = analyzer.get_average_slope(
        self._data, max_slopes, onsets, target_col, age_col, regex
    )
    return result

get_cumulated_heat_at_hours(processparams=None, target_h=4, **kwargs)

Get cumulated heat flow at specific age.

Source code in calocem/measurement.py

def get_cumulated_heat_at_hours(
    self,
    processparams: Optional[ProcessingParameters] = None,
    target_h: float = 4,
    **kwargs,
) -> pd.DataFrame:
    """Get cumulated heat flow at specific age."""
    if "cutoff_min" in kwargs:
        cutoff_min = kwargs["cutoff_min"]
        warnings.warn(
            "The cutoff_min parameter is deprecated. Use ProcessingParameters instead.",
            DeprecationWarning,
            stacklevel=2,
        )
    else:
        params = processparams or self.processparams
        cutoff_min = params.cutoff.cutoff_min

    return HeatCalculator.get_cumulated_heat_at_hours(
        self._data, target_h, cutoff_min
    )

get_data()

Get the processed calorimetry data.

Source code in calocem/measurement.py

def get_data(self) -> pd.DataFrame:
    """Get the processed calorimetry data."""
    return self._data

get_dormant_period_heatflow(processparams=None, regex=None, cutoff_min=5, upper_dormant_thresh_w_g=0.002, plot_right_boundary=200000.0, prominence=0.001, show_plot=False)

Get dormant period heat flow characteristics.

Source code in calocem/measurement.py

def get_dormant_period_heatflow(
    self,
    processparams: Optional[ProcessingParameters] = None,
    regex: Optional[str] = None,
    cutoff_min: int = 5,
    upper_dormant_thresh_w_g: float = 0.002,
    plot_right_boundary: float = 2e5,
    prominence: float = 1e-3,
    show_plot: bool = False,
) -> pd.DataFrame:
    """Get dormant period heat flow characteristics."""
    params = processparams or self.processparams

    # Get peaks first
    peaks = self.get_peaks(params, regex=regex, show_plot=False)

    # Analyze dormant period
    analyzer = DormantPeriodAnalyzer(params)
    dorm_hf = analyzer.get_dormant_period_heatflow(
        self._data, peaks, regex, upper_dormant_thresh_w_g
    )

    if not dorm_hf.empty:
        return dorm_hf
    else:
        return pd.DataFrame()

get_information()

Get the measurement information/metadata.

Source code in calocem/measurement.py

def get_information(self) -> pd.DataFrame:
    """Get the measurement information/metadata."""
    return self._info

get_mainpeak_params(processparams=None, target_col='normalized_heat_flow_w_g', age_col='time_s', show_plot=False, plot_type='mean', regex=None, plotpath=None, ax=None)

Unified method that calculates BOTH maximum and mean slope onset analyses.

This method performs both slope-based analysis approaches simultaneously: - Maximum slope: Uses single point with maximum gradient for onset determination - Mean slope: Uses averaged slope over flank windows for onset determination

Both results are returned in a single DataFrame with all slope values and onsets.

Parameters:

Name	Type	Description	Default
processparams	ProcessingParameters	Processing parameters, by default None	None
target_col	str	Column containing heat flow data. The default is 'normalized_heat_flow_w_g'.	'normalized_heat_flow_w_g'
age_col	str	Column containing time data. The default is 'time_s'.	'time_s'
show_plot	bool	Whether to plot the results	False
plot_type	str	Type of plot to show: 'max', 'mean', - 'max': Shows only maximum slope analysis plot - 'mean': Shows only mean slope (flank tangent) analysis plot	'mean'
regex	str	Regex to filter samples	None
plotpath	Path	Path to save plots	None
ax	Axes	Matplotlib axes to plot on	None

Returns:

Type	Description
DataFrame	Comprehensive DataFrame with both max and mean slope results including: - Gradients and curvatures at max slope - Gradients of mean slope - Onset times from both methods - Normalized heat flow and heat values at key points - Dormant period heat flow values - ASTM C1679 characteristic values

Examples:

>>> measurement = Measurement(folder="data/")
>>> mainpeak_params = measurement.get_mainpeak_params(
...     processparams=ProcessingParameters(),
...     show_plot=False,
...     plot_type="mean",
... )

Source code in calocem/measurement.py

def get_mainpeak_params(
    self,
    processparams: Optional[ProcessingParameters] = None,
    target_col: str = "normalized_heat_flow_w_g",
    age_col: str = "time_s",
    show_plot: bool = False,
    plot_type: str = "mean",
    regex: Optional[str] = None,
    plotpath: Optional[pathlib.Path] = None,
    ax=None,
) -> pd.DataFrame:
    """
    Unified method that calculates BOTH maximum and mean slope onset analyses.

    This method performs both slope-based analysis approaches simultaneously:
    - Maximum slope: Uses single point with maximum gradient for onset determination
    - Mean slope: Uses averaged slope over flank windows for onset determination

    Both results are returned in a single DataFrame with all slope values and onsets.

    Parameters
    ----------
    processparams : ProcessingParameters, optional
        Processing parameters, by default None
    target_col : str
        Column containing heat flow data. The default is 'normalized_heat_flow_w_g'.
    age_col : str
        Column containing time data. The default is 'time_s'.
    show_plot : bool
        Whether to plot the results
    plot_type : str
        Type of plot to show: 'max', 'mean',
        - 'max': Shows only maximum slope analysis plot
        - 'mean': Shows only mean slope (flank tangent) analysis plot
    regex : str, optional
        Regex to filter samples
    plotpath : pathlib.Path, optional
        Path to save plots
    ax : matplotlib.axes.Axes, optional
        Matplotlib axes to plot on

    Returns
    -------
    pd.DataFrame
        Comprehensive DataFrame with both max and mean slope results including:
        - Gradients and curvatures at max slope
        - Gradients of mean slope
        - Onset times from both methods
        - Normalized heat flow and heat values at key points
        - Dormant period heat flow values
        - ASTM C1679 characteristic values

    Examples
    --------
    >>> measurement = Measurement(folder="data/")
    >>> mainpeak_params = measurement.get_mainpeak_params(
    ...     processparams=ProcessingParameters(),
    ...     show_plot=False,
    ...     plot_type="mean",
    ... )
    """
    params = processparams or self.processparams

    max_slope_results = self._calculate_max_slope_analysis(
        params,
        target_col,
        age_col,
        regex,
    )

    mean_slope_results = self._calculate_mean_slope_analysis(
        params,
        target_col,
        age_col,
        regex,
    )

    dormant_minimum_heatflow = self.get_dormant_period_heatflow(
        params, regex, show_plot=False
    )

    astm_values = self.get_astm_c1679_characteristics(params, individual=True, show_plot=False, regex=regex)

    # Merge results into comprehensive DataFrame
    combined_results = self._merge_slope_results(
        max_slope_results, mean_slope_results, dormant_minimum_heatflow, astm_values
    )

    # Plot if requested
    if show_plot and not (mean_slope_results.empty or max_slope_results.empty):
        self._plot_combined_slope_analysis(
            combined_results,
            params,
            target_col,
            age_col,
            plot_type,
            regex,
            plotpath,
            ax,
        )
        if not ax:
            plt.show()
    elif mean_slope_results.empty:
        # logger.warning("No slope analysis results to plot.")
        print("No mean slope analysis obtained - check the processing parameters.")

    elif max_slope_results.empty:
        print(
            "No maximum slope analysis obtained - check the processing parameters."
        )

    return combined_results

get_maximum_slope(processparams=None, target_col='normalized_heat_flow_w_g', age_col='time_s', time_discarded_s=900, show_plot=False, show_info=True, exclude_discarded_time=False, regex=None, read_start_c3s=False, ax=None, save_path=None, xscale='linear', xunit='s')

Find the point in time of the maximum slope.

Source code in calocem/measurement.py

def get_maximum_slope(
    self,
    processparams: Optional[ProcessingParameters] = None,
    target_col: str = "normalized_heat_flow_w_g",
    age_col: str = "time_s",
    time_discarded_s: float = 900,
    show_plot: bool = False,
    show_info: bool = True,
    exclude_discarded_time: bool = False,
    regex: Optional[str] = None,
    read_start_c3s: bool = False,
    ax=None,
    save_path: Optional[pathlib.Path] = None,
    xscale: str = "linear",
    xunit: str = "s",
):
    """Find the point in time of the maximum slope."""
    params = processparams or self.processparams

    time_discarded_s = (
        params.cutoff.cutoff_min * 60 if params.cutoff.cutoff_min else 0
    )
    analyzer = SlopeAnalyzer(params)

    result = analyzer.get_maximum_slope(
        self._data,
        target_col,
        age_col,
        time_discarded_s,
        exclude_discarded_time,
        regex,
        # read_start_c3s,
        # self._metadata,
    )

    if show_plot and not result.empty:
        for sample, sample_data in SampleIterator.iter_samples(self._data, regex):
            sample_short = pathlib.Path(str(sample)).stem
            sample_result = result[result["sample_short"] == sample_short]
            sample_result = sample_result[
                sample_result[age_col] >= time_discarded_s
            ]
            if not sample_result.empty:
                self._plotter.plot_slopes(
                    sample_data,
                    sample_result,
                    str(sample_short),
                    ax,
                    age_col,
                    target_col,
                )

    return result

get_metadata()

Get added metadata and the ID column name.

Source code in calocem/measurement.py

def get_metadata(self) -> tuple:
    """Get added metadata and the ID column name."""
    return self._metadata, self._metadata_id

get_peak_onset_via_max_slope(processparams=None, show_plot=False, ax=None, regex=None, age_col='time_s', target_col='normalized_heat_flow_w_g', time_discarded_s=900, save_path=None, xscale='linear', xunit='s', intersection='dormant_hf')

Get reaction onset via maximum slope intersection method.

This is a wrapper around get_peak_onset_via_slope for backward compatibility. Returns only the max slope related columns for compatibility.

Source code in calocem/measurement.py

def get_peak_onset_via_max_slope(
    self,
    processparams: Optional[ProcessingParameters] = None,
    show_plot: bool = False,
    ax=None,
    regex: Optional[str] = None,
    age_col: str = "time_s",
    target_col: str = "normalized_heat_flow_w_g",
    time_discarded_s: float = 900,
    save_path: Optional[pathlib.Path] = None,
    xscale: str = "linear",
    xunit: str = "s",
    intersection: str = "dormant_hf",
):
    """
    Get reaction onset via maximum slope intersection method.

    This is a wrapper around get_peak_onset_via_slope for backward compatibility.
    Returns only the max slope related columns for compatibility.
    """
    full_results = self.get_mainpeak_params(
        processparams=processparams,
        target_col=target_col,
        age_col=age_col,
        show_plot=show_plot,
        regex=regex,
        ax=ax,
        plot_type="max",

        #time_discarded_s=time_discarded_s,
        #intersection=intersection,
        #xunit=xunit,
    )

    if full_results.empty:
        return full_results

    # Extract only max slope related columns for backward compatibility
    # max_slope_cols = [
    #     col
    #     for col in full_results.columns
    #     if col.startswith("max_slope_") or col in ["sample", "sample_short"]
    # ]

    # result = full_results[max_slope_cols].copy()

    # Rename columns to match old API
    # column_mapping = {
    #     "onset_time_s_from_max_slope": "onset_time_s",
    #     "max_slope_onset_time_min": "onset_time_min",
    #     "max_slope_value": "maximum_slope",
    #     "max_slope_time_s": "maximum_slope_time_s",
    # }

    # for old_name, new_name in column_mapping.items():
    #     if old_name in result.columns:
    #         result = result.rename(columns={old_name: new_name})

    return full_results

get_peak_onsets(target_col='normalized_heat_flow_w_g', age_col='time_s', time_discarded_s=900, rolling=1, gradient_threshold=0.0005, show_plot=False, exclude_discarded_time=False, regex=None, ax=None)

Get peak onsets based on gradient threshold.

Source code in calocem/measurement.py

def get_peak_onsets(
    self,
    target_col: str = "normalized_heat_flow_w_g",
    age_col: str = "time_s",
    time_discarded_s: float = 900,
    rolling: int = 1,
    gradient_threshold: float = 0.0005,
    show_plot: bool = False,
    exclude_discarded_time: bool = False,
    regex: Optional[str] = None,
    ax=None,
):
    """Get peak onsets based on gradient threshold."""
    analyzer = OnsetAnalyzer(self.processparams)
    return analyzer.get_peak_onsets(
        self._data,
        target_col,
        age_col,
        time_discarded_s,
        rolling,
        gradient_threshold,
        exclude_discarded_time,
        regex,
    )

get_peaks(processparams=None, target_col='normalized_heat_flow_w_g', regex=None, cutoff_min=None, show_plot=True, plt_right_s=200000.0, plt_top=0.01, ax=None, xunit='s', plot_labels=None, xmarker=False)

Get DataFrame of peak characteristics.

Source code in calocem/measurement.py

def get_peaks(
    self,
    processparams: Optional[ProcessingParameters] = None,
    target_col: str = "normalized_heat_flow_w_g",
    regex: Optional[str] = None,
    cutoff_min: Optional[float] = None,  # Deprecated parameter
    show_plot: bool = True,
    plt_right_s: float = 2e5,
    plt_top: float = 1e-2,
    ax=None,
    xunit: str = "s",
    plot_labels: Optional[bool] = None,
    xmarker: bool = False,
) -> pd.DataFrame:
    """Get DataFrame of peak characteristics."""
    if cutoff_min is not None:
        warnings.warn(
            "The cutoff_min parameter is deprecated. Use ProcessingParameters instead.",
            DeprecationWarning,
            stacklevel=2,
        )

    params = processparams or self.processparams
    analyzer = PeakAnalyzer(params)
    peaks_df = analyzer.get_peaks(self._data, target_col, regex)

    if show_plot and not peaks_df.empty:
        # Simple plotting implementation
        for sample, sample_data in SampleIterator.iter_samples(self._data, regex):
            sample_peaks = peaks_df[
                peaks_df["sample_short"] == pathlib.Path(str(sample)).stem
            ]
            if not sample_peaks.empty:
                # Get peak indices relative to sample data
                import numpy as np

                peak_indices = np.array(sample_peaks.index.tolist())
                self._plotter.plot_peaks(
                    sample_data, peak_indices, str(sample), ax, "time_s", target_col
                )

    return peaks_df

get_sample_names()

Get list of sample names.

Source code in calocem/measurement.py

def get_sample_names(self) -> list:
    """Get list of sample names."""
    return [
        pathlib.Path(str(sample)).stem
        for sample, _ in SampleIterator.iter_samples(self._data)
    ]

normalize_sample_to_mass(sample_short, mass_g, show_info=True)

Normalize heat flow values to a specific mass.

Source code in calocem/measurement.py

def normalize_sample_to_mass(
    self, sample_short: str, mass_g: float, show_info: bool = True
):
    """Normalize heat flow values to a specific mass."""
    self._data = DataNormalizer.normalize_sample_to_mass(
        self._data, sample_short, mass_g, show_info
    )

plot(t_unit='h', y='normalized_heat_flow_w_g', y_unit_milli=True, regex=None, show_info=True, ax=None)

Plot the calorimetry data.

Source code in calocem/measurement.py

def plot(
    self,
    t_unit: str = "h",
    y: str = "normalized_heat_flow_w_g",
    y_unit_milli: bool = True,
    regex: Optional[str] = None,
    show_info: bool = True,
    ax=None,
):
    """Plot the calorimetry data."""
    return self._plotter.plot_data(
        self._data, t_unit, y, y_unit_milli, regex, show_info, ax
    )

plot_by_category(categories, t_unit='h', y='normalized_heat_flow_w_g', y_unit_milli=True)

Plot data by metadata categories.

Source code in calocem/measurement.py

def plot_by_category(
    self,
    categories: str,
    t_unit: str = "h",
    y: str = "normalized_heat_flow_w_g",
    y_unit_milli: bool = True,
):
    """Plot data by metadata categories."""
    # Simplified implementation - would need full metadata integration
    logger.warning(
        "plot_by_category requires metadata integration - not fully implemented"
    )
    yield from []

remove_pickle_files()

Remove pickle cache files.

Source code in calocem/measurement.py

def remove_pickle_files(self):
    """Remove pickle cache files."""
    self._data_persistence.remove_pickle_files()