104 lines
3.5 KiB
Python
104 lines
3.5 KiB
Python
import logging
|
|
import logging.config
|
|
import yaml
|
|
import os
|
|
|
|
from PySide6.QtCore import QObject, Signal
|
|
|
|
class QtLogEmitter(QObject):
|
|
message = Signal(str)
|
|
|
|
class QtLogHandler(logging.Handler):
|
|
def __init__(self, emitter: QtLogEmitter):
|
|
super().__init__()
|
|
self.emitter = emitter
|
|
|
|
def emit(self, record: logging.LogRecord):
|
|
try:
|
|
msg = self.format(record)
|
|
self.emitter.message.emit(msg)
|
|
except Exception:
|
|
self.handleError(record)
|
|
|
|
def setup_logger(name="aareDAQ", base_dir: str | None = '/tmp/mxlogs'):
|
|
# switch to production mode using: $ APP_ENV=prod python main.py
|
|
env = os.getenv("APP_ENV", "dev") # default: dev
|
|
config_file = (
|
|
"/sls/MX/applications/aaredaq/common/src/aaredaqlib/logging_prod.yaml" if env == "prod" else "/sls/MX/applications/aaredaq/common/src/aaredaqlib/logging_dev.yaml"
|
|
)
|
|
with open(config_file, "r") as f:
|
|
config = yaml.safe_load(f.read())
|
|
|
|
env_log_dir = os.getenv("LOG_DIR")
|
|
base_dir = base_dir or env_log_dir
|
|
|
|
# Ensure directories for file handlers exist
|
|
handlers = config.get("handlers", {})
|
|
|
|
for h in handlers.values():
|
|
# RotatingFileHandler and FileHandler use "filename"
|
|
filename = h.get("filename")
|
|
if not filename:
|
|
continue
|
|
if base_dir:
|
|
abs_filename = os.path.join(base_dir, os.path.basename(filename))
|
|
else:
|
|
abs_filename = (
|
|
filename
|
|
if os.path.isabs(filename)
|
|
else os.path.join(os.path.dirname(config_file), filename)
|
|
)
|
|
|
|
h["filename"] = abs_filename
|
|
log_dir = os.path.dirname(abs_filename)
|
|
if log_dir and not os.path.exists(log_dir):
|
|
os.makedirs(log_dir, exist_ok=True)
|
|
|
|
logging.config.dictConfig(config)
|
|
logging.getLogger("redis_lock").setLevel(logging.WARNING)
|
|
logging.getLogger("urllib3").setLevel(logging.WARNING)
|
|
logging.getLogger("matplotlib.font_manager").setLevel(logging.WARNING)
|
|
logging.getLogger("aaredaq").setLevel(logging.DEBUG) # DAQ
|
|
logging.getLogger("aaregui").setLevel(logging.INFO)
|
|
return logging.getLogger(name)
|
|
|
|
def find_existing_formatter(default_fmt="%(asctime)s [%(levelname)s] %(name)s: %(message)s", default_datefmt=None):
|
|
root = logging.getLogger()
|
|
for h in root.handlers:
|
|
if h.formatter:
|
|
return h.formatter
|
|
|
|
for logger_name, logger in logging.Logger.manager.loggerDict.items():
|
|
if isinstance(logger, logging.Logger):
|
|
for h in logger.handlers:
|
|
if h.formatter:
|
|
return h.formatter
|
|
|
|
return logging.Formatter(default_fmt, datefmt=default_datefmt)
|
|
|
|
def attach_to_logger(logger_name: str = "", handler: logging.Handler = None):
|
|
logging.getLogger(logger_name).addHandler(handler)
|
|
|
|
if __name__ == "__main__":
|
|
|
|
os.environ["APP_ENV"] = "dev" # or "prod"
|
|
#os.environ["LOG_DIR"] = "/tmp/mxlogs" # optional; matches your setup code
|
|
|
|
log = setup_logger("aareDAQ", base_dir="/tmp/test_logs") # use the configured logger name
|
|
|
|
log.debug("debug smoke test")
|
|
log.error("error smoke test")
|
|
log.info("info smoke test")
|
|
log.warning("warning smoke test")
|
|
log.critical("critical smoke test")
|
|
|
|
for h in log.handlers:
|
|
if hasattr(h, "flush"):
|
|
h.flush()
|
|
|
|
for h in log.handlers:
|
|
log.debug("debug test")
|
|
log.error("error test")
|
|
print(type(h), getattr(h, "baseFilename", None), h.level)
|
|
|