Skip to content
Snippets Groups Projects
Commit c7661f2a authored by vertighel's avatar vertighel
Browse files

loguru again

parent 781420f4
No related branches found
No related tags found
No related merge requests found
Pipeline #29056 passed
...@@ -113,13 +113,10 @@ image_number = {v: k for k, v in image_state.items()} ...@@ -113,13 +113,10 @@ image_number = {v: k for k, v in image_state.items()}
# Directories # Directories
############ ############
DATA_FOLDER_NAME = "data" DATA_FOLDER = "data"
LOG_DIRECTORY_NAME = "log" LOG_FOLDER = "log"
fits_folder = "fits" FITS_FOLDER = "fits"
focus_folder = "focus" FOCUS_FOLDER = "focus"
# data_folder = "data"
# log_folder = "log"
dir_type = { dir_type = {
0: "dark", 0: "dark",
...@@ -131,21 +128,6 @@ dir_type = { ...@@ -131,21 +128,6 @@ dir_type = {
# Reversing dir_type # Reversing dir_type
dir_number = {v: k for k, v in dir_type.items()} dir_number = {v: k for k, v in dir_type.items()}
############
# Logging
############
# Log file naming
LOG_FILE_EXTENSION = ".log"
LATEST_LOG_FILENAME = "latest"
# Rotation settings (used by logger.py)
LOG_ROTATION_TIME_H = 12 # Midday (hour in UTC)
LOG_ROTATION_TIME_M = 0 # Midday (minute in UTC)
LOG_ROTATION_TIME_S = 0 # Midday (second in UTC)
LOG_ROTATION_INTERVAL_DAYS = 1 # Rotate daily
LOG_BACKUP_COUNT = 0 # 0 means keep all backup files (infinite)
############ ############
# FITS # FITS
############ ############
...@@ -154,6 +136,11 @@ LOG_BACKUP_COUNT = 0 # 0 means keep all backup files (infinite) ...@@ -154,6 +136,11 @@ LOG_BACKUP_COUNT = 0 # 0 means keep all backup files (infinite)
imagetyp = "IMAGETYP" # 'Light' imagetyp = "IMAGETYP" # 'Light'
dateobs = "DATE-OBS" # '2021-12-18T05:09:56.163' dateobs = "DATE-OBS" # '2021-12-18T05:09:56.163'
FILE_PREFIX = "OARPAF." ############
ext = ".fits" # File extension # EXTENSIONS
focus_ext = ".foc" ############
FILE_PREFIX = "OARPAF"
FITS_EXT = "fits" # File extension
FOCUS_EXT = "foc"
LOG_EXT = "log"
...@@ -143,9 +143,9 @@ class Sequencer(): ...@@ -143,9 +143,9 @@ class Sequencer():
full_module_name = f"noctua.templates.{template_name}" full_module_name = f"noctua.templates.{template_name}"
try: try:
tplmodule = importlib.import_module(full_module_name) tplmodule = importlib.import_module(full_module_name)
except ModuleNotFoundError: except ModuleNotFoundError as e:
log.error( log.error(
f"SEQUENCER: Template module '{full_module_name}' not found.") f"SEQUENCER: Template module '{full_module_name}' not found: {e}")
self.error.append( self.error.append(
f"Template module '{full_module_name}' not found.") f"Template module '{full_module_name}' not found.")
continue continue
......
...@@ -5,7 +5,7 @@ ...@@ -5,7 +5,7 @@
from time import sleep from time import sleep
# Third-party modules # Third-party modules
from templates.basetemplate import BaseTemplate from .basetemplate import BaseTemplate
# Other templates # Other templates
from ..config.constants import on_off from ..config.constants import on_off
......
...@@ -7,7 +7,7 @@ from astropy.time import Time ...@@ -7,7 +7,7 @@ from astropy.time import Time
# Other templates # Other templates
from ..config.constants import on_off from ..config.constants import on_off
from ..directory.structure import foc_path from ..utils.structure import foc_path
# from devices import lamp, light # from devices import lamp, light
from ..utils.logger import log from ..utils.logger import log
from .basetemplate import BaseTemplate from .basetemplate import BaseTemplate
......
#!/usr/bin/env python3 #!/usr/bin/env python
# -*- coding: utf-8 -*-
"""Custom format log""" """Custom format log"""
# System modules # System modules
import logging
import logging.handlers
import sys import sys
from pathlib import Path import os # Will be used if log_path only returns a directory
from datetime import datetime, time
# Third-party modules # Third-party modules
from loguru import logger
# Custom modules import datetime
from ..config.constants import ( LOG_FILE_EXTENSION, LATEST_LOG_FILENAME, LOG_ROTATION_TIME_H, from .structure import log_path
LOG_ROTATION_TIME_M, LOG_ROTATION_TIME_S,
LOG_ROTATION_INTERVAL_DAYS, LOG_BACKUP_COUNT ) def mylog():
from .structure import get_log_dir_path, get_dated_log_filepath, get_latest_log_filepath "logger function"
LOG_DIR = get_log_dir_path() # Creates directory if it doesn't exist logger.remove() # Remove default handler to prevent duplicate console logs
LATEST_LOG_FILEPATH = get_latest_log_filepath()
time_fmt = "{time:YYYY-MM-DD HH:mm:ss.SSSSSS!UTC} "
log = None # Global logger instance level_fmt = "<level>{level: <8}</level> "
message_fmt = "| {message} "
class ColorizingFormatter(logging.Formatter): stack_fmt = "<bold>({module}.{function}:{line})</bold>"
""" fmt = time_fmt + level_fmt + message_fmt + stack_fmt
A custom logging formatter that adds ANSI color codes to log level names
for console output. # On standard output
""" logger.add(
sys.stderr,
# Color codes format=fmt,
BLACK, RED, GREEN, YELLOW, BLUE, MAGENTA, CYAN, WHITE = range(8) level="DEBUG"
# Log level to color mapping Using bold for emphasis
LEVEL_COLORS = {
logging.DEBUG: f"\x1b[3{MAGENTA};1m", # Bold Magenta
logging.INFO: f"\x1b[3{GREEN};1m", # Bold Green
logging.WARNING: f"\x1b[3{YELLOW};1m", # Bold Yellow
logging.ERROR: f"\x1b[3{RED};1m", # Bold Red
logging.CRITICAL: f"\x1b[4{RED};1m", # Bold Red with Red Background (or just very bold red)
}
RESET_SEQ = "\x1b[0m" # Reset all attributes
def __init__(self, fmt=None, datefmt=None, style='%', validate=True, *, defaults=None):
super().__init__(fmt, datefmt, style, validate, defaults=defaults)
# For UTC timestamps if needed by the base Formatter
self.converter = lambda *args: datetime.utcnow().timetuple()
def format(self, record):
s = super().format(record)
levelname_str = record.levelname
color = self.LEVEL_COLORS.get(record.levelno)
if color:
padded_levelname = f"{levelname_str:<8}"
if padded_levelname in s:
colored_levelname = f"{color}{padded_levelname}{self.RESET_SEQ}"
s = s.replace(padded_levelname, colored_levelname, 1) # Replace only the first occurrence
else: # Fallback if padded version not found, try raw (less likely to match in typical formats)
if levelname_str in s:
colored_levelname = f"{color}{levelname_str}{self.RESET_SEQ}"
s = s.replace(levelname_str, colored_levelname, 1)
return s
def setup_logger(name="noctua_logger", log_level=logging.DEBUG):
global log
if log:
return log
logger_instance = logging.getLogger(name)
logger_instance.setLevel(log_level)
if logger_instance.hasHandlers():
logger_instance.handlers.clear()
log_format_str = "%(asctime)s.%(msecs)03d | %(levelname)-8s | %(message)s (%(module)s.%(funcName)s:%(lineno)d)"
formatter = logging.Formatter(log_format_str, datefmt="%Y-%m-%d %H:%M:%S")
# Ensure timestamps are in UTC
formatter.converter = lambda *args: datetime.utcnow().timetuple()
console_handler = logging.StreamHandler(sys.stderr)
formatter = ColorizingFormatter(log_format_str, datefmt="%Y-%m-%d %H:%M:%S")
console_handler.setFormatter(formatter)
logger_instance.addHandler(console_handler)
# --- TimedRotatingFileHandler for LATEST log ---
# Rotate at midday UTC
rotation_time_utc = time(
hour=LOG_ROTATION_TIME_H,
minute=LOG_ROTATION_TIME_M,
second=LOG_ROTATION_TIME_S,
tzinfo=None
) )
rotating_file_handler = logging.handlers.TimedRotatingFileHandler( full_log_file_path = log_path()
filename=LATEST_LOG_FILEPATH,
when="midnight", # Base rotation type (daily) logger.add(
atTime=rotation_time_utc, # Specific time for rotation full_log_file_path,
interval=LOG_ROTATION_INTERVAL_DAYS, # Interval (1 for daily) format=fmt,
backupCount=LOG_BACKUP_COUNT, # 0 means keep all backups colorize=True, # For file logs
encoding='utf-8', rotation="16:19", # Local time
utc=True # retention="7 days", # Old logs
level="DEBUG"
) )
def custom_namer_astronomical(default_name): # Custom levels
try: logger.level("DEBUG", color="<magenta><bold>")
# Example: '/.../OARPAF.latest.log.2023-04-16_12-00-00' (if rotated with atTime) logger.level("INFO", color="<green><bold>")
base_filepath = Path(default_name)
suffix = base_filepath.suffix # .log or or .2023-04-16_12-00-00
# Remove base filename and first dot
date_str_part = base_filepath.name.replace(LATEST_LOG_FILEPATH.name + '.', '')
date_str_part = date_str_part.split('_')[0] # Takes YYYY-MM-DD part if HH-MM-SS is appended
rotation_calendar_date = datetime.strptime(date_str_part, "%Y-%m-%d").date()
astronomical_date_of_log = rotation_calendar_date - timedelta(days=1)
astro_date_str = astronomical_date_of_log.strftime("%Y-%m-%d")
return str(get_dated_log_filepath(astro_date_str))
except ValueError:
# Fallback if date parsing fails
log.warning(f"Could not parse date from default rotated log name: {default_name}. Using default.")
return default_name
def simpler_custom_namer(default_name):
base_filepath = Path(default_name)
# Extract the date part from the suffix
# Suffix is ".YYYY-MM-DD_HH-MM-SS"
name_parts = base_filepath.name.split(LATEST_LOG_FILEPATH.name + '.')
if len(name_parts) > 1:
timestamp_suffix = name_parts[1]
date_str = timestamp_suffix.split('_')[0] # Get the YYYY-MM-DD part
try:
rotation_event_date = datetime.strptime(date_str, "%Y-%m-%d").date()
# The log contains data for the astronomical night *before* this rotation event date.
log_file_astro_date = rotation_event_date - timedelta(days=1)
final_date_str = log_file_astro_date.strftime("%Y-%m-%d")
# Construct the desired filename using structure.py for consistency
return str(get_dated_log_filepath(final_date_str))
except ValueError:
pass # Fall through to default if parsing fails
# Fallback to a cleaned-up version of the default name if parsing is tricky
dir_name = base_filepath.parent
original_plus_date = base_filepath.name.replace(LOG_FILE_EXTENSION, '') # Remove .log
# original_plus_date is like OARPAF.latest.YYYY-MM-DD_HH-MM-SS
cleaned_name = original_plus_date.replace(LATEST_LOG_FILENAME + '.', '') # Becomes OARPAF.YYYY-MM-DD...
return str(dir_name / (cleaned_name + LOG_FILE_EXTENSION))
rotating_file_handler.namer = simpler_custom_namer # Use the simpler namer
rotating_file_handler.setFormatter(formatter)
logger_instance.addHandler(rotating_file_handler)
log = logger_instance
log.info(f"Logger '{name}' configured. Current logs to: {LATEST_LOG_FILEPATH}")
log.info(f"Rotated daily at {rotation_time_utc} UTC. Backup count: {'Infinite' if LOG_BACKUP_COUNT == 0 else LOG_BACKUP_COUNT}.")
return log
# Initialize logger on module import
from datetime import timedelta # Ensure timedelta is imported for namer
log = setup_logger()
if __name__ == "__main__": return logger
# This ensures mylog() is called only once.
log = mylog()
def main():
"""Main function"""
# log is already initialized globally
log.debug("This is a debug message.") log.debug("This is a debug message.")
log.info("This is an info message.") log.info("This is an info message.")
log.warning("This is a warning message.") log.warning("This is a warning message.")
log.error("This is an error message.") log.error("This is an error message.")
log.critical("This is a critical message.") log.critical("This is a critical message.")
if __name__ == "__main__":
main()
...@@ -14,16 +14,13 @@ from astropy.io import fits ...@@ -14,16 +14,13 @@ from astropy.io import fits
from astropy.time import Time from astropy.time import Time
# Custom modules # Custom modules
from ..config.constants import ( from ..config.constants import ( DATA_FOLDER, LOG_FOLDER, FITS_FOLDER,
DATA_FOLDER_NAME, LOG_DIRECTORY_NAME, FOCUS_FOLDER, FILE_PREFIX, dateobs,
FILE_PREFIX, LOG_FILE_EXTENSION, LATEST_LOG_FILENAME, dir_type, frame_number, imagetyp,
dateobs, dir_type, ext, fits_folder, FITS_EXT, FOCUS_EXT, LOG_EXT )
frame_number, imagetyp, focus_folder, focus_ext
)
PROJECT_ROOT = Path(__file__).parent.parent.parent PROJECT_ROOT = Path(__file__).parent.parent.parent
def date_folder(): def date_folder():
"""Create a date folder string based on astronomical convention """Create a date folder string based on astronomical convention
(changes at midday UTC). (changes at midday UTC).
...@@ -59,7 +56,7 @@ def fits_path(header, dry=False): ...@@ -59,7 +56,7 @@ def fits_path(header, dry=False):
Create a fits file path where the file will be stored Create a fits file path where the file will be stored
""" """
root = PROJECT_ROOT / DATA_FOLDER_NAME / fits_folder root = PROJECT_ROOT / DATA_FOLDER / FITS_FOLDER
date_str = date_folder() date_str = date_folder()
date_path_part = Path(date_str) date_path_part = Path(date_str)
...@@ -72,34 +69,23 @@ def fits_path(header, dry=False): ...@@ -72,34 +69,23 @@ def fits_path(header, dry=False):
return path return path
def get_log_dir_path(dry=False): def log_path(dry=False):
""" """
Returns the Path object for the log directory. Returns the Path object for the log directory.
Creates it if it doesn't exist (unless dry=True). Creates it if it doesn't exist (unless dry=True).
""" """
log_dir = PROJECT_ROOT / DATA_FOLDER_NAME / LOG_DIRECTORY_NAME
if not dry:
log_dir.mkdir(parents=True, exist_ok=True)
return log_dir
def get_dated_log_filepath(date_str, dry=False): path = PROJECT_ROOT / DATA_FOLDER / LOG_FOLDER
"""
Constructs a path for an expected date-stamped (rotated) log
file. date_str: A string like "YYYY-MM-DD"
"""
log_dir = get_log_dir_path(dry=dry) if not dry:
# Filename: OARPAF.YYYY-MM-DD.log path.mkdir(parents=True, exist_ok=True)
log_filename = f"{FILE_PREFIX}{date_str}{LOG_FILE_EXTENSION}"
return log_dir / log_filename
def get_latest_log_filepath(dry=False): # OARPAF.YYYY-MM-DD.foc
""" outfile = f"{FILE_PREFIX}.{LOG_EXT}"
Constructs the path for the "latest" log file. outpath = path / outfile
""" return outpath
log_dir = get_log_dir_path(dry=dry)
latest_filename = f"{FILE_PREFIX}{LATEST_LOG_FILENAME}{LOG_FILE_EXTENSION}" return outpath
return log_dir / latest_filename
def foc_path(timestamp, dry=False): def foc_path(timestamp, dry=False):
...@@ -107,14 +93,13 @@ def foc_path(timestamp, dry=False): ...@@ -107,14 +93,13 @@ def foc_path(timestamp, dry=False):
Create the focus output text file name and its path Create the focus output text file name and its path
""" """
root = PROJECT_ROOT / DATA_FOLDER_NAME / focus_folder path = PROJECT_ROOT / DATA_FOLDER / FOCUS_FOLDER
path = root
if not dry: if not dry:
path.mkdir(parents=True, exist_ok=True) path.mkdir(parents=True, exist_ok=True)
# OARPAF.YYYY-MM-DD.foc # OARPAF.YYYY-MM-DD.foc
outfile = f"{FILE_PREFIX}{timestamp}{focus_ext}" outfile = f"{FILE_PREFIX}.{timestamp}.{FOCUS_EXT}"
outpath = path / outfile outpath = path / outfile
return outpath return outpath
...@@ -131,7 +116,7 @@ def save_filename(infile_path_str): ...@@ -131,7 +116,7 @@ def save_filename(infile_path_str):
date_obs_str = header[dateobs] # DATE-OBS from FITS header date_obs_str = header[dateobs] # DATE-OBS from FITS header
name_for_file = Time(date_obs_str).isot name_for_file = Time(date_obs_str).isot
outfile_name = f"{FILE_PREFIX}{name_for_file}{ext}" outfile_name = f"{FILE_PREFIX}.{name_for_file}.{FITS_EXT}"
outfile = Path(outfile_name) outfile = Path(outfile_name)
outdir = fits_path(header) # This already creates the directory outdir = fits_path(header) # This already creates the directory
......
...@@ -27,6 +27,7 @@ classifiers = [ ...@@ -27,6 +27,7 @@ classifiers = [
dependencies = [ dependencies = [
"requests", "requests",
"astropy", "astropy",
"loguru",
"gnuplotlib", ## If focus.py "gnuplotlib", ## If focus.py
"pyvantagepro", # If meteo.py "pyvantagepro", # If meteo.py
] ]
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment