Repository navigation
Expand file tree
/
Copy pathlogging.py
More file actions
808 lines (680 loc) Β· 25.1 KB
/
Copy pathlogging.py
File metadata and controls
808 lines (680 loc) Β· 25.1 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
# SPDX-FileCopyrightText: 2025 EasyScience contributors <https://github.com/easyscience>
# SPDX-License-Identifier: BSD-3-Clause
"""
Project-wide logging utilities built on top of Rich.
Provides a shared Rich console, a compact/verbose logger with consistent
formatting, Jupyter traceback handling, and a small printing faΓ§ade
tailored to the configured console.
"""
from __future__ import annotations
import logging
import os
import shutil
import warnings
from contextlib import suppress
from enum import Enum
from enum import IntEnum
from enum import auto
from typing import TYPE_CHECKING
from typing import ClassVar
if TYPE_CHECKING: # pragma: no cover
from types import TracebackType
import html
import re
import sys
from pathlib import Path
from rich import traceback
from rich.console import Console
from rich.console import Group
from rich.console import RenderableType
from rich.logging import RichHandler
from rich.markup import MarkupError
from rich.text import Text
from easydiffraction.utils.environment import in_jupyter
from easydiffraction.utils.environment import in_pytest
from easydiffraction.utils.environment import in_warp
CONSOLE_PARAGRAPH_STYLE = 'bold deep_sky_blue3'
# ======================================================================
# HANDLERS
# ======================================================================
class IconifiedRichHandler(RichHandler):
"""RichHandler using icons (compact) or names (verbose)."""
_icons: ClassVar[dict] = {
logging.CRITICAL: 'π',
logging.ERROR: 'β',
logging.WARNING: 'β οΈ',
logging.DEBUG: 'βοΈ',
logging.INFO: 'βΉοΈ', # noqa: RUF001
}
def __init__(
self,
*args: object,
mode: str = 'compact',
**kwargs: object,
) -> None:
super().__init__(*args, **kwargs)
self.mode = mode
def get_level_text(self, record: logging.LogRecord) -> Text:
"""
Return an icon or level name for the log record.
Parameters
----------
record : logging.LogRecord
The log record being rendered.
Returns
-------
Text
A Rich Text object with the level indicator.
"""
if self.mode == 'compact':
icon = self._icons.get(record.levelno, record.levelname)
if in_warp() and not in_jupyter() and icon in {'β οΈ', 'βοΈ', 'βΉοΈ'}: # noqa: RUF001
icon += ' ' # add space to align with two-char icons
return Text(icon)
# Use RichHandler's default level text for verbose mode
return super().get_level_text(record)
def render_message(
self,
record: logging.LogRecord,
message: str,
) -> Text:
"""
Render the log message body as a Rich Text object.
Parameters
----------
record : logging.LogRecord
The log record being rendered.
message : str
Pre-formatted log message string.
Returns
-------
Text
A Rich Text object with the rendered message.
"""
if self.mode == 'compact':
try:
return Text.from_markup(message)
except (ValueError, KeyError, TypeError, MarkupError):
return Text(str(message))
return super().render_message(record, message)
# ======================================================================
# CONSOLE MANAGER
# ======================================================================
class ConsoleManager:
"""Central provider for shared Rich Console instance."""
_MIN_CONSOLE_WIDTH = 130
_instance: Console | None = None
@staticmethod
def _detect_width() -> int:
"""
Detect a suitable console width for the shared Console.
Returns
-------
int
The detected terminal width, clamped at
``_MIN_CONSOLE_WIDTH`` to avoid cramped layouts.
"""
min_width = ConsoleManager._MIN_CONSOLE_WIDTH
try:
width = shutil.get_terminal_size().columns
except (ValueError, OSError):
width = min_width
return max(width, min_width)
@classmethod
def get(cls) -> Console:
"""Return a shared Rich Console instance."""
if cls._instance is None:
cls._instance = Console(
width=cls._detect_width(),
force_jupyter=False,
)
return cls._instance
# ======================================================================
# LOGGER CONFIGURATION HELPERS
# ======================================================================
class LoggerConfig:
"""Facade for logger configuration, delegates to helpers."""
@staticmethod
def setup_handlers(
logger: logging.Logger,
*,
level: int,
rich_tracebacks: bool,
mode: str = 'compact',
) -> None:
"""
Install Rich handler and optional Jupyter traceback support.
Parameters
----------
logger : logging.Logger
Logger instance to attach handlers to.
level : int
Minimum log level to emit.
rich_tracebacks : bool
Whether to enable Rich tracebacks.
mode : str, default='compact'
Output mode name ("compact" or "verbose").
"""
logger.handlers.clear()
logger.propagate = False
logger.setLevel(level)
if in_jupyter():
traceback.install(
show_locals=False,
suppress=['easydiffraction'],
)
console = ConsoleManager.get()
handler = IconifiedRichHandler(
rich_tracebacks=rich_tracebacks,
markup=True,
show_time=False,
show_path=False,
tracebacks_show_locals=False,
tracebacks_suppress=['easydiffraction'],
tracebacks_max_frames=10,
console=console,
mode=mode,
)
handler.setFormatter(logging.Formatter('%(message)s'))
logger.addHandler(handler)
@staticmethod
def configure(
logger: logging.Logger,
*,
mode: Logger.Mode,
level: Logger.Level,
rich_tracebacks: bool,
) -> None:
"""
Configure the logger with RichHandler and exception hooks.
Parameters
----------
logger : logging.Logger
Logger instance to configure.
mode : Logger.Mode
Output mode (compact or verbose).
level : Logger.Level
Minimum log level to emit.
rich_tracebacks : bool
Whether to enable Rich tracebacks.
"""
LoggerConfig.setup_handlers(
logger,
level=int(level),
rich_tracebacks=rich_tracebacks,
mode=mode.value,
)
if rich_tracebacks and mode == Logger.Mode.VERBOSE:
ExceptionHookManager.install_verbose_hook(logger)
elif mode == Logger.Mode.COMPACT:
ExceptionHookManager.install_compact_hook(logger)
ExceptionHookManager.install_jupyter_traceback_suppressor(logger)
else:
ExceptionHookManager.restore_original_hook()
class ExceptionHookManager:
"""Handles installation and restoration of exception hooks."""
@staticmethod
def install_verbose_hook(logger: logging.Logger) -> None:
"""
Install a verbose exception hook that prints rich tracebacks.
Parameters
----------
logger : logging.Logger
Logger used to emit the exception information.
"""
if not hasattr(Logger, '_orig_excepthook'):
Logger._orig_excepthook = sys.excepthook # type: ignore[attr-defined]
def aligned_excepthook(
exc_type: type[BaseException],
exc: BaseException,
tb: TracebackType | None,
) -> None:
"""Log the exception with full traceback via Rich."""
original_args = getattr(exc, 'args', ())
message = str(exc)
with suppress(Exception):
exc.args = ()
try:
logger.error(message, exc_info=(exc_type, exc, tb))
except Exception:
logger.exception('Unhandled exception (logging failure)')
finally:
with suppress(Exception):
exc.args = original_args
sys.excepthook = aligned_excepthook # type: ignore[assignment]
@staticmethod
def install_compact_hook(logger: logging.Logger) -> None:
"""
Install a compact exception hook that logs message-only.
Parameters
----------
logger : logging.Logger
Logger used to emit the error message.
"""
if not hasattr(Logger, '_orig_excepthook'):
Logger._orig_excepthook = sys.excepthook # type: ignore[attr-defined]
def compact_excepthook(
_exc_type: type[BaseException],
exc: BaseException,
_tb: TracebackType | None,
) -> None:
"""Log the exception message and exit."""
logger.error(str(exc))
raise SystemExit(1)
sys.excepthook = compact_excepthook # type: ignore[assignment]
@staticmethod
def restore_original_hook() -> None:
"""Restore the original sys.excepthook if it was overridden."""
if hasattr(Logger, '_orig_excepthook'):
sys.excepthook = Logger._orig_excepthook # type: ignore[attr-defined]
# Jupyter-specific traceback suppression (inlined here)
@staticmethod
def _suppress_traceback(logger: object) -> object:
"""
Build a Jupyter exception callback that logs the message only.
Parameters
----------
logger : object
Logger used to emit error messages.
Returns
-------
object
A callable suitable for IPython's set_custom_exc that
suppresses full tracebacks and logs only the exception
message.
"""
def suppress_jupyter_traceback(*args: object, **kwargs: object) -> None:
"""Log only the exception message."""
# IPython's custom_exc handler passes
# (shell, etype, evalue, tb, tb_offset)
evalue_arg_index = 2
try:
evalue = (
args[evalue_arg_index]
if len(args) > evalue_arg_index
else kwargs.get('_evalue') or kwargs.get('evalue')
)
logger.error(str(evalue))
except (IndexError, TypeError, AttributeError, ValueError) as err:
logger.debug('Jupyter traceback suppressor failed: %r', err)
return suppress_jupyter_traceback
@staticmethod
def install_jupyter_traceback_suppressor(logger: logging.Logger) -> None:
"""
Install a Jupyter/IPython exception handler for tracebacks.
Parameters
----------
logger : logging.Logger
Logger used to emit error messages.
"""
try:
from IPython import get_ipython # noqa: PLC0415
ip = get_ipython()
if ip is not None:
ip.set_custom_exc(
(BaseException,), ExceptionHookManager._suppress_traceback(logger)
)
except (ImportError, AttributeError, TypeError) as err:
msg = f'Failed to install Jupyter traceback suppressor: {err!r}'
logger.debug(msg)
# ======================================================================
# LOGGER CORE
# ======================================================================
class Logger:
"""
Centralized logging with Rich formatting and two modes.
Environment variables: ED_LOG_MODE: set default mode ('verbose' or
'compact') ED_LOG_LEVEL: set default level ('DEBUG', 'INFO', etc.)
"""
# --- Enums ---
class Mode(Enum):
"""Output modes (see :class:`Logger`)."""
VERBOSE = 'verbose' # rich traceback panel
COMPACT = 'compact' # single line; no traceback
@classmethod
def default(cls) -> Logger.Mode:
"""Return the default output mode (compact)."""
return cls.COMPACT
class Level(IntEnum):
"""Mirror stdlib logging levels."""
DEBUG = logging.DEBUG
INFO = logging.INFO
WARNING = logging.WARNING
ERROR = logging.ERROR
CRITICAL = logging.CRITICAL
@classmethod
def default(cls) -> Logger.Level:
"""Return the default log level (WARNING)."""
return cls.WARNING
class Reaction(Enum):
"""Reaction to errors (see :class:`Logger`)."""
RAISE = auto()
WARN = auto()
@classmethod
def default(cls) -> Logger.Reaction:
"""Return the default error reaction (RAISE)."""
return cls.RAISE
# --- Internal state ---
_logger = logging.getLogger('easydiffraction')
_configured = False
_mode: Mode = Mode.VERBOSE
_reaction: Reaction = Reaction.RAISE # TODO: not default?
_console = ConsoleManager.get()
# ===== CONFIGURATION =====
@classmethod
def configure(
cls,
*,
mode: Mode | None = None,
level: Level | None = None,
reaction: Reaction | None = None,
rich_tracebacks: bool | None = None,
) -> None:
"""
Configure logger.
mode: default COMPACT in Jupyter else VERBOSE level: minimum log
level rich_tracebacks: override automatic choice
Environment variables: ED_LOG_MODE: set default mode ('verbose'
or 'compact') ED_LOG_LEVEL: set default level ('DEBUG', 'INFO',
etc.)
"""
env_mode = os.getenv('ED_LOG_MODE')
env_level = os.getenv('ED_LOG_LEVEL')
env_reaction = os.getenv('ED_LOG_REACTION')
# Read from environment if not provided
if mode is None and env_mode is not None:
with suppress(ValueError):
mode = Logger.Mode(env_mode.lower())
if level is None and env_level is not None:
with suppress(KeyError):
level = Logger.Level[env_level.upper()]
if reaction is None and env_reaction is not None:
with suppress(KeyError):
reaction = Logger.Reaction[env_reaction.upper()]
# Set defaults if still None
if mode is None:
mode = Logger.Mode.default()
if level is None:
level = Logger.Level.default()
if reaction is None:
reaction = Logger.Reaction.default()
cls._mode = mode
cls._reaction = reaction
if rich_tracebacks is None:
rich_tracebacks = mode == Logger.Mode.VERBOSE
LoggerConfig.configure(
logger=cls._logger,
mode=mode,
level=level,
rich_tracebacks=rich_tracebacks,
)
cls._configured = True
@classmethod
def _install_jupyter_traceback_suppressor(cls) -> None:
"""Install the Jupyter traceback suppressor safely."""
ExceptionHookManager.install_jupyter_traceback_suppressor(cls._logger)
# ===== Helpers =====
@classmethod
def set_mode(cls, mode: Mode) -> None:
"""
Set the output mode and reconfigure the logger.
Parameters
----------
mode : Mode
The desired output mode (VERBOSE or COMPACT).
"""
cls.configure(mode=mode, level=cls.Level(cls._logger.level))
@classmethod
def set_level(cls, level: Level) -> None:
"""
Set the minimum log level and reconfigure the logger.
Parameters
----------
level : Level
The desired minimum log level.
"""
cls.configure(mode=cls._mode, level=level)
@classmethod
def mode(cls) -> Mode:
"""
Return the currently active output mode.
Returns
-------
Mode
The current Logger.Mode value.
"""
return cls._mode
@classmethod
def _lazy_config(cls) -> None:
if not cls._configured: # pragma: no cover - trivial
cls.configure()
# ===== Core Routing =====
@classmethod
def handle(
cls,
*messages: str,
level: Level = Level.ERROR,
exc_type: type[BaseException] | None = AttributeError,
) -> None:
"""Route a log message (see class docs for policy)."""
cls._lazy_config()
message = ' '.join(messages)
# Prioritize explicit UserWarning path so pytest captures
# warnings
if exc_type is UserWarning:
if in_pytest():
warnings.warn(message, UserWarning, stacklevel=2)
else:
cls._logger.warning(message)
return
# Special handling for Reaction.WARN (non-warning cases)
if cls._reaction is cls.Reaction.WARN:
# Log as error/critical (keep icon) but continue execution
cls._logger.log(int(level), message)
return
if exc_type is not None:
if cls._mode is cls.Mode.VERBOSE:
raise exc_type(message)
if cls._mode is cls.Mode.COMPACT:
raise exc_type(message) from None
cls._logger.log(int(level), message)
# ==================================================================
# CONVENIENCE API
# ==================================================================
@classmethod
def debug(cls, *messages: str) -> None:
"""
Log one or more messages at DEBUG level.
Parameters
----------
*messages : str
Message parts joined with a space before logging.
"""
cls.handle(*messages, level=cls.Level.DEBUG, exc_type=None)
@classmethod
def info(cls, *messages: str) -> None:
"""
Log one or more messages at INFO level.
Parameters
----------
*messages : str
Message parts joined with a space before logging.
"""
cls.handle(*messages, level=cls.Level.INFO, exc_type=None)
@classmethod
def warning(cls, *messages: str, exc_type: type[BaseException] | None = None) -> None:
"""
Log one or more messages at WARNING level.
Parameters
----------
*messages : str
Message parts joined with a space before logging.
exc_type : type[BaseException] | None, default=None
If provided, raise this exception type instead of logging.
"""
cls.handle(*messages, level=cls.Level.WARNING, exc_type=exc_type)
@classmethod
def error(cls, *messages: str, exc_type: type[BaseException] = AttributeError) -> None:
"""
Log one or more messages at ERROR level.
Parameters
----------
*messages : str
Message parts joined with a space before logging.
exc_type : type[BaseException], default=AttributeError
Exception type to raise in VERBOSE/COMPACT mode.
"""
cls.handle(*messages, level=cls.Level.ERROR, exc_type=exc_type)
@classmethod
def critical(cls, *messages: str, exc_type: type[BaseException] = RuntimeError) -> None:
"""
Log one or more messages at CRITICAL level.
Parameters
----------
*messages : str
Message parts joined with a space before logging.
exc_type : type[BaseException], default=RuntimeError
Exception type to raise in VERBOSE/COMPACT mode.
"""
cls.handle(*messages, level=cls.Level.CRITICAL, exc_type=exc_type)
# ======================================================================
# PRINTER
# ======================================================================
_RICH_RED_MARKUP_PATTERN = re.compile(r'\[red\](.*?)\[/red\]')
_RICH_DIM_MARKUP_PATTERN = re.compile(r'\[/?dim\]')
def _rich_markup_to_inline_html(markup: str) -> str:
"""
Translate a narrow subset of Rich markup to inline HTML.
Handles ``[red]β¦[/red]`` (β a red ``<span>``) and silently strips
``[dim]`` / ``[/dim]`` tags (the surrounding container already
conveys dimness via CSS opacity). Other Rich markup passes through
unescaped β callers must only emit markup from this allow-list when
calling ``ConsolePrinter.small`` in Jupyter.
Parameters
----------
markup : str
Rich markup string.
Returns
-------
str
HTML-escaped string with the allow-listed Rich tags translated
to inline ``<span>`` styles.
"""
escaped = html.escape(markup)
without_dim = _RICH_DIM_MARKUP_PATTERN.sub('', escaped)
return _RICH_RED_MARKUP_PATTERN.sub(r'<span style="color:#dc3545">\1</span>', without_dim)
class ConsolePrinter:
"""Printer utility for the shared console with left padding."""
_console = ConsoleManager.get()
@classmethod
def print(cls, *objects: object, **kwargs: object) -> None:
"""
Print objects to the console with left padding.
- Renderables (Rich types like Text, Table, Panel, etc.) are
kept as-is. - Non-renderables (ints, floats, Path, etc.) are
converted to str().
"""
safe_objects = []
for obj in objects:
if isinstance(obj, RenderableType):
safe_objects.append(obj)
elif isinstance(obj, Path):
safe_objects.append(str(obj))
else:
safe_objects.append(str(obj))
# If multiple objects, join with spaces
renderable = (
' '.join(str(o) for o in safe_objects)
if all(isinstance(o, str) for o in safe_objects)
else Group(*safe_objects)
)
cls._console.print(renderable, **kwargs)
@classmethod
def paragraph(cls, title: str) -> None:
"""
Print a bold blue paragraph heading.
Parameters
----------
title : str
Heading text; substrings enclosed in single quotes are
rendered without the bold deep_sky_blue3 style.
"""
parts = re.split(r"('.*?')", title)
text = Text()
for part in parts:
if part.startswith("'") and part.endswith("'"):
text.append(part)
else:
text.append(part, style=CONSOLE_PARAGRAPH_STYLE)
formatted = f'{text.markup}'
if not in_jupyter():
formatted = f'\n{formatted}'
cls._console.print(formatted)
@classmethod
def section(cls, title: str) -> None:
"""Format a section header with bold green text."""
full_title = f'{title.upper()}'
line = 'β' * len(full_title)
formatted = f'[bold green]{line}\n{full_title}\n{line}[/bold green]'
if not in_jupyter():
formatted = f'\n{formatted}'
cls._console.print(formatted)
@classmethod
def small(cls, *lines: str) -> None:
"""
Print one or more lines as dim, smaller supplementary text.
Intended for table footnote glossaries and inline warning notes
that should read as subordinate to the table or block they sit
beneath. In Jupyter the lines render inside a single
``<small>``-style HTML element so the font size matches
Jupyter's ``.dataframe`` table-cell text. In a terminal the
lines render with Rich's ``dim`` style. Rich ``[red]β¦[/red]``
markup inside ``lines`` is preserved in both renderers.
Parameters
----------
*lines : str
Pre-formatted display lines. Each may contain Rich
``[red]β¦[/red]`` markup; other Rich markup is rendered in
the terminal and stripped in the HTML output.
"""
if not lines:
return
if in_jupyter():
try:
from IPython.display import HTML # noqa: PLC0415
from IPython.display import display # noqa: PLC0415
body = '<br>'.join(_rich_markup_to_inline_html(line) for line in lines)
except ImportError: # pragma: no cover
pass
else:
display(
HTML(
'<div style="font-size:smaller;opacity:0.7;'
'font-family:monospace;white-space:pre">'
f'{body}</div>'
)
)
return
for line in lines:
cls._console.print(f'[dim]{line}[/dim]')
@classmethod
def chapter(cls, title: str) -> None:
"""Format a chapter header in bold magenta, uppercase."""
width = ConsoleManager._detect_width()
symbol = 'β'
full_title = f' {title.upper()} '
pad_len = (width - len(full_title)) // 2
padding = symbol * pad_len
line = f'[bold magenta]{padding}{full_title}{padding}[/bold magenta]'
if len(line) < width:
line += symbol
formatted = f'{line}'
if not in_jupyter():
formatted = f'\n{formatted}'
cls._console.print(formatted)
# Configure logging at import time
Logger.configure()
# Convenient alias for logger
log = Logger
# Convenient alias for console printer
console = ConsolePrinter