Loading noctua/config/constants.py +26 −8 Original line number Diff line number Diff line #!/usr/bin/env python3 # -*- coding: utf-8 -*- ''' Constants used in the whole project ''' # Telescope lat = 44.5912 # [°] Latitude North from Greenwich. Loading Loading @@ -109,10 +113,13 @@ image_number = {v: k for k, v in image_state.items()} # Directories ############ data_folder = "data" DATA_FOLDER_NAME = "data" LOG_DIRECTORY_NAME = "log" fits_folder = "fits" focus_folder = "focus" log_folder = "log" # data_folder = "data" # log_folder = "log" dir_type = { 0: "dark", Loading @@ -124,6 +131,21 @@ dir_type = { # Reversing dir_type 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 ############ Loading @@ -132,10 +154,6 @@ dir_number = {v: k for k, v in dir_type.items()} imagetyp = "IMAGETYP" # 'Light' dateobs = "DATE-OBS" # '2021-12-18T05:09:56.163' # File prefix prefix = "OARPAF." # File extension ext = ".fits" log_ext = ".log" FILE_PREFIX = "OARPAF." ext = ".fits" # File extension focus_ext = ".foc" noctua/sequencer.py +2 −3 Original line number Diff line number Diff line Loading @@ -8,6 +8,7 @@ Sequencer module for Observation Blocks (OBs) in JSON format # System modules import argparse import importlib import importlib.resources import json import os import signal Loading Loading @@ -396,7 +397,5 @@ def cli(): if __name__ == "__main__": ''' If called alone ''' # If called alone cli() noctua/utils/analysis.py +3 −6 Original line number Diff line number Diff line Loading @@ -152,14 +152,11 @@ def fit_star( gp.plotimage( (data, dict( xlabel="data")), (data, {"xlabel": "data"}), (fitted_model(x, y), dict( xlabel="model")), (fitted_model(x, y), {"xlabel": "model"}), (residuals, dict( xlabel="Residuals")), (residuals, {"xlabel": "Residuals"}), unset="grid", set="size square", Loading noctua/utils/logger.py +221 −45 Original line number Diff line number Diff line #!/usr/bin/env python #!/usr/bin/env python3 # -*- coding: utf-8 -*- """Custom format log""" # System modules import logging import logging.handlers import sys from pathlib import Path from datetime import datetime, time # Third-party modules from loguru import logger # Other templates from .structure import log_path def mylog(): "logger function" logger.remove() time = "{time:YYYY-MM-DD HH:mm:ss.SSSSSS!UTC} " level = "<level>{level: <8}</level> " message = "| {message} " stack = "<bold>({module}.{function}:{line})</bold>" fmt = time + level + message + stack # On standard output logger.add(sys.stderr, format=fmt) # On file log_day = log_path("{time:YYYY-MM-DD!UTC}") logger.add(log_day, format=fmt, colorize=True, rotation="1 day") # Custom levels logger.level("DEBUG", color="<magenta><bold>") logger.level("INFO", color="<green><bold>") return logger def main(): """Main function""" log = mylog() log.debug("debug message") log.info("info message") log.warning("warning message") log.error("error message") log.critical("critical message") # Custom modules from ..config.constants import ( LOG_DIRECTORY_NAME, DATA_FOLDER_NAME, FILE_PREFIX, LOG_FILE_EXTENSION, LATEST_LOG_FILENAME, LOG_ROTATION_TIME_H, LOG_ROTATION_TIME_M, LOG_ROTATION_TIME_S, LOG_ROTATION_INTERVAL_DAYS, LOG_BACKUP_COUNT ) from .structure import get_log_dir_path, get_dated_log_filepath, get_latest_log_filepath LOG_DIR = get_log_dir_path() # Creates directory if it doesn't exist LATEST_LOG_FILEPATH = get_latest_log_filepath() log = None # Global logger instance class ColorizingFormatter(logging.Formatter): """ A custom logging formatter that adds ANSI color codes to log level names for console output. """ # Color codes BLACK, RED, GREEN, YELLOW, BLUE, MAGENTA, CYAN, WHITE = range(8) # 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) # Let's use bold bright red: f"\x1b[31;1;91m" # Or just very bold red: f"\x1b[31;1m" - let's stick to simpler bold red for CRITICAL } 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"{record.levelname:<8}" # Adjust padding if different 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 record.levelname in s: colored_levelname = f"{color}{record.levelname}{self.RESET_SEQ}" s = s.replace(record.levelname, 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 # Naive time, assumed UTC by the handler when utc=True ) rotating_file_handler = logging.handlers.TimedRotatingFileHandler( filename=LATEST_LOG_FILEPATH, when="midnight", # Base rotation type (daily) atTime=rotation_time_utc, # Specific time for rotation interval=LOG_ROTATION_INTERVAL_DAYS, # Interval (1 for daily) backupCount=LOG_BACKUP_COUNT, # 0 means keep all backups encoding='utf-8', utc=True # IMPORTANT: Use UTC for rotation calculations and atTime ) def custom_namer_astronomical(default_name): # default_name will be like /path/to/OARPAF.latest.log.YYYY-MM-DD_HH-MM-SS (if atTime is used) # or /path/to/OARPAF.latest.log.YYYY-MM-DD (if when='midnight' and no atTime) # We want /path/to/OARPAF.ASTRO_DATE.log where ASTRO_DATE changes at midday. # The date embedded by TimedRotatingFileHandler in default_name is the *calendar date* # of the *end* of the logging period (i.e., when rotation occurs). # For midday rotation, if rotation happens on 2023-04-16 at 12:00 UTC, # the log being rotated contains messages from 2023-04-15 12:00 UTC to 2023-04-16 11:59 UTC. # So, the log *pertains* to the astronomical night of 2023-04-15. # Extract the calendar date suffix added by the handler try: # Example: '/.../OARPAF.latest.log.2023-04-16' (if rotated at midnight) # Example: '/.../OARPAF.latest.log.2023-04-16_12-00-00' (if rotated with atTime) base_filepath = Path(default_name) suffix = base_filepath.suffix # .log or .2023-04-16 or .2023-04-16_12-00-00 (depends on exact TimedRotatingFileHandler version and usage) # A more robust way to get the date part that TimedRotatingFileHandler uses for rotation: # The date used for the suffix is based on the rolloverAt time. # We need to determine the "astronomical date" this log corresponds to. # `self.rolloverAt` is when the next rollover will happen. # The file being closed contains logs *before* `self.rolloverAt`. # So, the timestamp of the logs is generally `self.rolloverAt - interval`. # Let's assume the default_name already contains the calendar date of rotation. # Try to parse the date from the default_name # This is a bit fragile as it depends on TimedRotatingFileHandler's naming convention date_str_part = base_filepath.name.replace(LATEST_LOG_FILEPATH.name + '.', '') # Removes base filename and first dot 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() # This rotation_calendar_date is when the rotation *occurred*. # If rotation is at midday (e.g., 2023-04-16 12:00 UTC), # the log file being closed contains data for the astronomical night # that *ended* on this date. So, the astronomical date is one day prior. astronomical_date_of_log = rotation_calendar_date - timedelta(days=1) astro_date_str = astronomical_date_of_log.strftime("%Y-%m-%d") # Use structure.py to build the final path for consistency 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 # For `atTime` to work as expected and `custom_namer_astronomical` to get the correct date, # it's simpler to handle the astronomical date logic more directly if possible. # The `TimedRotatingFileHandler` is primarily designed around calendar days. # A common workaround for astronomical logging is to rotate daily at UTC 00:00 or 12:00 # and then have a *separate script* or process that renames the rotated log files # according to the astronomical date convention if the handler's naming isn't perfect. # Given the complexity of precisely aligning TimedRotatingFileHandler's internal date logic # with astronomical "day" changes via just `namer`, let's use a simpler `namer` # that just cleans up the default TimedRotatingFileHandler name format. # The astronomical date logic is better handled in how you *interpret* or search for logs. def simpler_custom_namer(default_name): # default_name for TimedRotatingFileHandler with atTime and utc=True # is typically `basename.YYYY-MM-DD_HH-MM-SS` where YYYY-MM-DD_HH-MM-SS is the UTC time of rotation. # Or `basename.YYYY-MM-DD` if `when` is 'midnight' and `atTime` is not used, or if `atTime` is midnight. # We want `LOG_FILE_PREFIX + YYYY-MM-DD + LOG_FILE_EXTENSION` # The date in the filename should represent the *start* of the observing night. base_filepath = Path(default_name) # Example default_name: /path/to/OARPAF.latest.log.2023-04-15_12-00-00 # The date/time in the suffix is when the rotation *occurred*. # The logs *before* this time are in this rotated file. # Extract the date part from the suffix # Suffix might be ".YYYY-MM-DD" or ".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 # This date_str is the calendar date when the file was *closed* (rotation happened). # For midday rotation (e.g., 12:00 UTC on YYYY-MM-DD), the logs in this file # pertain to the astronomical night of YYYY-MM-(DD-1). 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 or 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__": main() else: log = mylog() log.debug("This is a debug message from logger.py main.") log.info("This is an info message.") # ... rest of your test code ... noctua/utils/structure.py +71 −81 Original line number Diff line number Diff line Loading @@ -13,49 +13,44 @@ from pathlib import Path from astropy.io import fits from astropy.time import Time # Other templates # from utils.logger import log from ..config.constants import (data_folder, dateobs, dir_type, ext, fits_folder, focus_ext, focus_folder, frame_number, imagetyp, log_ext, log_folder, prefix) # Custom modules from ..config.constants import ( DATA_FOLDER_NAME, LOG_DIRECTORY_NAME, FILE_PREFIX, LOG_FILE_EXTENSION, LATEST_LOG_FILENAME, dateobs, dir_type, ext, fits_folder, frame_number, imagetyp, focus_folder, focus_ext ) PROJECT_ROOT = Path(__file__).parent.parent.parent def date_folder(): """Create a date folder string based on astronomical convention (changes at midday UTC). """ Create a date folder """ now = Time.now() # 2021-12-28T10:00:00.123456 if now.datetime.hour < 12: # from midnight to midday... folder_date = now - 1 # ...subtract one day now = Time.now() # If current UTC hour is before midday UTC, the "observing night" # belongs to the previous calendar date. if now.datetime.hour < 12: folder_date_obj = now.datetime.date() - timedelta(days=1) else: folder_date = now # 2021-12-27 folder_name = str(folder_date).split()[0] return folder_name folder_date_obj = now.datetime.date() return folder_date_obj.strftime("%Y-%m-%d") def frame_folder(header): """ Create a folder depending on the image type in FITS header """ frame = header[imagetyp] # Takes # frame_type or frame_number if isinstance(frame, int): # 1 -> object/ folder_name = dir_type[frame] else: # str # 'Light Frame' -> Light -> [1] -> 1 frame = [v for k, v in frame_number.items() if k in frame][0] # 1 -> object/ folder_name = dir_type[frame] else: frame_num_list = [v for k, v in frame_number.items() if k in frame] if not frame_num_list: # Fallback if frame type string is not recognized return "unknown_type" folder_name = dir_type[frame_num_list[0]] return folder_name Loading @@ -64,46 +59,47 @@ def fits_path(header, dry=False): Create a fits file path where the file will be stored """ # data/fits/ root = Path(data_folder, fits_folder) root = PROJECT_ROOT / DATA_FOLDER_NAME / fits_folder # 2021-12-27 date = date_folder() date = Path(date) date_str = date_folder() date_path_part = Path(date_str) # 'Light Frame' -> Light -> object/ frame = frame_folder(header) frame = Path(frame) # data/fits/2021-12-27/object/OARPAF.blabla.fits path = [root, date, frame] path = Path.joinpath(*path) frame_path_part = Path(frame_folder(header)) path = root / date_path_part / frame_path_part if not dry: path.mkdir(parents=True, exist_ok=True) return path def log_path(timestamp, dry=False): def get_log_dir_path(dry=False): """ Create the log file name and its path Returns the Path object for the log directory. Creates it if it doesn't exist (unless dry=True). """ # data/log/ root = Path(data_folder, log_folder) path = root log_dir = PROJECT_ROOT / DATA_FOLDER_NAME / LOG_DIRECTORY_NAME if not dry: path.mkdir(parents=True, exist_ok=True) log_dir.mkdir(parents=True, exist_ok=True) return log_dir # OARPAF.2021-12-27.log outfile = prefix + timestamp + log_ext def get_dated_log_filepath(date_str, dry=False): """ Constructs a path for an expected date-stamped (rotated) log file. date_str: A string like "YYYY-MM-DD" """ # data/log/OARPAF.2021-12-27.log outpath = Path.joinpath(path, outfile) log_dir = get_log_dir_path(dry=dry) # Filename: OARPAF.YYYY-MM-DD.log log_filename = f"{FILE_PREFIX}{date_str}{LOG_FILE_EXTENSION}" return log_dir / log_filename return outpath def get_latest_log_filepath(dry=False): """ Constructs the path for the "latest" log file. """ log_dir = get_log_dir_path(dry=dry) latest_filename = f"{FILE_PREFIX}{LATEST_LOG_FILENAME}{LOG_FILE_EXTENSION}" return log_dir / latest_filename def foc_path(timestamp, dry=False): Loading @@ -111,41 +107,35 @@ def foc_path(timestamp, dry=False): Create the focus output text file name and its path """ # data/focus/ root = Path(data_folder, focus_folder) root = PROJECT_ROOT / DATA_FOLDER_NAME / focus_folder path = root if not dry: path.mkdir(parents=True, exist_ok=True) # OARPAF.2021-12-27.foc outfile = prefix + timestamp + focus_ext # data/focus/OARPAF.2021-12-27.foc outpath = Path.joinpath(path, outfile) # OARPAF.YYYY-MM-DD.foc outfile = f"{FILE_PREFIX}{timestamp}{focus_ext}" outpath = path / outfile return outpath def save_filename(infile): def save_filename(infile_path_str): """ Save a fits file in its path with an ESO-style filename. """ inpath = Path(infile) header = fits.getheader(infile) inpath = Path(infile_path_str) header = fits.getheader(inpath) # '2021-12-28T20:09:56.163' name = Time(header[dateobs]).isot # "OARPAF." + name + ".fits" outfile = Path(prefix + name + ext) # data/fits/2021-12-27/object/OARPAF.blabla.fits outdir = fits_path(header) date_obs_str = header[dateobs] # DATE-OBS from FITS header name_for_file = Time(date_obs_str).isot outpath = Path.joinpath(outdir, outfile) outfile_name = f"{FILE_PREFIX}{name_for_file}{ext}" outfile = Path(outfile_name) shutil.copy2(inpath, outpath) # For Python 3.8+. outdir = fits_path(header) # This already creates the directory outpath = outdir / outfile return outpath shutil.copy2(inpath, outpath) return str(outpath) Loading
noctua/config/constants.py +26 −8 Original line number Diff line number Diff line #!/usr/bin/env python3 # -*- coding: utf-8 -*- ''' Constants used in the whole project ''' # Telescope lat = 44.5912 # [°] Latitude North from Greenwich. Loading Loading @@ -109,10 +113,13 @@ image_number = {v: k for k, v in image_state.items()} # Directories ############ data_folder = "data" DATA_FOLDER_NAME = "data" LOG_DIRECTORY_NAME = "log" fits_folder = "fits" focus_folder = "focus" log_folder = "log" # data_folder = "data" # log_folder = "log" dir_type = { 0: "dark", Loading @@ -124,6 +131,21 @@ dir_type = { # Reversing dir_type 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 ############ Loading @@ -132,10 +154,6 @@ dir_number = {v: k for k, v in dir_type.items()} imagetyp = "IMAGETYP" # 'Light' dateobs = "DATE-OBS" # '2021-12-18T05:09:56.163' # File prefix prefix = "OARPAF." # File extension ext = ".fits" log_ext = ".log" FILE_PREFIX = "OARPAF." ext = ".fits" # File extension focus_ext = ".foc"
noctua/sequencer.py +2 −3 Original line number Diff line number Diff line Loading @@ -8,6 +8,7 @@ Sequencer module for Observation Blocks (OBs) in JSON format # System modules import argparse import importlib import importlib.resources import json import os import signal Loading Loading @@ -396,7 +397,5 @@ def cli(): if __name__ == "__main__": ''' If called alone ''' # If called alone cli()
noctua/utils/analysis.py +3 −6 Original line number Diff line number Diff line Loading @@ -152,14 +152,11 @@ def fit_star( gp.plotimage( (data, dict( xlabel="data")), (data, {"xlabel": "data"}), (fitted_model(x, y), dict( xlabel="model")), (fitted_model(x, y), {"xlabel": "model"}), (residuals, dict( xlabel="Residuals")), (residuals, {"xlabel": "Residuals"}), unset="grid", set="size square", Loading
noctua/utils/logger.py +221 −45 Original line number Diff line number Diff line #!/usr/bin/env python #!/usr/bin/env python3 # -*- coding: utf-8 -*- """Custom format log""" # System modules import logging import logging.handlers import sys from pathlib import Path from datetime import datetime, time # Third-party modules from loguru import logger # Other templates from .structure import log_path def mylog(): "logger function" logger.remove() time = "{time:YYYY-MM-DD HH:mm:ss.SSSSSS!UTC} " level = "<level>{level: <8}</level> " message = "| {message} " stack = "<bold>({module}.{function}:{line})</bold>" fmt = time + level + message + stack # On standard output logger.add(sys.stderr, format=fmt) # On file log_day = log_path("{time:YYYY-MM-DD!UTC}") logger.add(log_day, format=fmt, colorize=True, rotation="1 day") # Custom levels logger.level("DEBUG", color="<magenta><bold>") logger.level("INFO", color="<green><bold>") return logger def main(): """Main function""" log = mylog() log.debug("debug message") log.info("info message") log.warning("warning message") log.error("error message") log.critical("critical message") # Custom modules from ..config.constants import ( LOG_DIRECTORY_NAME, DATA_FOLDER_NAME, FILE_PREFIX, LOG_FILE_EXTENSION, LATEST_LOG_FILENAME, LOG_ROTATION_TIME_H, LOG_ROTATION_TIME_M, LOG_ROTATION_TIME_S, LOG_ROTATION_INTERVAL_DAYS, LOG_BACKUP_COUNT ) from .structure import get_log_dir_path, get_dated_log_filepath, get_latest_log_filepath LOG_DIR = get_log_dir_path() # Creates directory if it doesn't exist LATEST_LOG_FILEPATH = get_latest_log_filepath() log = None # Global logger instance class ColorizingFormatter(logging.Formatter): """ A custom logging formatter that adds ANSI color codes to log level names for console output. """ # Color codes BLACK, RED, GREEN, YELLOW, BLUE, MAGENTA, CYAN, WHITE = range(8) # 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) # Let's use bold bright red: f"\x1b[31;1;91m" # Or just very bold red: f"\x1b[31;1m" - let's stick to simpler bold red for CRITICAL } 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"{record.levelname:<8}" # Adjust padding if different 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 record.levelname in s: colored_levelname = f"{color}{record.levelname}{self.RESET_SEQ}" s = s.replace(record.levelname, 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 # Naive time, assumed UTC by the handler when utc=True ) rotating_file_handler = logging.handlers.TimedRotatingFileHandler( filename=LATEST_LOG_FILEPATH, when="midnight", # Base rotation type (daily) atTime=rotation_time_utc, # Specific time for rotation interval=LOG_ROTATION_INTERVAL_DAYS, # Interval (1 for daily) backupCount=LOG_BACKUP_COUNT, # 0 means keep all backups encoding='utf-8', utc=True # IMPORTANT: Use UTC for rotation calculations and atTime ) def custom_namer_astronomical(default_name): # default_name will be like /path/to/OARPAF.latest.log.YYYY-MM-DD_HH-MM-SS (if atTime is used) # or /path/to/OARPAF.latest.log.YYYY-MM-DD (if when='midnight' and no atTime) # We want /path/to/OARPAF.ASTRO_DATE.log where ASTRO_DATE changes at midday. # The date embedded by TimedRotatingFileHandler in default_name is the *calendar date* # of the *end* of the logging period (i.e., when rotation occurs). # For midday rotation, if rotation happens on 2023-04-16 at 12:00 UTC, # the log being rotated contains messages from 2023-04-15 12:00 UTC to 2023-04-16 11:59 UTC. # So, the log *pertains* to the astronomical night of 2023-04-15. # Extract the calendar date suffix added by the handler try: # Example: '/.../OARPAF.latest.log.2023-04-16' (if rotated at midnight) # Example: '/.../OARPAF.latest.log.2023-04-16_12-00-00' (if rotated with atTime) base_filepath = Path(default_name) suffix = base_filepath.suffix # .log or .2023-04-16 or .2023-04-16_12-00-00 (depends on exact TimedRotatingFileHandler version and usage) # A more robust way to get the date part that TimedRotatingFileHandler uses for rotation: # The date used for the suffix is based on the rolloverAt time. # We need to determine the "astronomical date" this log corresponds to. # `self.rolloverAt` is when the next rollover will happen. # The file being closed contains logs *before* `self.rolloverAt`. # So, the timestamp of the logs is generally `self.rolloverAt - interval`. # Let's assume the default_name already contains the calendar date of rotation. # Try to parse the date from the default_name # This is a bit fragile as it depends on TimedRotatingFileHandler's naming convention date_str_part = base_filepath.name.replace(LATEST_LOG_FILEPATH.name + '.', '') # Removes base filename and first dot 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() # This rotation_calendar_date is when the rotation *occurred*. # If rotation is at midday (e.g., 2023-04-16 12:00 UTC), # the log file being closed contains data for the astronomical night # that *ended* on this date. So, the astronomical date is one day prior. astronomical_date_of_log = rotation_calendar_date - timedelta(days=1) astro_date_str = astronomical_date_of_log.strftime("%Y-%m-%d") # Use structure.py to build the final path for consistency 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 # For `atTime` to work as expected and `custom_namer_astronomical` to get the correct date, # it's simpler to handle the astronomical date logic more directly if possible. # The `TimedRotatingFileHandler` is primarily designed around calendar days. # A common workaround for astronomical logging is to rotate daily at UTC 00:00 or 12:00 # and then have a *separate script* or process that renames the rotated log files # according to the astronomical date convention if the handler's naming isn't perfect. # Given the complexity of precisely aligning TimedRotatingFileHandler's internal date logic # with astronomical "day" changes via just `namer`, let's use a simpler `namer` # that just cleans up the default TimedRotatingFileHandler name format. # The astronomical date logic is better handled in how you *interpret* or search for logs. def simpler_custom_namer(default_name): # default_name for TimedRotatingFileHandler with atTime and utc=True # is typically `basename.YYYY-MM-DD_HH-MM-SS` where YYYY-MM-DD_HH-MM-SS is the UTC time of rotation. # Or `basename.YYYY-MM-DD` if `when` is 'midnight' and `atTime` is not used, or if `atTime` is midnight. # We want `LOG_FILE_PREFIX + YYYY-MM-DD + LOG_FILE_EXTENSION` # The date in the filename should represent the *start* of the observing night. base_filepath = Path(default_name) # Example default_name: /path/to/OARPAF.latest.log.2023-04-15_12-00-00 # The date/time in the suffix is when the rotation *occurred*. # The logs *before* this time are in this rotated file. # Extract the date part from the suffix # Suffix might be ".YYYY-MM-DD" or ".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 # This date_str is the calendar date when the file was *closed* (rotation happened). # For midday rotation (e.g., 12:00 UTC on YYYY-MM-DD), the logs in this file # pertain to the astronomical night of YYYY-MM-(DD-1). 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 or 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__": main() else: log = mylog() log.debug("This is a debug message from logger.py main.") log.info("This is an info message.") # ... rest of your test code ...
noctua/utils/structure.py +71 −81 Original line number Diff line number Diff line Loading @@ -13,49 +13,44 @@ from pathlib import Path from astropy.io import fits from astropy.time import Time # Other templates # from utils.logger import log from ..config.constants import (data_folder, dateobs, dir_type, ext, fits_folder, focus_ext, focus_folder, frame_number, imagetyp, log_ext, log_folder, prefix) # Custom modules from ..config.constants import ( DATA_FOLDER_NAME, LOG_DIRECTORY_NAME, FILE_PREFIX, LOG_FILE_EXTENSION, LATEST_LOG_FILENAME, dateobs, dir_type, ext, fits_folder, frame_number, imagetyp, focus_folder, focus_ext ) PROJECT_ROOT = Path(__file__).parent.parent.parent def date_folder(): """Create a date folder string based on astronomical convention (changes at midday UTC). """ Create a date folder """ now = Time.now() # 2021-12-28T10:00:00.123456 if now.datetime.hour < 12: # from midnight to midday... folder_date = now - 1 # ...subtract one day now = Time.now() # If current UTC hour is before midday UTC, the "observing night" # belongs to the previous calendar date. if now.datetime.hour < 12: folder_date_obj = now.datetime.date() - timedelta(days=1) else: folder_date = now # 2021-12-27 folder_name = str(folder_date).split()[0] return folder_name folder_date_obj = now.datetime.date() return folder_date_obj.strftime("%Y-%m-%d") def frame_folder(header): """ Create a folder depending on the image type in FITS header """ frame = header[imagetyp] # Takes # frame_type or frame_number if isinstance(frame, int): # 1 -> object/ folder_name = dir_type[frame] else: # str # 'Light Frame' -> Light -> [1] -> 1 frame = [v for k, v in frame_number.items() if k in frame][0] # 1 -> object/ folder_name = dir_type[frame] else: frame_num_list = [v for k, v in frame_number.items() if k in frame] if not frame_num_list: # Fallback if frame type string is not recognized return "unknown_type" folder_name = dir_type[frame_num_list[0]] return folder_name Loading @@ -64,46 +59,47 @@ def fits_path(header, dry=False): Create a fits file path where the file will be stored """ # data/fits/ root = Path(data_folder, fits_folder) root = PROJECT_ROOT / DATA_FOLDER_NAME / fits_folder # 2021-12-27 date = date_folder() date = Path(date) date_str = date_folder() date_path_part = Path(date_str) # 'Light Frame' -> Light -> object/ frame = frame_folder(header) frame = Path(frame) # data/fits/2021-12-27/object/OARPAF.blabla.fits path = [root, date, frame] path = Path.joinpath(*path) frame_path_part = Path(frame_folder(header)) path = root / date_path_part / frame_path_part if not dry: path.mkdir(parents=True, exist_ok=True) return path def log_path(timestamp, dry=False): def get_log_dir_path(dry=False): """ Create the log file name and its path Returns the Path object for the log directory. Creates it if it doesn't exist (unless dry=True). """ # data/log/ root = Path(data_folder, log_folder) path = root log_dir = PROJECT_ROOT / DATA_FOLDER_NAME / LOG_DIRECTORY_NAME if not dry: path.mkdir(parents=True, exist_ok=True) log_dir.mkdir(parents=True, exist_ok=True) return log_dir # OARPAF.2021-12-27.log outfile = prefix + timestamp + log_ext def get_dated_log_filepath(date_str, dry=False): """ Constructs a path for an expected date-stamped (rotated) log file. date_str: A string like "YYYY-MM-DD" """ # data/log/OARPAF.2021-12-27.log outpath = Path.joinpath(path, outfile) log_dir = get_log_dir_path(dry=dry) # Filename: OARPAF.YYYY-MM-DD.log log_filename = f"{FILE_PREFIX}{date_str}{LOG_FILE_EXTENSION}" return log_dir / log_filename return outpath def get_latest_log_filepath(dry=False): """ Constructs the path for the "latest" log file. """ log_dir = get_log_dir_path(dry=dry) latest_filename = f"{FILE_PREFIX}{LATEST_LOG_FILENAME}{LOG_FILE_EXTENSION}" return log_dir / latest_filename def foc_path(timestamp, dry=False): Loading @@ -111,41 +107,35 @@ def foc_path(timestamp, dry=False): Create the focus output text file name and its path """ # data/focus/ root = Path(data_folder, focus_folder) root = PROJECT_ROOT / DATA_FOLDER_NAME / focus_folder path = root if not dry: path.mkdir(parents=True, exist_ok=True) # OARPAF.2021-12-27.foc outfile = prefix + timestamp + focus_ext # data/focus/OARPAF.2021-12-27.foc outpath = Path.joinpath(path, outfile) # OARPAF.YYYY-MM-DD.foc outfile = f"{FILE_PREFIX}{timestamp}{focus_ext}" outpath = path / outfile return outpath def save_filename(infile): def save_filename(infile_path_str): """ Save a fits file in its path with an ESO-style filename. """ inpath = Path(infile) header = fits.getheader(infile) inpath = Path(infile_path_str) header = fits.getheader(inpath) # '2021-12-28T20:09:56.163' name = Time(header[dateobs]).isot # "OARPAF." + name + ".fits" outfile = Path(prefix + name + ext) # data/fits/2021-12-27/object/OARPAF.blabla.fits outdir = fits_path(header) date_obs_str = header[dateobs] # DATE-OBS from FITS header name_for_file = Time(date_obs_str).isot outpath = Path.joinpath(outdir, outfile) outfile_name = f"{FILE_PREFIX}{name_for_file}{ext}" outfile = Path(outfile_name) shutil.copy2(inpath, outpath) # For Python 3.8+. outdir = fits_path(header) # This already creates the directory outpath = outdir / outfile return outpath shutil.copy2(inpath, outpath) return str(outpath)