deeplabcut.core.debug

Modules:

Name	Description
debug_logger

Classes:

Name	Description
ExecutableSpec	Small description of an external executable to report.
InMemoryDebugRecorder	Lightweight, fail-open in-memory log recorder.
LibrarySpec	Small description of a library to report.

Functions:

Name	Description
install_debug_recorder	Attach a single in-memory recorder to the requested logger namespace.
log_timing	Lightweight scoped timer for debug instrumentation.

ExecutableSpec dataclass

Small description of an external executable to report.

Parameters

key: Label used in the output report. command: Executable name or absolute path to resolve. version_args: Arguments used to query the executable version.

Source code in deeplabcut/core/debug/debug_logger.py

@dataclass(frozen=True)
class ExecutableSpec:
    """Small description of an external executable to report.

    Parameters
    ----------
    key:
        Label used in the output report.
    command:
        Executable name or absolute path to resolve.
    version_args:
        Arguments used to query the executable version.
    """

    key: str
    command: str | None = None
    version_args: tuple[str, ...] = ("-version",)

    def resolved_command(self) -> str:
        return self.command or self.key

InMemoryDebugRecorder

Bases: Handler

Lightweight, fail-open in-memory log recorder.

Safety properties: - bounded memory via deque(maxlen=...) - no file/network I/O - swallow-all-errors in emit() - does not log from inside itself - stores only small text snapshots

Source code in deeplabcut/core/debug/debug_logger.py

class InMemoryDebugRecorder(logging.Handler):
    """Lightweight, fail-open in-memory log recorder.

    Safety properties:
    - bounded memory via deque(maxlen=...)
    - no file/network I/O
    - swallow-all-errors in emit()
    - does not log from inside itself
    - stores only small text snapshots
    """

    def __init__(self, *, capacity: int = LOG_QUEUE_MAXLEN, level: int = logging.DEBUG):
        super().__init__(level=level)
        self._records: deque[RecordedLog] = deque(maxlen=max(1, int(capacity)))
        self._lock = threading.Lock()
        self._dropped = 0

    @property
    def dropped_count(self) -> int:
        with self._lock:
            return self._dropped

    def emit(self, record: logging.LogRecord) -> None:
        try:
            # Never call logging from here.
            # Never inspect application objects.
            msg = self._safe_message(record)
            exc_text = self._safe_exception_text(record)

            snap = RecordedLog(
                created=float(getattr(record, "created", 0.0) or 0.0),
                level=str(getattr(record, "levelname", "UNKNOWN")),
                logger_name=str(getattr(record, "name", "")),
                message=msg,
                exc_text=exc_text,
            )

            with self._lock:
                self._records.append(snap)

        except Exception:
            # Fail open: never let diagnostics interfere with runtime behavior.
            try:
                self._dropped += 1
            except Exception:
                pass

    def clear(self) -> None:
        try:
            with self._lock:
                self._records.clear()
                self._dropped = 0
        except Exception:
            pass

    def snapshot(self) -> list[RecordedLog]:
        try:
            with self._lock:
                return list(self._records)
        except Exception:
            return []

    def render_text(self, *, limit: int = 200) -> str:
        lines: list[str] = []
        try:
            records = self.snapshot()[-max(1, int(limit)) :]
            if not records:
                if self._dropped:
                    return f"[debug-recorder] no captured logs, {self._dropped} internal failures"
                return ""

            base = records[0].created
            for rec in records:
                ts = datetime.fromtimestamp(rec.created).strftime("%H:%M:%S.%f")[:-3]
                if DLC_LOG_TIMING:
                    rel_ms = (rec.created - base) * 1000.0
                    lines.append(f"{ts} (+{rel_ms:8.1f} ms) | {rec.level:<8} | {rec.logger_name} | {rec.message}")
                else:
                    lines.append(f"{ts} | {rec.level:<8} | {rec.logger_name} | {rec.message}")
                if rec.exc_text:
                    lines.append(rec.exc_text.rstrip())

            if self._dropped:
                lines.append(f"[debug-recorder] dropped internal failures: {self._dropped}")
        except Exception:
            return "[debug-recorder] failed to render logs"
        return "\n".join(lines)

    @staticmethod
    def _safe_message(record: logging.LogRecord) -> str:
        try:
            return record.getMessage()
        except Exception:
            try:
                return str(record.msg)
            except Exception:
                return "<unrenderable log message>"

    @staticmethod
    def _safe_exception_text(record: logging.LogRecord) -> str | None:
        try:
            if not record.exc_info:
                return None
            return "".join(traceback.format_exception(*record.exc_info))
        except Exception:
            return "<exception details unavailable>"

