Source code for qcodes.logger.log_analysis

"""
This module defines a number of functions to make it easier to
work with log messages from QCoDeS. Specifically it enables
exports of logs and log files to a :class:`pd.DataFrame`

"""

from __future__ import annotations

import io
import logging
from contextlib import contextmanager
from typing import TYPE_CHECKING

if TYPE_CHECKING:
    from collections.abc import Callable, Iterator, Sequence

    import numpy as np
    import numpy.typing as npt
    import pandas as pd

from .logger import (
    FORMAT_STRING_DICT,
    LOGGING_SEPARATOR,
    LevelType,
    get_formatter,
    get_log_file_name,
)


[docs] def log_to_dataframe( log: Sequence[str], columns: Sequence[str] | None = None, separator: str | None = None, ) -> pd.DataFrame: """ Return the provided or default log string as a :class:`pd.DataFrame`. Unless :data:`qcodes.logger.logger.LOGGING_SEPARATOR` or :data:`qcodes.logger.logger.FORMAT_STRING_DICT` have been changed using the default for the columns and separator arguments is encouraged. Lines starting with a digit are ignored. In the log setup of :func:`qcodes.logger.logger.start_logger` Traceback messages are also logged. These start with a digit. Args: log: Log content. columns: Column headers for the returned dataframe, defaults to columns used by handlers set up by :func:`qcodes.logger.logger.start_logger`. separator: Separator of the log file to separate the columns, defaults to separator used by handlers set up by :func:`qcodes.logger.logger.start_logger`. Returns: A :class:`pd.DataFrame` containing the log content. """ import pandas as pd separator = separator or LOGGING_SEPARATOR columns = columns or list(FORMAT_STRING_DICT.keys()) # note: if we used commas as separators, pandas read_csv # could be faster than this string comprehension split_cont = [ line.split(separator) for line in log if line[0].isdigit() ] # avoid tracebacks dataframe = pd.DataFrame(split_cont, columns=list(columns)) return dataframe
[docs] def logfile_to_dataframe( logfile: str | None = None, columns: Sequence[str] | None = None, separator: str | None = None, ) -> pd.DataFrame: """ Return the provided or default logfile as a :class:`pd.DataFrame`. Unless :data:`qcodes.logger.logger.LOGGING_SEPARATOR` or :data:`qcodes.logger.logger.FORMAT_STRING_DICT` have been changed using the default for the columns and separator arguments is encouraged. Lines starting with a digit are ignored. In the log setup of :func:`qcodes.logger.logger.start_logger` Traceback messages are also logged. These start with a digit. Args: logfile: Name of the logfile, defaults to current default log file. columns: Column headers for the returned dataframe, defaults to columns used by handlers set up by :func:`qcodes.logger.logger.start_logger`. separator: Separator of the logfile to separate the columns, defaults to separator used by handlers set up by :func:`qcodes.logger.logger.start_logger`. Returns: A :class:`pd.DataFrame` containing the logfile content. """ logfile = logfile or get_log_file_name() with open(logfile) as f: raw_cont = f.readlines() return log_to_dataframe(raw_cont, columns, separator)
[docs] def time_difference( firsttimes: pd.Series, secondtimes: pd.Series, use_first_series_labels: bool = True ) -> pd.Series: """ Calculate the time differences between two series containing time stamp strings as their values. Args: firsttimes: The oldest times secondtimes: The youngest times use_first_series_labels: If true, the returned Series has the same labels as firsttimes. Else it has the labels of secondtimes Returns: A :class:`pd.Series` with float values of the time difference (s) """ import pandas as pd if "," in firsttimes.iloc[0]: nfirsttimes = firsttimes.str.replace(",", ".") else: nfirsttimes = firsttimes if "," in secondtimes.iloc[0]: nsecondtimes = secondtimes.str.replace(",", ".") else: nsecondtimes = secondtimes t0s = nfirsttimes.astype("datetime64[ns]") t1s = nsecondtimes.astype("datetime64[ns]") t1s_np: npt.NDArray[np.datetime64] = t1s.to_numpy() t0s_np: npt.NDArray[np.datetime64] = t0s.to_numpy() timedeltas = (t1s_np - t0s_np).astype("float") * 1e-9 if use_first_series_labels: output = pd.Series(timedeltas, index=nfirsttimes.index) else: output = pd.Series(timedeltas, index=nsecondtimes.index) return output
[docs] @contextmanager def capture_dataframe( level: LevelType = logging.DEBUG, logger: logging.Logger | None = None ) -> Iterator[tuple[logging.StreamHandler, Callable[[], pd.DataFrame]]]: """ Context manager to capture the logs in a :class:`pd.DataFrame` Example: >>> with logger.capture_dataframe() as (handler, cb): >>> qdac.ch01(1) # some commands >>> data_frame = cb() Args: level: Level at which to capture. logger: Logger used to capture the data. Will default to root logger if None is supplied. Returns: Tuple of handler that is used to capture the log messages and callback that returns the cumulative :class:`pd.DataFrame` at any given point (within the context) """ # get root logger if none is specified. logger = logger or logging.getLogger() with io.StringIO() as log_capture: string_handler = logging.StreamHandler(log_capture) string_handler.setLevel(level) string_handler.setFormatter(get_formatter()) logger.addHandler(string_handler) try: yield ( string_handler, lambda: log_to_dataframe(log_capture.getvalue().splitlines()), ) finally: logger.removeHandler(string_handler)