Spaces:
Sleeping
Sleeping
Download utils/logging_setup.py from Krixter/application: direct link, hf CLI and curl.
- Browser
- Download file 2.08 kB
-
https://huggingface.co/spaces/Krixter/application/resolve/main/utils/logging_setup.py
- Command line
-
hf download hf://spaces/Krixter/application/utils/logging_setup.py
-
curl -L -o logging_setup.py https://huggingface.co/spaces/Krixter/application/resolve/main/utils/logging_setup.py
2.08 kB
| """ | |
| Logging system: in-memory ring buffer + file output. | |
| """ | |
| from __future__ import annotations | |
| import logging | |
| import time | |
| from collections import deque | |
| from pathlib import Path | |
| from typing import Optional | |
| class RingBufferHandler(logging.Handler): | |
| """Keeps last N log lines in memory for UI display.""" | |
| def __init__(self, maxlen: int = 500): | |
| super().__init__() | |
| self._buffer: deque[str] = deque(maxlen=maxlen) | |
| def emit(self, record: logging.LogRecord) -> None: | |
| msg = self.format(record) | |
| self._buffer.append(msg) | |
| def get_lines(self, last: int = 100) -> list[str]: | |
| lines = list(self._buffer) | |
| return lines[-last:] if last else lines | |
| def get_text(self, last: int = 100) -> str: | |
| return "\n".join(self.get_lines(last)) | |
| def clear(self) -> None: | |
| self._buffer.clear() | |
| class AppLogger: | |
| _instance: Optional["AppLogger"] = None | |
| def __init__(self, log_dir: Path = Path("output")): | |
| self.log_dir = log_dir | |
| self.log_dir.mkdir(parents=True, exist_ok=True) | |
| self.log_file = log_dir / "logs.txt" | |
| self.ring = RingBufferHandler(maxlen=1000) | |
| self.ring.setFormatter(logging.Formatter("%(asctime)s %(levelname)s %(message)s", "%H:%M:%S")) | |
| file_handler = logging.FileHandler(self.log_file, encoding="utf-8") | |
| file_handler.setFormatter(logging.Formatter("%(asctime)s %(levelname)s %(name)s %(message)s")) | |
| root = logging.getLogger() | |
| root.setLevel(logging.INFO) | |
| root.addHandler(self.ring) | |
| root.addHandler(file_handler) | |
| def get(cls, log_dir: Path = Path("output")) -> "AppLogger": | |
| if cls._instance is None: | |
| cls._instance = cls(log_dir) | |
| return cls._instance | |
| def ui_log(self, msg: str) -> None: | |
| """Add a message directly to the ring buffer (bypassing Python log level).""" | |
| ts = time.strftime("%H:%M:%S") | |
| self.ring._buffer.append(f"{ts} {msg}") | |
| def get_ui_text(self, last: int = 200) -> str: | |
| return self.ring.get_text(last) | |