"""
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 collections.abc import Iterator, Sequence
from contextlib import contextmanager
from typing import TYPE_CHECKING, Callable
if TYPE_CHECKING:
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 # pylint: disable=import-outside-toplevel
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 # pylint: disable=import-outside-toplevel
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)