LibrarySpec dataclass

Small description of a library to report.

Source code in deeplabcut/core/debug/debug_logger.py

@dataclass(frozen=True)
class LibrarySpec:
    """Small description of a library to report."""

    key: str
    dist_name: str | None = None
    module_name: str | None = None
    prefer_module_version: bool = False

    def resolved_dist_name(self) -> str:
        return self.dist_name or self.key

    def resolved_module_name(self) -> str:
        return self.module_name or self.key

install_debug_recorder

install_debug_recorder(
    *,
    logger_name: str = "deeplabcut",
    capacity: int = LOG_QUEUE_MAXLEN,
    handler_level: int = logging.INFO,
    ensure_logger_level: int | None = None
) -> InMemoryDebugRecorder

Attach a single in-memory recorder to the requested logger namespace.

Idempotent: repeated calls return the same recorder.

Parameters

logger_name: Logger namespace to attach the recorder to. capacity: Maximum number of captured records. By default, uses LOG_QUEUE_MAXLEN. handler_level: Minimum level stored by the recorder itself. ensure_logger_level: Controls whether to adjust the target logger level.

- None: never modify the logger level
- int: lower the logger only if its effective level is more restrictive

Source code in deeplabcut/core/debug/debug_logger.py

def install_debug_recorder(
    *,
    logger_name: str = "deeplabcut",
    capacity: int = LOG_QUEUE_MAXLEN,
    handler_level: int = logging.INFO,
    ensure_logger_level: int | None = None,
) -> InMemoryDebugRecorder:
    """Attach a single in-memory recorder to the requested logger namespace.

    Idempotent: repeated calls return the same recorder.

    Parameters
    ----------
    logger_name:
        Logger namespace to attach the recorder to.
    capacity:
        Maximum number of captured records. By default, uses LOG_QUEUE_MAXLEN.
    handler_level:
        Minimum level stored by the recorder itself.
    ensure_logger_level:
        Controls whether to adjust the target logger level.

        - None: never modify the logger level
        - int: lower the logger only if its effective level is more restrictive
    """

    root_logger = logging.getLogger(logger_name)

    existing = getattr(root_logger, _DEBUG_HANDLER_ATTR, None)
    if isinstance(existing, InMemoryDebugRecorder):
        return existing

    recorder = InMemoryDebugRecorder(capacity=capacity, level=handler_level)
    recorder.set_name("deeplabcut-debug-recorder")

    # Important:
    # - attach only to a DLC-owned logger namespace, not the global root logger
    # - keep propagation unchanged
    # - logger level adjustment, if any, is handled below; "auto" initializes
    #   an unset logger to ``handler_level`` rather than forcing DEBUG
    root_logger.addHandler(recorder)

    if isinstance(ensure_logger_level, int):
        # Only lower verbosity if explicitly requested.
        if root_logger.getEffectiveLevel() > ensure_logger_level:
            root_logger.setLevel(ensure_logger_level)

    setattr(root_logger, _DEBUG_HANDLER_ATTR, recorder)
    return recorder

log_timing

log_timing(logger: Logger, label: str, *, level: int = logging.DEBUG, threshold_ms: float | None = None)

Lightweight scoped timer for debug instrumentation.

Uses perf_counter_ns() for monotonic timing. Logs only if logger is enabled for the requested level. Optionally suppresses tiny timings below threshold_ms.

Source code in deeplabcut/core/debug/debug_logger.py

@contextmanager
def log_timing(
    logger: logging.Logger,
    label: str,
    *,
    level: int = logging.DEBUG,
    threshold_ms: float | None = None,
):
    """Lightweight scoped timer for debug instrumentation.

    Uses perf_counter_ns() for monotonic timing.
    Logs only if logger is enabled for the requested level.
    Optionally suppresses tiny timings below ``threshold_ms``.
    """
    if not logger.isEnabledFor(level) or not DLC_LOG_TIMING:
        yield
        return

    effective_threshold_ms = threshold_ms if threshold_ms is not None else DLC_LOG_TIMING_THRESHOLD_MS
    t0 = perf_counter_ns()
    try:
        yield
    finally:
        dt_ms = (perf_counter_ns() - t0) / 1e6
        if effective_threshold_ms is None or dt_ms >= effective_threshold_ms:
            logger.log(level, "%s took %.3f ms", label, dt_ms)