Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 5 additions & 1 deletion software/main_hcs.py
Original file line number Diff line number Diff line change
Expand Up @@ -41,7 +41,11 @@ def show_acq_config(cfm):

if args.verbose:
log.info("Turning on debug logging.")
squid.logging.set_log_level(logging.DEBUG)
squid.logging.set_stdout_log_level(logging.DEBUG)

if not squid.logging.add_file_logging(f"{squid.logging.get_default_log_directory()}/main_hcs.log"):
log.error("Couldn't setup logging to file!")
sys.exit(1)

legacy_config = False
cf_editor_parser = ConfigParser()
Expand Down
2 changes: 1 addition & 1 deletion software/main_malaria.py
Original file line number Diff line number Diff line change
Expand Up @@ -37,7 +37,7 @@ def show_acq_config(cfm):
args = parser.parse_args()

if args.verbose:
squid.logging.set_log_level(logging.DEBUG)
squid.logging.set_stdout_log_level(logging.DEBUG)

legacy_config = False
cf_editor_parser = ConfigParser()
Expand Down
78 changes: 59 additions & 19 deletions software/squid/logging.py
Original file line number Diff line number Diff line change
@@ -1,10 +1,15 @@
import logging as py_logging
import logging.handlers
import os.path
import threading
from typing import Optional, Type
from types import TracebackType
import sys
import platformdirs

_squid_root_logger_name= "squid"
_squid_root_logger_name = "squid"
_baseline_log_format = "%(asctime)s.%(msecs)03d - %(name)s - %(levelname)s - %(message)s (%(filename)s:%(lineno)d)"
_baseline_log_dateformat = "%Y-%m-%d %H:%M:%S"


# The idea for this CustomFormatter is cribbed from https://stackoverflow.com/a/56944256
Expand All @@ -14,7 +19,7 @@ class _CustomFormatter(py_logging.Formatter):
RED = "\x1b[31;20m"
BOLD_RED = "\x1b[31;1m"
RESET = "\x1b[0m"
FORMAT = "%(asctime)s.%(msecs)03d - %(name)s - %(levelname)s - %(message)s (%(filename)s:%(lineno)d)"
FORMAT = _baseline_log_format

