"""
This module defines a :class:`logging.LoggerAdapter` and
:class:`logging.Filter`. They are used to enable the capturing of output from
specific
instruments.
"""
import collections.abc
import logging
from collections.abc import Iterator, MutableMapping, Sequence
from contextlib import contextmanager
from typing import TYPE_CHECKING, Any, Optional, Union
from .logger import LevelType, get_console_handler, handler_level
if TYPE_CHECKING:
from qcodes.instrument import InstrumentBase
class InstrumentLoggerAdapter(logging.LoggerAdapter):
"""
In the Python logging module adapters are used to add context information
to logging. The :class:`logging.LoggerAdapter` has the same methods as the
:class:`logging.Logger` and can thus be used as such.
Here it is used to add the instruments full name to the log records so that
they can be filtered (using the :class:`InstrumentFilter`) by instrument
instance.
The context data gets stored in the `extra` dictionary as a property of the
Adapter. It is filled by the ``__init__`` method::
>>> LoggerAdapter(log, {'instrument': instrument_instance})
"""
def process(
self, msg: str, kwargs: MutableMapping[str, Any]
) -> tuple[str, MutableMapping[str, Any]]:
"""
Returns the message and the kwargs for the handlers.
"""
assert self.extra is not None
extra = dict(self.extra)
inst = extra.pop("instrument")
full_name = getattr(inst, "full_name", None)
instr_type = str(type(inst).__name__)
kwargs["extra"] = extra
kwargs["extra"]["instrument_name"] = str(full_name)
kwargs["extra"]["instrument_type"] = instr_type
return f"[{full_name}({instr_type})] {msg}", kwargs
class InstrumentFilter(logging.Filter):
"""
Filter to filter out records that originate from the given instruments.
Records created through the :class:`InstrumentLoggerAdapter` have additional
properties as specified in the `extra` dictionary which is a property of
the adapter.
Here the ``instrument_name`` property gets used to reject records that don't
originate from the list of instruments that has been passed to the
``__init__`` method.
"""
def __init__(self, instruments: Union['InstrumentBase',
Sequence['InstrumentBase']]):
super().__init__()
if not isinstance(instruments, collections.abc.Sequence):
instrument_seq: Sequence[str] = (instruments.full_name,)
else:
instrument_seq = [inst.full_name for inst in instruments]
self.instrument_set = set(instrument_seq)
def filter(self, record: logging.LogRecord) -> bool:
inst: Optional[str] = getattr(record, "instrument_name", None)
if inst is None:
return False
insrument_match = any(
inst.startswith(instrument_name) for instrument_name in self.instrument_set
)
return insrument_match
[docs]
def get_instrument_logger(instrument_instance: 'InstrumentBase',
logger_name: Optional[str] = None
) -> InstrumentLoggerAdapter:
"""
Returns an :class:`InstrumentLoggerAdapter` that can be used to log
messages
including ``instrument_instance`` as an additional context.
The :class:`logging.LoggerAdapter` object can be used as any logger.
Args:
instrument_instance: The instrument instance to be added to the context
of the log record.
logger_name: Name of the logger to which the records will be passed.
If `None`, defaults to the root logger.
Returns:
:class:`logging.LoggerAdapter` instance, that can be used for instrument
specific logging.
"""
logger_name = logger_name or ''
return InstrumentLoggerAdapter(logging.getLogger(logger_name),
{'instrument': instrument_instance})
[docs]
@contextmanager
def filter_instrument(instrument: Union['InstrumentBase',
Sequence['InstrumentBase']],
handler: Optional[
Union[logging.Handler,
Sequence[logging.Handler]]] = None,
level: Optional[LevelType] = None) -> Iterator[None]:
"""
Context manager that adds a filter that only enables the log messages of
the supplied instruments to pass.
Example:
>>> h1, h2 = logger.get_console_handler(), logger.get_file_handler()
>>> with logger.filter_instruments((qdac, dmm2), handler=[h1, h2]):
>>> qdac.ch01(1) # logged
>>> v1 = dmm2.v() # logged
>>> v2 = keithley.v() # not logged
Args:
instrument: The instrument or sequence of instruments to enable
messages from.
level: Level to set the handlers to.
handler: Single or sequence of handlers to change.
"""
handlers: Sequence[logging.Handler]
if handler is None:
myhandler = get_console_handler()
if myhandler is None:
raise RuntimeError("Trying to filter instrument but no handler "
"defined. Did you forget to call "
"`start_logger` before?")
handlers = (myhandler,)
elif not isinstance(handler, collections.abc.Sequence):
handlers = (handler,)
else:
handlers = handler
instrument_filter = InstrumentFilter(instrument)
for h in handlers:
h.addFilter(instrument_filter)
try:
if level is not None:
with handler_level(level, handlers):
yield
else:
yield
finally:
for h in handlers:
h.removeFilter(instrument_filter)