Spaces:
Sleeping
Sleeping
File size: 2,372 Bytes
f957d0a |
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 |
import logging
import logging.handlers
import multiprocessing
import datetime
from pathlib import Path
import atexit
from typing import Tuple, Optional
DEFAULT_LOGS_DIR = "logs"
def setup_logging(project_name: str, logs_dir: str = DEFAULT_LOGS_DIR, level: int = logging.INFO,
queue_logging: bool = True) -> Tuple[logging.Logger, Optional[logging.handlers.QueueListener]]:
"""
Configure parallel-safe logging.
Creates a log file named ``{project_name}-{timestamp}.log`` inside ``logs_dir``.
If ``queue_logging`` is True, uses ``multiprocessing.Queue`` with ``QueueHandler``/``QueueListener``
for process-safe logging. Returns the application logger and the listener (if any).
Caller does not need to manage handlers individually; the listener is auto-stopped at exit.
"""
timestamp = datetime.datetime.now().strftime("%Y%m%d-%H%M%S")
logs_path = Path(logs_dir)
logs_path.mkdir(parents=True, exist_ok=True)
logfile = logs_path / f"{project_name}-{timestamp}.log"
formatter = logging.Formatter(
fmt="%(asctime)s | %(levelname)s | %(processName)s | %(name)s | %(message)s",
datefmt="%Y-%m-%d %H:%M:%S"
)
logger = logging.getLogger("bidsifier")
logger.setLevel(level)
listener: Optional[logging.handlers.QueueListener] = None
if queue_logging:
log_queue: multiprocessing.Queue = multiprocessing.Queue(-1)
queue_handler = logging.handlers.QueueHandler(log_queue)
file_handler = logging.FileHandler(str(logfile), encoding="utf-8")
file_handler.setFormatter(formatter)
listener = logging.handlers.QueueListener(log_queue, file_handler)
listener.start()
logger.addHandler(queue_handler)
def _stop_listener() -> None:
try:
listener.stop()
except Exception:
pass
atexit.register(_stop_listener)
else:
file_handler = logging.FileHandler(str(logfile), encoding="utf-8")
file_handler.setFormatter(formatter)
logger.addHandler(file_handler)
# Also add a simple stderr stream handler for immediate feedback.
stream_handler = logging.StreamHandler()
stream_handler.setFormatter(formatter)
logger.addHandler(stream_handler)
logger.debug("Logging initialized: %s", logfile)
return logger, listener
|