"""
Package-scope Python `logging` setup for spacr.
Central configuration for every spacr subsystem — core pipelines,
I/O, measure, utilities, and the Qt GUI all funnel through the same
rotating file handler at ``~/.spacr/logs/spacr.log``.
Two ways to opt in:
- Automatic — the Qt GUI calls :func:`setup_logging` at launch, so
once you run ``spacr-qt`` the log file is populated for the life
of the session.
- Manual — for headless scripts and notebooks:
.. code-block:: python
from spacr.logging_util import setup_logging, get_logger
setup_logging() # once, at program start
LOG = get_logger(__name__) # in every module that logs
LOG.info("started")
The log level can be overridden by ``SPACR_LOG_LEVEL`` in the env
(``DEBUG``, ``INFO``, ``WARNING``, …). :func:`enable_debug` and
:func:`disable_debug` are convenience toggles for interactive use.
Third-party libraries that spam INFO records during a spacr pipeline
(torch, cellpose, matplotlib, PIL, urllib3, botocore, tensorflow,
asyncio) are pinned to WARNING so the log stays useful. Add more to
:data:`QUIET_LOGGERS` if a new dependency starts spamming.
Public API:
setup_logging(level=INFO, log_file=None) — call once early.
get_logger(name) — module-scoped logger.
enable_debug() — crank spacr.* to DEBUG.
disable_debug() — revert to session level.
log_dir() — folder holding the log.
log_path() — absolute log file path.
"""
from __future__ import annotations
import functools
import logging
import logging.handlers
import os
import sys
import threading
import time
from pathlib import Path
from typing import Any, Callable, Iterable, Optional
DEFAULT_LOG_FILENAME = "spacr.log"
MAX_BYTES = 5 * 1024 * 1024
BACKUP_COUNT = 3
FILE_FORMAT = (
"%(asctime)s [%(levelname)s] %(name)s:%(filename)s:%(lineno)d "
"— %(message)s"
)
STREAM_FORMAT = "%(levelname)s %(name)s: %(message)s"
_NEAR_LIMIT_BYTES = 64 * 1024
def _quicken(handler):
"""Stop a rotating file handler from statting its file on every record.
The standard ``RotatingFileHandler`` checks that the log path is a
regular file and formats each record a second time to see whether it
would cross ``maxBytes``. A run logging thousands of records a second
spent most of its logging time there. After this, those checks run only
once the file is within :data:`_NEAR_LIMIT_BYTES` of the limit, so
rollover happens at the same point for any record shorter than that.
:param handler: a ``RotatingFileHandler``; anything else is returned
unchanged.
:returns: ``handler``.
"""
full_check = getattr(handler, "shouldRollover", None)
if full_check is None or not hasattr(handler, "maxBytes"):
return handler
handler.shouldRollover = functools.partial(
_quick_should_rollover, handler, full_check)
real_flush = handler.flush
handler._spacr_flushed_at = 0.0
handler._spacr_late_flush_pid = 0
handler.flush = functools.partial(_paced_flush, handler, real_flush)
handler.emit = functools.partial(
_emit_flushing_the_loud, handler, handler.emit, real_flush)
return handler
_FLUSH_INTERVAL_S = 0.25
def _in_safe_mode() -> bool:
"""Whether this process runs as ``safespacr``, which starts no thread.
Safe mode flushes every record at once rather than pacing flushes with a
late-flush timer thread.
"""
preferences = sys.modules.get("spacr.qt.preferences")
try:
return bool(preferences is not None and preferences.in_safe_mode())
except Exception:
return False
def _paced_flush(handler, real_flush) -> None:
"""Flush ``handler`` at most every :data:`_FLUSH_INTERVAL_S` seconds.
A run logging thousands of records a second paid one write system
call per record per file. Records between two flushes stay in the
stream's buffer, and one timer per interval writes out whatever the
last of a burst left there, so nothing waits longer than one interval.
Closing the handler still writes everything.
:param handler: the file handler.
:param real_flush: its own ``flush``.
"""
now = time.monotonic()
if _in_safe_mode() or now - handler._spacr_flushed_at >= _FLUSH_INTERVAL_S:
handler._spacr_flushed_at = now
real_flush()
return
pid = os.getpid()
if handler._spacr_late_flush_pid == pid:
return
handler._spacr_late_flush_pid = pid
timer = threading.Timer(_FLUSH_INTERVAL_S, _late_flush,
(handler, real_flush))
timer.daemon = True
timer.start()
def _late_flush(handler, real_flush) -> None:
"""Write out what a burst left in ``handler``'s buffer."""
handler._spacr_late_flush_pid = 0
handler._spacr_flushed_at = time.monotonic()
try:
real_flush()
except Exception:
pass
def _emit_flushing_the_loud(handler, real_emit, real_flush, record) -> None:
"""Write ``record``, and flush at once when it is a warning or worse.
:param handler: the file handler.
:param real_emit: its own ``emit``.
:param real_flush: its own ``flush``.
:param record: the record.
"""
real_emit(record)
if record.levelno >= logging.WARNING:
handler._spacr_flushed_at = time.monotonic()
try:
real_flush()
except Exception:
handler.handleError(record)
def _quick_should_rollover(handler, full_check, record) -> bool:
"""Whether ``record`` would take ``handler``'s file past ``maxBytes``.
The position is read from the byte buffer under the text layer: the
text stream's own ``tell`` flushes, which would undo the paced flush.
The text layer holds at most a few kilobytes more, well inside the
margin.
:param handler: the rotating file handler.
:param full_check: the handler's own ``shouldRollover``.
:param record: the record about to be written.
"""
if handler.maxBytes <= 0:
return False
if handler.stream is None:
handler.stream = handler._open()
stream = handler.stream
try:
position = stream.buffer.tell()
except (AttributeError, OSError, ValueError):
position = stream.tell()
if position + _NEAR_LIMIT_BYTES < handler.maxBytes:
return False
return full_check(record)
class _SecondFormatter(logging.Formatter):
"""A ``logging.Formatter`` that renders each second's timestamp once.
``formatTime`` converts and formats the clock for every record; a run
logging thousands of records a second asks for the same second
thousands of times.
"""
_second = (None, "")
def formatTime(self, record, datefmt=None):
"""The record's time, as ``logging.Formatter.formatTime`` gives it.
:param record: the record.
:param datefmt: a ``strftime`` format, or ``None`` for the default
with milliseconds.
:returns: the formatted time.
"""
key = (int(record.created), datefmt)
cached_key, text = self._second
if key != cached_key:
text = time.strftime(datefmt or self.default_time_format,
self.converter(record.created))
self._second = (key, text)
if datefmt or not self.default_msec_format:
return text
return self.default_msec_format % (text, record.msecs)
class _CompactTraceFormat(_SecondFormatter):
"""The ordinary format, except for `spacr.trace`, which gets a short one.
MEASURED (297): twenty calls to a no-argument function wrote 5,520 bytes
-- 276 a call, of which the message itself is about forty. The rest is a
prefix repeated on every line, and three of its fields say nothing here:
the level is always DEBUG, the logger is always `spacr.trace`, and the
file and line are always this module's own tracer rather than the code
being traced, which is actively misleading.
A trail nobody can read through is not a trail. What is left is the time
and the arrow, which is what the reader is actually following.
"""
#: Time only, to the millisecond: a trace is read for ORDER and for where
#: a gap is, and the date is the same on every line of one run.
TRACE_FORMAT = "%(asctime)s %(message)s"
def __init__(self, fmt: str, datefmt: Optional[str] = None):
"""Create ordinary and compact trace-record formatters.
:param fmt: format string used for every non-trace record.
:param datefmt: optional date format used by the ordinary formatter.
"""
super().__init__(fmt, datefmt)
self._trace = logging.Formatter(self.TRACE_FORMAT, "%H:%M:%S")
def format(self, record: logging.LogRecord) -> str:
"""Render one record with the compact form reserved for trace events.
:param record: logging record to render.
The ordinary text is kept on the record, so the master log and the
per-level file, which share this format, format a record once.
:returns: time-and-message text for ``spacr.trace``, otherwise the
ordinary configured format.
"""
if record.name == "spacr.trace":
return self._trace.format(record)
key = (self._fmt, self.datefmt)
cached = record.__dict__.get("_spacr_file_text")
if cached is not None and cached[0] == key:
return cached[1]
text = super().format(record)
record._spacr_file_text = (key, text)
return text
#: Third-party loggers that spam INFO records — capped at WARNING.
QUIET_LOGGERS: tuple[str, ...] = (
"PIL",
"matplotlib",
"fontTools",
"urllib3",
"asyncio",
"torch",
"torchvision",
"cellpose",
"tensorflow",
"botocore",
"numba",
"h5py",
)
#: Loggers quieted to ERROR rather than WARNING.
#:
#: :data:`QUIET_LOGGERS` sets WARNING, which is right for a library whose
#: warnings a user can act on. These emit CAPABILITY NOTICES at warning level
#: -- "this optional extra is not installed" -- on every import, for extras
#: spaCR does not use. `cellpose.vit` announces that CPDINO is unavailable
#: every time a module screen opens, and it was reported against Mask, Measure
#: and Map Barcodes as three separate bugs before it was recognised as one
#: line printed everywhere.
#:
#: ERROR rather than CRITICAL: a genuine cellpose failure still reaches the
#: user. What is dropped is the advertisement.
SILENT_LOGGERS: tuple[str, ...] = (
"cellpose.vit",
)
#: The five levels the user can switch on and off, lowest first.
LEVELS: tuple[int, ...] = (
logging.DEBUG, logging.INFO, logging.WARNING,
logging.ERROR, logging.CRITICAL,
)
#: Per-level file names, alongside the master :data:`DEFAULT_LOG_FILENAME`.
LEVEL_LOG_FILENAMES: dict = {
logging.DEBUG: "spacr-debug.log",
logging.INFO: "spacr-info.log",
logging.WARNING: "spacr-warning.log",
logging.ERROR: "spacr-error.log",
logging.CRITICAL: "spacr-critical.log",
}
#: What a fresh install writes and shows. The file keeps everything from INFO
#: up, so a bug report is useful without being asked for; the console shows
#: only what went wrong, because it is a panel the user is reading while
#: working rather than a transcript.
DEFAULT_FILE_LEVELS: frozenset = frozenset(
{logging.INFO, logging.WARNING, logging.ERROR, logging.CRITICAL})
DEFAULT_CONSOLE_LEVELS: frozenset = frozenset(
{logging.WARNING, logging.ERROR, logging.CRITICAL})
[docs]
class LevelSetFilter(logging.Filter):
"""Pass only records whose level is in an explicitly enabled set.
``setLevel`` is a *threshold*: enabling DEBUG necessarily enables
everything above it. The preference this serves is a set of independent
switches, where DEBUG on with INFO off is a legitimate choice, so the
gate has to be membership rather than comparison.
The set is mutable in place so a live handler can be re-gated from the
Preferences dialog without being torn down and rebuilt, which would race
against any thread logging at that moment.
:param levels: the ``logging`` level numbers to pass. Empty -- the
default -- passes NOTHING, which is deliberate: a filter that let
everything through until configured would leak records during the
window before Preferences applies its choice.
"""
def __init__(self, levels: Iterable[int] = ()) -> None:
"""Create an independently switchable logging-level filter.
:param levels: exact numeric levels to pass; an empty iterable passes
no records.
"""
super().__init__()
self.levels = set(levels)
[docs]
def filter(self, record: logging.LogRecord) -> bool:
"""Return whether *record* has one of the enabled exact levels.
:param record: logging record presented by a handler.
:returns: ``True`` only when ``record.levelno`` is enabled.
"""
return record.levelno in self.levels
[docs]
def normalise_levels(levels: Iterable[int]) -> frozenset:
"""Keep only the five switchable levels, discarding anything else.
:param levels: numeric logging levels (anything ``int()`` accepts); values
other than DEBUG, INFO, WARNING, ERROR and CRITICAL are dropped.
"""
return frozenset(int(level) for level in levels if int(level) in LEVELS)
[docs]
def clamp_console_to_file(console: Iterable[int],
file_levels: Iterable[int]) -> frozenset:
"""Console levels are a subset of what the file records.
A level the log file discards cannot reach the console, because the
console is fed from the same records. Showing a user a line they will
not find in the log file they are about to attach to a bug report is
worse than not showing it.
:param console: numeric logging levels requested for the console.
:param file_levels: numeric logging levels the log file records; only
levels in both sets are kept.
"""
return normalise_levels(console) & normalise_levels(file_levels)
_INITIALISED: bool = False
_SESSION_LEVEL: int = logging.INFO
_LOG_PATH: Optional[Path] = None
_FILE_FILTER: Optional[LevelSetFilter] = None
_LEVEL_HANDLERS: dict = {}
_TRACE_ROOT = os.path.realpath(os.path.dirname(__file__)) + os.sep
_TRACE_THIS_FILE = os.path.realpath(__file__)
_TRACE_ENABLED: bool = False
_TRACE_STATE = threading.local()
_PREVIOUS_SYS_PROFILE = None
_PREVIOUS_THREAD_PROFILE = None
#: Qt virtual-method overrides the trace must never fire on.
#:
#: These are not application logic. Qt calls them once per delivered event,
#: thousands of times a second, and the GUI console is one of the sinks the
#: resulting records go to -- which closes a loop through the event queue:
#: delivering an event logs, logging writes a widget, writing a widget posts
#: a repaint, delivering the repaint logs again. Measured as a Qt shard that
#: made no forward progress at 100% CPU for twenty-five minutes with the GUI
#: thread parked in ``spacr.qt.button_roles.eventFilter -> _trace_profile ->
#: ConsolePanel.append_stdout``, and the same loop is reachable in the shipped
#: app the moment "Verbose logging" is switched on.
#:
#: Excluding them costs nothing worth having. The hook exists to say which
#: spaCR function a run went through, and ``paintEvent`` is not that. It is
#: also what :func:`_trace_profile`'s own contract demands -- a tracing aid
#: must never alter the code it observes, and one that stops event delivery
#: keeping up has altered it beyond recognition.
_TRACE_SKIP_NAMES = frozenset({
"event", "eventFilter", "customEvent", "childEvent", "timerEvent",
"paintEvent", "resizeEvent", "moveEvent", "showEvent", "hideEvent",
"closeEvent", "changeEvent", "enterEvent", "leaveEvent", "focusInEvent",
"focusOutEvent", "wheelEvent", "mouseMoveEvent", "mousePressEvent",
"mouseReleaseEvent", "mouseDoubleClickEvent", "hoverMoveEvent",
"hoverEnterEvent", "hoverLeaveEvent", "keyPressEvent", "keyReleaseEvent",
"dragEnterEvent", "dragMoveEvent", "dragLeaveEvent", "dropEvent",
"contextMenuEvent", "viewportEvent", "sizeHint", "minimumSizeHint",
"heightForWidth",
})
#: Modules whose functions run on the ANIMATION TIMER, and are never traced.
#:
#: With verbose logging on, animation tracing can write three 5 MB log files
#: inside one minute and make the interface unusable. The backdrop
#: shades a frame up to sixty times a second and each frame calls these
#: helpers hundreds of times -- `ambient._with_alpha` entered and left, each
#: one formatted and written to disk. Nothing about that trail is diagnostic;
#: it is the same handful of names repeating until the log rotates and buries
#: whatever the user was actually trying to catch.
#:
#: `_TRACE_SKIP_NAMES` cannot cover this. It names `paintEvent`, but the cost
#: is in the ordinary helpers the paint calls, which have unremarkable names
#: and are indistinguishable from any other function by name alone. The
#: module is the thing they have in common.
_TRACE_SKIP_MODULES = (
"spacr.qt.widgets.ambient",
"spacr.qt.widgets.fractal_travel",
"spacr.qt.widgets.fractal_cascade",
"spacr.qt.widgets.fractal_space",
)
_PORTABLE_ENV = "SPACR_PORTABLE"
_PORTABLE_MARKER = "spacr-portable"
_PORTABLE_DATA = "spacr-data"
_PORTABLE_ON = frozenset({"1", "true", "yes", "on"})
_PORTABLE_OFF = frozenset({"0", "false", "no", "off"})
def _app_folders() -> list:
"""Return the folders that count as "next to the app", nearest first.
They are ``SPACR_LAUNCHER_DIR`` when an installer set it, the folder
holding the Python executable, the environment's prefix and the folder
that contains that environment, which is the install folder of the
Windows, macOS and Linux installers.
"""
found = []
launcher = os.environ.get("SPACR_LAUNCHER_DIR", "").strip()
candidates = [Path(launcher).expanduser()] if launcher else []
candidates += [Path(os.path.abspath(sys.executable)).parent,
Path(sys.prefix), Path(sys.prefix).parent]
for folder in candidates:
if folder not in found:
found.append(folder)
return found
def _portable_root() -> Optional[Path]:
"""Return the folder portable mode keeps its data beside, or ``None``.
``SPACR_PORTABLE`` decides first: ``0``/``off`` turns portable mode off
even when a marker exists, a folder path turns it on in that folder, and
``1``/``on`` turns it on next to the app. Otherwise portable mode is on
when an empty ``spacr-portable`` marker file sits in one of
:func:`_app_folders`, and it is off by default. The answer is remembered
per process for each value of the two variables, so a marker created
while spaCR runs takes effect at the next start.
"""
return _portable_root_for(
os.environ.get(_PORTABLE_ENV, "").strip(),
os.environ.get("SPACR_LAUNCHER_DIR", "").strip())
@functools.lru_cache(maxsize=16)
def _portable_root_for(raw: str, launcher: str) -> Optional[Path]:
"""Resolve :func:`_portable_root` for one ``SPACR_PORTABLE`` value."""
if raw.lower() in _PORTABLE_OFF:
return None
if raw and raw.lower() not in _PORTABLE_ON:
return Path(raw).expanduser().absolute()
folders = _app_folders()
for folder in folders:
try:
if (folder / _PORTABLE_MARKER).is_file():
return folder
except OSError:
continue
if raw:
return folders[0]
return None
def _spacr_home() -> Path:
"""Return the folder spaCR keeps settings, caches, logs and runs in.
``SPACR_HOME`` wins when set. In portable mode it is the ``spacr-data``
folder beside the app; otherwise ``~/.spacr``. Every per-user spaCR
folder (runs, logs, backends, plugins, models, recipes, macros) lives
under it.
"""
configured = os.environ.get("SPACR_HOME", "").strip()
if configured:
return Path(configured).expanduser()
root = _portable_root()
if root is not None:
return root / _PORTABLE_DATA
return Path.home() / ".spacr"
def _apply_portable_mode() -> Optional[Path]:
"""Point every per-user folder at the portable data folder, if enabled.
Sets ``SPACR_HOME``, ``SPACR_LOG_DIR``, ``SPACR_BACKENDS_DIR`` and
``SPACR_PLUGIN_HOME`` and the cache variables of the libraries spaCR
loads (``XDG_CACHE_HOME``, ``TORCH_HOME``, ``HF_HOME``,
``MPLCONFIGDIR``, ``CELLPOSE_LOCAL_MODELS_PATH``,
``XDG_STATE_HOME``) beneath it, so child
processes and backend workers follow. A variable the user already set is
kept. Does nothing when portable mode is off.
:returns: the data folder, or ``None`` when portable mode is off.
"""
root = _portable_root()
if root is None:
return None
data = Path(os.environ.get("SPACR_HOME", "").strip()
or (root / _PORTABLE_DATA)).expanduser()
cache = data / "cache"
for name, value in (
("SPACR_HOME", data),
("SPACR_LOG_DIR", data / "logs"),
("SPACR_BACKENDS_DIR", data / "backends"),
("SPACR_PLUGIN_HOME", data / "plugins"),
("XDG_CACHE_HOME", cache),
("XDG_STATE_HOME", data / "state"),
("TORCH_HOME", cache / "torch"),
("HF_HOME", cache / "huggingface"),
("MPLCONFIGDIR", cache / "matplotlib"),
("CELLPOSE_LOCAL_MODELS_PATH", data / "models" / "cellpose"),
):
os.environ.setdefault(name, str(value))
try:
data.mkdir(parents=True, exist_ok=True)
except OSError:
pass
return data
def _portable_settings_dir() -> Optional[Path]:
"""Return the folder Qt settings are kept in when portable, else ``None``."""
if _portable_root() is None:
return None
return _spacr_home() / "settings"
[docs]
def log_dir() -> Path:
"""Return the folder where spacr log files live.
``SPACR_LOG_DIR`` overrides the default, which is useful for portable
installs, read-only home directories and test/embedding hosts.
:returns: the configured directory, otherwise ``~/.spacr/logs``.
"""
override = os.environ.get("SPACR_LOG_DIR", "").strip()
root = Path(override).expanduser() if override else (
_spacr_home() / "logs")
root.mkdir(parents=True, exist_ok=True)
return root
[docs]
def log_path() -> Path:
"""Return the absolute path of the rotating log file.
Uses whatever was passed to :func:`setup_logging` last, or the
default under :func:`log_dir` when never set.
"""
return _LOG_PATH if _LOG_PATH is not None else (
log_dir() / DEFAULT_LOG_FILENAME
)
[docs]
def setup_logging(level: Optional[int] = None,
log_file: Optional[Path] = None,
stream: bool = False,
quiet: Iterable[str] = QUIET_LOGGERS) -> Path:
"""Install the rotating file handler on the root logger.
Idempotent — subsequent calls only re-apply the level, they don't
stack additional handlers. Honours the ``SPACR_LOG_LEVEL``
environment variable when ``level`` is not given.
:param level: minimum record level for the log file. Defaults to
``SPACR_LOG_LEVEL`` env var (any of ``DEBUG``/``INFO``/…) or
:data:`logging.INFO`.
:param log_file: override for where the file lands. Defaults to
:func:`log_path`.
:param stream: also attach a StreamHandler to stderr — handy for
headless / CI runs where the log file isn't inspected.
:param quiet: iterable of logger names to pin at WARNING. Defaults
to :data:`QUIET_LOGGERS`.
:returns: the resolved log-file path.
"""
global _INITIALISED, _SESSION_LEVEL, _LOG_PATH
if level is None:
level = _env_level()
_SESSION_LEVEL = level
resolved_path = Path(log_file) if log_file else log_path()
resolved_path.parent.mkdir(parents=True, exist_ok=True)
_LOG_PATH = resolved_path
if _INITIALISED:
logging.getLogger().setLevel(_third_party_level(level))
logging.getLogger("spacr").setLevel(level)
return resolved_path
root = logging.getLogger()
root.setLevel(_third_party_level(level))
logging.getLogger("spacr").setLevel(level)
try:
file_h = _quicken(logging.handlers.RotatingFileHandler(
resolved_path,
maxBytes=MAX_BYTES,
backupCount=BACKUP_COUNT,
encoding="utf-8",
))
except OSError as exc:
sys.stderr.write(
f"spaCR could not open diagnostic log {resolved_path}: {exc}\n")
file_h = None
stream = True
if file_h is not None:
file_h.setLevel(logging.DEBUG)
file_h.addFilter(_file_filter(_levels_at_or_above(level)))
file_h.setFormatter(_CompactTraceFormat(FILE_FORMAT))
root.addHandler(file_h)
_install_level_handlers(resolved_path, _levels_at_or_above(level))
if stream:
stream_h = logging.StreamHandler()
stream_h.setLevel(level)
stream_h.setFormatter(logging.Formatter(STREAM_FORMAT))
root.addHandler(stream_h)
for name in quiet:
logging.getLogger(name).setLevel(logging.WARNING)
for name in SILENT_LOGGERS:
logging.getLogger(name).setLevel(logging.ERROR)
_INITIALISED = True
get_logger("spacr").info("logging initialised → %s", resolved_path)
return resolved_path
def _levels_at_or_above(level: int) -> frozenset:
"""The switch set equivalent to a classic threshold, for first setup."""
return frozenset(item for item in LEVELS if item >= level)
def _file_filter(levels: Iterable[int]) -> LevelSetFilter:
"""The one filter shared by the master log file, created on first use."""
global _FILE_FILTER
if _FILE_FILTER is None:
_FILE_FILTER = LevelSetFilter(levels)
else:
_FILE_FILTER.levels = set(normalise_levels(levels))
return _FILE_FILTER
def _install_level_handlers(master_path: Path, levels: Iterable[int]) -> None:
"""Give every level its own file beside the master log.
One file per level answers "show me only the errors" without grep, and
the master keeps the interleaved order that makes a sequence of events
readable. Each is rotated on the same terms as the master.
A level that is switched off keeps its handler, filtered to nothing,
rather than being detached: attaching and detaching handlers on a live
root logger races with any thread that is logging, and an idle handler
costs one open file. Each handler's threshold is its own level, so a
record never reaches the files of the levels above it.
"""
enabled = normalise_levels(levels)
root = logging.getLogger()
for level in LEVELS:
handler = _LEVEL_HANDLERS.get(level)
if handler is None:
path = master_path.parent / LEVEL_LOG_FILENAMES[level]
try:
handler = _quicken(logging.handlers.RotatingFileHandler(
path, maxBytes=MAX_BYTES, backupCount=BACKUP_COUNT,
encoding="utf-8"))
except OSError as exc:
sys.stderr.write(
f"spaCR could not open {path}: {exc}\n")
continue
handler.setLevel(level)
handler.setFormatter(_CompactTraceFormat(FILE_FORMAT))
handler.addFilter(LevelSetFilter())
root.addHandler(handler)
_LEVEL_HANDLERS[level] = handler
for existing in handler.filters:
if isinstance(existing, LevelSetFilter):
existing.levels = {level} if level in enabled else set()
[docs]
def apply_level_policy(file_levels: Iterable[int],
console_levels: Iterable[int] = ()) -> tuple:
"""Re-gate the live handlers from the Preferences switches.
:param file_levels: levels written to the log files.
:param console_levels: levels echoed to the in-app console; silently
clamped to a subset of ``file_levels``.
:returns: ``(file_levels, console_levels)`` as actually applied.
"""
files = normalise_levels(file_levels)
console = clamp_console_to_file(console_levels, files)
_file_filter(files)
if _LOG_PATH is not None:
_install_level_handlers(_LOG_PATH, files)
lowest = min(files) if files else logging.CRITICAL
logging.getLogger("spacr").setLevel(lowest)
logging.getLogger().setLevel(_third_party_level(lowest))
try:
from .qt.verbose_logger import apply_console_levels
except Exception:
pass
else:
apply_console_levels(console)
return files, console
def _third_party_level(level: int) -> int:
"""The root logger's level: never below WARNING.
Every library logger that sets no level of its own inherits the root's.
The root used to sit at DEBUG so spaCR's own DEBUG could pass, and with
Verbose logging on the log files then took DEBUG from every library
in the process -- an HTTP client's per-request trace, a model
downloader's lock chatter -- 2100 lines for 300 calls in a measured
run, burying the spaCR records the log exists for. spaCR's loggers
carry their own level on ``spacr``, so the root can stay at WARNING
without hiding any of them.
:param level: the lowest level the user asked spaCR to record.
:returns: ``level`` or WARNING, whichever is higher.
"""
return max(int(level), logging.WARNING)
def _env_level() -> int:
"""Read ``SPACR_LOG_LEVEL`` from env; fall back to INFO."""
raw = os.environ.get("SPACR_LOG_LEVEL", "").upper().strip()
if raw and hasattr(logging, raw):
return getattr(logging, raw)
return logging.INFO
[docs]
def get_logger(name: str) -> logging.Logger:
"""Return a spacr-scoped :class:`logging.Logger`.
Idiomatic usage from any module:
.. code-block:: python
from spacr.logging_util import get_logger
LOG = get_logger(__name__)
:param name: logger name — typically ``__name__`` so the log
stream shows which module the record came from.
"""
return logging.getLogger(name)
[docs]
def enable_debug() -> None:
"""Crank every ``spacr.*`` logger to DEBUG.
Useful when debugging interactively:
.. code-block:: pycon
>>> from spacr.logging_util import enable_debug
>>> enable_debug()
Third-party loggers listed in :data:`QUIET_LOGGERS` are left at
WARNING to keep the log readable.
"""
logging.getLogger("spacr").setLevel(logging.DEBUG)
for h in logging.getLogger().handlers:
h.setLevel(logging.DEBUG)
[docs]
def disable_debug() -> None:
"""Revert every ``spacr.*`` logger to the level chosen at setup.
Inverse of :func:`enable_debug`.
"""
logging.getLogger("spacr").setLevel(_SESSION_LEVEL)
for h in logging.getLogger().handlers:
h.setLevel(_SESSION_LEVEL)
[docs]
def function_trace_enabled() -> bool:
"""Return whether spaCR function-level DEBUG tracing is active.
The trace is controlled by :func:`enable_function_trace` and
:func:`disable_function_trace`. The GUI's *Verbose logging* preference
does not install it. It never records arguments or return values, which
avoids copying large arrays and keeps API keys or filesystem metadata out
of the diagnostic log.
"""
return _TRACE_ENABLED
#: Resolved source paths, keyed by the ``co_filename`` they came from.
#:
#: WHY THIS EXISTS. :func:`_trace_one_event` runs on EVERY Python call and
#: return in the process while the function trace is on, and it used to call
#: ``os.path.realpath`` on each one. That is not a string operation: it
#: resolves every component of the path against the filesystem, and it
#: measured 11,739 ns per call on this machine against 45 ns for a dict hit
#: -- 261x, on a step taken twice per traced function.
#:
#: At a conservative ten thousand calls a second in a Qt application that is
#: roughly a quarter of a core spent resolving the same few hundred paths
#: over and over, which is what kept the function trace too expensive to
#: leave switched on.
#:
#: UNBOUNDED ON PURPOSE, and safe: the key space is the set of Python source
#: files the process actually executes, which is a few hundred, fixed after
#: import, and already all resident in ``sys.modules``. A bounded cache would
#: add an eviction policy to guard a dictionary that cannot meaningfully
#: grow.
#:
#: No lock. Two threads racing compute the same value and store it twice,
#: and ``dict`` assignment is atomic; a lock here would serialise every
#: traced call in the process to protect against writing the same string.
_TRACE_REALPATH_CACHE: dict = {}
def _traced_realpath(co_filename: str) -> str:
"""``os.path.realpath(co_filename)``, resolved once per distinct path.
:param co_filename: a code object's ``co_filename``, as the profile hook
receives it.
:returns: the resolved absolute path, from cache after the first call.
"""
resolved = _TRACE_REALPATH_CACHE.get(co_filename)
if resolved is None:
resolved = os.path.realpath(co_filename)
_TRACE_REALPATH_CACHE[co_filename] = resolved
return resolved
def _trace_profile(frame, event, arg):
"""Profile-hook implementation used by :func:`enable_function_trace`.
Only Python ``call`` and ``return`` events for files inside the installed
:mod:`spacr` package are recorded. The logger implementation itself is
excluded to prevent recursion, and so are Qt's event-delivery overrides
(:data:`_TRACE_SKIP_NAMES`) -- tracing those feeds the GUI console from
inside event delivery, and the console's repaint is another event.
"""
if event not in {"call", "return"}:
return
try:
return _trace_one_event(frame, event)
except BaseException: # noqa: BLE001
return None
def _trace_one_event(frame, event):
"""The body of :func:`_trace_profile`, minus its shutdown guard."""
if _TRACE_SKIP_NAMES is None or logging is None:
return None
if frame.f_code.co_name in _TRACE_SKIP_NAMES:
return None
if not logging.getLogger("spacr.trace").isEnabledFor(logging.DEBUG):
return
module = frame.f_globals.get("__name__", "spacr")
if _TRACE_SKIP_MODULES and module.startswith(_TRACE_SKIP_MODULES):
return
filename = _traced_realpath(frame.f_code.co_filename)
if not filename.startswith(_TRACE_ROOT) or filename == _TRACE_THIS_FILE:
return
if getattr(_TRACE_STATE, "busy", False):
return
_TRACE_STATE.busy = True
try:
qualname = getattr(
frame.f_code, "co_qualname", frame.f_code.co_name)
marker = "→" if event == "call" else "←"
logging.getLogger("spacr.trace").debug(
"%s %s.%s", marker, module, qualname)
except Exception:
pass
finally:
_TRACE_STATE.busy = False
[docs]
def enable_function_trace() -> None:
"""Trace every spaCR Python function and method at DEBUG level.
The hook is installed for the calling thread, all future Python threads,
and—on Python 3.12+—threads that already exist. Calls outside the spaCR
package are ignored. Repeated calls are idempotent.
This is intentionally verbose and has measurable overhead. With it
installed, the Home screen took 7.4 s to become usable, against 4.0-4.2 s
without it. Nothing in the GUI installs it, and the *Verbose logging*
preference does not either. Normal operation has no profile hook
installed.
"""
global _TRACE_ENABLED, _PREVIOUS_SYS_PROFILE, _PREVIOUS_THREAD_PROFILE
if _TRACE_ENABLED:
return
_PREVIOUS_SYS_PROFILE = sys.getprofile()
get_thread_profile = getattr(threading, "getprofile", None)
_PREVIOUS_THREAD_PROFILE = (
get_thread_profile() if get_thread_profile is not None else None)
_TRACE_ENABLED = True
set_all = getattr(threading, "setprofile_all_threads", None)
if set_all is not None:
set_all(_trace_profile)
else:
sys.setprofile(_trace_profile)
threading.setprofile(_trace_profile)
[docs]
def disable_function_trace() -> None:
"""Remove spaCR's function trace and restore prior profile hooks."""
global _TRACE_ENABLED, _PREVIOUS_SYS_PROFILE, _PREVIOUS_THREAD_PROFILE
if not _TRACE_ENABLED:
return
_TRACE_ENABLED = False
set_all = getattr(threading, "setprofile_all_threads", None)
if set_all is not None:
set_all(_PREVIOUS_THREAD_PROFILE)
sys.setprofile(_PREVIOUS_SYS_PROFILE)
else:
threading.setprofile(_PREVIOUS_THREAD_PROFILE)
sys.setprofile(_PREVIOUS_SYS_PROFILE)
_PREVIOUS_SYS_PROFILE = None
_PREVIOUS_THREAD_PROFILE = None
import functools
import time
TimingCallable = Callable[..., Any]
_TIMING_ENABLED: bool = True
_TIMING_THRESHOLD_MS: int = int(
os.environ.get("SPACR_TIME_THRESHOLD_MS", "5")
)
[docs]
def enable_timing() -> None:
"""Turn on the ``@timed`` decorator + :class:`Timer` context manager.
Enabled by default. Call this after :func:`disable_timing` to
re-enable timing logs at runtime.
"""
global _TIMING_ENABLED
_TIMING_ENABLED = True
[docs]
def disable_timing() -> None:
"""Turn off all timing logs — the decorators become pass-through."""
global _TIMING_ENABLED
_TIMING_ENABLED = False
[docs]
def set_timing_threshold_ms(ms: int) -> None:
"""Only log timings that exceed ``ms`` milliseconds.
Defaults to 5 ms (env-overridable via ``SPACR_TIME_THRESHOLD_MS``).
Setting to 0 logs every call.
:param ms: threshold in milliseconds, converted with ``int()``; negative
values are clamped to 0.
"""
global _TIMING_THRESHOLD_MS
_TIMING_THRESHOLD_MS = max(0, int(ms))
[docs]
def timed(fn: Optional[TimingCallable] = None, *,
name: Optional[str] = None,
level: int = logging.INFO) -> TimingCallable:
"""Decorator that logs wall-clock time for every call to ``fn``.
Usable as ``@timed`` or ``@timed(name="…", level=DEBUG)``::
@timed
def preprocess_generate_masks(settings):
...
@timed(name="cellpose.batch", level=logging.DEBUG)
def _run_batch(...):
...
Log line format: ``func_name took 1234.5 ms``. The logger name is
``spacr.timing`` unless the wrapped function's module starts with
``spacr.``, in which case that module's logger is reused so
timings interleave with the function's own log records.
:param fn: the function to wrap (when used as ``@timed``).
:param name: override the label used in the log line.
:param level: logging level for the "took Xms" line.
"""
def _decorate(inner: TimingCallable) -> TimingCallable:
"""Wrap ``inner`` with the selected label, logger, and timing marks."""
label = name or f"{inner.__module__}.{inner.__qualname__}"
mod = inner.__module__
log_name = mod if mod.startswith("spacr") else "spacr.timing"
@functools.wraps(inner)
def _wrapped(*args, **kwargs):
"""Call ``inner`` and log qualifying elapsed time, even on error."""
if not _TIMING_ENABLED:
return inner(*args, **kwargs)
log = logging.getLogger(log_name)
t0 = time.perf_counter()
try:
return inner(*args, **kwargs)
finally:
elapsed_ms = (time.perf_counter() - t0) * 1000.0
if elapsed_ms >= _TIMING_THRESHOLD_MS:
log.log(level, "%s took %.1f ms", label, elapsed_ms)
_wrapped.__wrapped__ = inner # type: ignore[attr-defined]
_wrapped.__spacr_timed__ = True # type: ignore[attr-defined]
return _wrapped
if fn is not None:
return _decorate(fn)
return _decorate
[docs]
class Timer:
"""Context manager that logs the elapsed wall-clock for a block.
Example::
from spacr.logging_util import Timer
with Timer("preprocess field"):
do_work()
# -> logs "preprocess field took 12.3 ms"
Nested Timers work fine — each logs its own block independently.
:param label: human-readable name shown in the log line.
:param logger: name of the logger to write to (default
``"spacr.timing"``).
:param level: logging level for the line (default INFO).
:ivar elapsed_ms: filled on exit; None while still running.
"""
def __init__(self, label: str,
logger: str = "spacr.timing",
level: int = logging.INFO):
"""Create an idle timer with its logging destination.
:param label: human-readable operation name included in timing records.
:param logger: logger name that receives completed timings.
:param level: logging level used for completed timings.
"""
self.label = label
self._logger_name = logger
self._level = level
self._t0: Optional[float] = None
self.elapsed_ms: Optional[float] = None
[docs]
def __enter__(self) -> "Timer":
"""Start or restart timing and clear any previous duration.
:returns: this timer, with :attr:`elapsed_ms` unset while it runs.
"""
self.elapsed_ms = None
self._t0 = time.perf_counter()
return self
[docs]
def __exit__(self, exc_type, exc, tb) -> None:
"""Finish one active interval without suppressing body exceptions.
:param exc_type: exception type raised by the block, when present.
:param exc: exception instance raised by the block, when present.
:param tb: traceback raised by the block, when present.
:returns: ``None`` so any body exception continues to propagate.
"""
started = self._t0
self._t0 = None
if started is None:
return
self.elapsed_ms = (time.perf_counter() - started) * 1000.0
if (_TIMING_ENABLED
and self.elapsed_ms >= _TIMING_THRESHOLD_MS):
logging.getLogger(self._logger_name).log(
self._level, "%s took %.1f ms",
self.label, self.elapsed_ms,
)
[docs]
def time_module(module, exclude: tuple = ()) -> int:
"""Wrap every public function in ``module`` with :func:`timed`.
Idempotent — functions that already carry ``__spacr_timed__`` are
skipped. Handy during ad-hoc profiling; not recommended for
production imports because it slows every call by ~1 µs.
Example::
import spacr.core
from spacr.logging_util import time_module
time_module(spacr.core)
# Every public function on spacr.core now logs its timing.
:param module: the module object to wrap.
:param exclude: iterable of function names to skip.
:returns: how many functions were wrapped.
"""
wrapped = 0
for name in dir(module):
if name.startswith("_") or name in exclude:
continue
obj = getattr(module, name)
if not callable(obj):
continue
if getattr(obj, "__spacr_timed__", False):
continue
if getattr(obj, "__module__", None) != module.__name__:
continue
setattr(module, name, timed(obj))
wrapped += 1
return wrapped