FORMATS = {
py_logging.DEBUG: GRAY + FORMAT + RESET,
Expand All @@ -26,7 +31,7 @@ class _CustomFormatter(py_logging.Formatter):

# NOTE(imo): The datetime hackery is so that we can have millisecond timestamps using a period instead
# of comma. The default asctime + datefmt uses a comma.
FORMATTERS = {level: py_logging.Formatter(fmt, datefmt="%Y-%m-%d %H:%M:%S") for (level, fmt) in FORMATS.items()}
FORMATTERS = {level: py_logging.Formatter(fmt, datefmt=_baseline_log_dateformat) for (level, fmt) in FORMATS.items()}

def format(self, record):
return self.FORMATTERS[record.levelno].format(record)
Expand All @@ -36,9 +41,11 @@ def format(self, record):

# Make sure the squid root logger has all the handlers we want setup. We could move this into a helper so it
# isn't done at the module level, but not needing to remember to call some helper to setup formatting is nice.
# Also set the default logging level to INFO
# Also set the default logging level to INFO on the stream handler, but DEBUG on the root logger so we can have
# other loggers at different levels.
_COLOR_STREAM_HANDLER.setLevel(py_logging.INFO)
py_logging.getLogger(_squid_root_logger_name).addHandler(_COLOR_STREAM_HANDLER)
py_logging.getLogger(_squid_root_logger_name).setLevel(py_logging.INFO)
py_logging.getLogger(_squid_root_logger_name).setLevel(py_logging.DEBUG)


def get_logger(name: Optional[str] = None) -> py_logging.Logger:
Expand All @@ -53,28 +60,24 @@ def get_logger(name: Optional[str] = None) -> py_logging.Logger:

return logger

log = get_logger(__name__)

def set_log_level(level):
def set_stdout_log_level(level):
"""
All squid code should use this set_log_level method, and the corresponding squid.logging.get_logger,
All squid code should use this set_stdout_log_level method, and the corresponding squid.logging.get_logger,
to control squid-only logging.

This does not modify the log level of loggers outside the squid logger hierarchy! If global logging control
is needed the normal logging package tools can be used instead.
is needed the normal logging package tools can be used instead. It also leaves FileHandler log levels such that
they can always be outputting everything (regardless of what we set the stdout log level to)
"""
squid_root_logger = get_logger()
squid_root_logger.setLevel(level)

# There's no `getAllChildren` method on the logger or its manager, so we just grab the manager
# for our root logger and then check all other loggers to see if they start with our root logger prefix
# to find all the squid specific logger.
for (name, logger) in squid_root_logger.manager.loggerDict.items():
# The logging module uses the PlaceHolder object for nodes in the hierarchy that
# have children, but no associated loggers. EG if we create a logger at
# squid.control.gui_hcs but not at squid.control, then the logger for squid.control
# exists but is a PlaceHolder (until someone explicitly requests it).
if name.startswith(_squid_root_logger_name) and isinstance(logger, py_logging.Logger):
logger.setLevel(level)
for handler in squid_root_logger.handlers:
# We always want the file handlers to capture everything, so don't touch them.
if isinstance(handler, logging.FileHandler):
continue
handler.setLevel(level)


def register_crash_handler(handler, call_existing_too=True):
Expand Down Expand Up @@ -141,3 +144,40 @@ def uncaught_exception_logger(exception_type: Type[BaseException], value: BaseEx
logger.exception("Uncaught Exception!", exc_info=value)

register_crash_handler(uncaught_exception_logger, call_existing_too=False)

def get_default_log_directory():
return platformdirs.user_log_path(_squid_root_logger_name, "cephla")

def add_file_logging(log_filename, replace_existing=False):
root_logger = get_logger()
abs_path = os.path.abspath(log_filename)
for handler in root_logger.handlers:
if isinstance(handler, logging.handlers.BaseRotatingHandler):
if handler.baseFilename == abs_path:
if replace_existing:
root_logger.removeHandler(handler)
else:
log.error(f"RotatingFileHandler already exists for {abs_path}, and replace_existing==False!")
return False

log_file_existed = False
if os.path.isfile(abs_path):
log_file_existed = True

os.makedirs(os.path.dirname(abs_path), exist_ok=True)

# For now, don't worry about rollover after a certain size or time. Just get a new file per call.
new_handler = logging.handlers.RotatingFileHandler(abs_path, maxBytes=0, backupCount=25)
new_handler.setLevel(py_logging.DEBUG)

formatter = py_logging.Formatter(fmt=_baseline_log_format, datefmt=_baseline_log_dateformat)
new_handler.setFormatter(formatter)

log.info(f"Adding new file logger writing to file '{new_handler.baseFilename}")
root_logger.addHandler(new_handler)

# We want a new log file every time we start, so force one at startup if the log file already existed.
if log_file_existed:
new_handler.doRollover()

return True
37 changes: 36 additions & 1 deletion software/tests/squid/test_logging.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,6 @@
import logging
import tempfile

import squid.logging

def test_root_logger():
Expand All @@ -12,4 +15,36 @@ def test_children_loggers():
child_b_logger = child_a_logger.getChild(child_b)

assert child_a_logger.name == f"{squid.logging._squid_root_logger_name}.{child_a}"
assert child_b_logger.name == f"{squid.logging._squid_root_logger_name}.{child_a}.{child_b}"
assert child_b_logger.name == f"{squid.logging._squid_root_logger_name}.{child_a}.{child_b}"

def test_file_loggers():
log_file_name = tempfile.mktemp()

def line_count():
with open(log_file_name, "r") as fh:
return len(list(fh))

def contains(string):
with open(log_file_name, "r") as fh:
for l in fh:
if string in l:
return True
return False

assert squid.logging.add_file_logging(log_file_name)
assert not squid.logging.add_file_logging(log_file_name)

initial_line_count = line_count()
log = squid.logging.get_logger("log test")
squid.logging.set_stdout_log_level(logging.DEBUG)

log.debug("debug msg")
debug_ling_count = line_count()
assert debug_ling_count > initial_line_count

squid.logging.set_stdout_log_level(logging.INFO)

a_debug_message = "another message but when stdout is at INFO"
log.debug(a_debug_message)
assert line_count() > debug_ling_count
assert contains(a_debug_message)