Loading noctua/defaults/testpause.json +6 −0 Original line number Diff line number Diff line Loading @@ -4,5 +4,11 @@ "params": { "repeat": "ciao" } }, { "template" : "testpause", "params": { "repeat": "ciccio" } } ] noctua/sequencer.py +43 −25 Original line number Diff line number Diff line Loading @@ -35,7 +35,7 @@ class Sequencer(): self.quitting = False self.embedded = embedded # To manage sys.exit if called alone signal.signal(signal.SIGINT, self.interrupt) self.original_sigint = signal.getsignal(signal.SIGINT) # Store original self.original_sigint = signal.getsignal(signal.SIGINT) def load_script(self, template_script_path, params={}): '''Load a python file where the template is implemented''' Loading Loading @@ -140,7 +140,6 @@ class Sequencer(): continue # Skip to next template if malformed # Importing the module given its name in the json. # Assumes templates are in noctua.templates full_module_name = f"noctua.templates.{template_name}" try: tplmodule = importlib.import_module(full_module_name) Loading Loading @@ -191,16 +190,34 @@ class Sequencer(): example to modify a pause attribute or call a method. ''' # Temporarily restore original handler to avoid nested inputs if Ctrl+C # is hit again # signal.signal(signal.SIGINT, self.original_sigint) try: # THIS IS WHERE THE MENU SHOULD BE PRINTED sys.stderr.write("\n--- INTERRUPT ---\n") sys.stderr.write("(P)ause, (R)esume, (T)ry again paragraph, (N)ext template, (Q)uit:\n") sys.stderr.write("(P)ause, (R)esume, (N)ext template, (Q)uit:\n") sys.stderr.flush() answer = sys.stdin.readline().strip().lower() raw_line = sys.stdin.readline() try: # answer = input( # "(P)ause, (R)esume, (N)ext template, (Q)uit:\n") if not raw_line: # THIS IS THE CHECK FOR EOF (Ctrl+D) msg = "SEQUENCER: CTRL-D (EOF) detected with readline!" log.critical(msg) if not self.embedded: log.info("SEQUENCER: EOF in non-embedded mode. Forcing exit.") if hasattr(self, 'tpl') and self.tpl and hasattr(self.tpl, 'abort'): try: self.tpl.abort() except Exception: pass # Best effort os._exit(1) else: log.info("SEQUENCER: EOF in embedded mode. Signaling quit.") self.quit() return # Exit the interrupt handler answer = raw_line.strip().lower() if answer.startswith('p'): self.pause() Loading @@ -213,22 +230,23 @@ class Sequencer(): else: log.info("SEQUENCER: No valid action taken on interrupt.") # except EOFError: # CTRL+D # msg = "SEQUENCER: CTRL-D detected! Exiting for real!" # log.critical(msg) # if not self.embedded: # os._exit(1) # Force exit if not embedded # else: # self.quit() # Attempt graceful quit if embedded except (KeyboardInterrupt, RuntimeError): # If Ctrl+C is hit again during input log.warning( "SEQUENCER: Interrupted during interrupt handling. Quitting.") self.quit() except EOFError: # CTRL+D msg = "SEQUENCER: CTRL-D detected! Exiting for real!" log.critical(msg) if not self.embedded: os._exit(1) # Force exit if not embedded else: self.quit() # Attempt graceful quit if embedded # except Exception as e: # Add a broad catch here for debugging # sys.stderr.write(f"ERROR IN INTERRUPT HANDLER: {type(e).__name__}: {e}\n") # log.error("Error in interrupt handler", exc_info=True) # # Fallback to quitting if the handler itself fails # self.quit() except Exception as e: # Add a broad catch here for debugging sys.stderr.write(f"ERROR IN INTERRUPT HANDLER: {type(e).__name__}: {e}\n") log.error("Error in interrupt handler", exc_info=True) # Fallback to quitting if the handler itself fails self.quit() finally: signal.signal(signal.SIGINT, self.interrupt) # Re-hook Loading Loading @@ -277,9 +295,9 @@ class Sequencer(): self.ob_file}." log.error(msg) self.error.append(msg) self.tpl.aborted = True # Signal the template to stop its current paragraph # self.tpl.abort() # Call template's specific abort logic (e.g., # stop camera) self.tpl.aborted = True # Signal the template to stop its # current paragraph self.tpl.abort() # Call template's # specific abort logic (e.g., stop camera) else: log.warning( "SEQUENCER: No active template to abort or template doesn't support aborting.") Loading noctua/utils/logger.py +21 −76 Original line number Diff line number Diff line Loading @@ -13,12 +13,9 @@ from datetime import datetime, time # Third-party modules # 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 ..config.constants import ( 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 Loading @@ -42,8 +39,6 @@ class ColorizingFormatter(logging.Formatter): 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 Loading @@ -60,15 +55,15 @@ class ColorizingFormatter(logging.Formatter): color = self.LEVEL_COLORS.get(record.levelno) if color: padded_levelname = f"{record.levelname:<8}" # Adjust padding if different 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 record.levelname in s: colored_levelname = f"{color}{record.levelname}{self.RESET_SEQ}" s = s.replace(record.levelname, colored_levelname, 1) if levelname_str in s: colored_levelname = f"{color}{levelname_str}{self.RESET_SEQ}" s = s.replace(levelname_str, colored_levelname, 1) return s Loading Loading @@ -99,7 +94,7 @@ def setup_logger(name="noctua_logger", log_level=logging.DEBUG): 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 tzinfo=None ) rotating_file_handler = logging.handlers.TimedRotatingFileHandler( Loading @@ -109,50 +104,20 @@ def setup_logger(name="noctua_logger", log_level=logging.DEBUG): 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 utc=True ) 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 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() # 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: Loading @@ -160,40 +125,17 @@ def setup_logger(name="noctua_logger", log_level=logging.DEBUG): 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" # 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 # 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. Loading @@ -208,7 +150,7 @@ def setup_logger(name="noctua_logger", log_level=logging.DEBUG): # 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 # 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)) Loading @@ -226,6 +168,9 @@ from datetime import timedelta # Ensure timedelta is imported for namer log = setup_logger() if __name__ == "__main__": log.debug("This is a debug message from logger.py main.") log.debug("This is a debug message.") log.info("This is an info message.") # ... rest of your test code ... log.warning("This is a warning message.") log.error("This is an error message.") log.critical("This is a critical message.") Loading
noctua/defaults/testpause.json +6 −0 Original line number Diff line number Diff line Loading @@ -4,5 +4,11 @@ "params": { "repeat": "ciao" } }, { "template" : "testpause", "params": { "repeat": "ciccio" } } ]
noctua/sequencer.py +43 −25 Original line number Diff line number Diff line Loading @@ -35,7 +35,7 @@ class Sequencer(): self.quitting = False self.embedded = embedded # To manage sys.exit if called alone signal.signal(signal.SIGINT, self.interrupt) self.original_sigint = signal.getsignal(signal.SIGINT) # Store original self.original_sigint = signal.getsignal(signal.SIGINT) def load_script(self, template_script_path, params={}): '''Load a python file where the template is implemented''' Loading Loading @@ -140,7 +140,6 @@ class Sequencer(): continue # Skip to next template if malformed # Importing the module given its name in the json. # Assumes templates are in noctua.templates full_module_name = f"noctua.templates.{template_name}" try: tplmodule = importlib.import_module(full_module_name) Loading Loading @@ -191,16 +190,34 @@ class Sequencer(): example to modify a pause attribute or call a method. ''' # Temporarily restore original handler to avoid nested inputs if Ctrl+C # is hit again # signal.signal(signal.SIGINT, self.original_sigint) try: # THIS IS WHERE THE MENU SHOULD BE PRINTED sys.stderr.write("\n--- INTERRUPT ---\n") sys.stderr.write("(P)ause, (R)esume, (T)ry again paragraph, (N)ext template, (Q)uit:\n") sys.stderr.write("(P)ause, (R)esume, (N)ext template, (Q)uit:\n") sys.stderr.flush() answer = sys.stdin.readline().strip().lower() raw_line = sys.stdin.readline() try: # answer = input( # "(P)ause, (R)esume, (N)ext template, (Q)uit:\n") if not raw_line: # THIS IS THE CHECK FOR EOF (Ctrl+D) msg = "SEQUENCER: CTRL-D (EOF) detected with readline!" log.critical(msg) if not self.embedded: log.info("SEQUENCER: EOF in non-embedded mode. Forcing exit.") if hasattr(self, 'tpl') and self.tpl and hasattr(self.tpl, 'abort'): try: self.tpl.abort() except Exception: pass # Best effort os._exit(1) else: log.info("SEQUENCER: EOF in embedded mode. Signaling quit.") self.quit() return # Exit the interrupt handler answer = raw_line.strip().lower() if answer.startswith('p'): self.pause() Loading @@ -213,22 +230,23 @@ class Sequencer(): else: log.info("SEQUENCER: No valid action taken on interrupt.") # except EOFError: # CTRL+D # msg = "SEQUENCER: CTRL-D detected! Exiting for real!" # log.critical(msg) # if not self.embedded: # os._exit(1) # Force exit if not embedded # else: # self.quit() # Attempt graceful quit if embedded except (KeyboardInterrupt, RuntimeError): # If Ctrl+C is hit again during input log.warning( "SEQUENCER: Interrupted during interrupt handling. Quitting.") self.quit() except EOFError: # CTRL+D msg = "SEQUENCER: CTRL-D detected! Exiting for real!" log.critical(msg) if not self.embedded: os._exit(1) # Force exit if not embedded else: self.quit() # Attempt graceful quit if embedded # except Exception as e: # Add a broad catch here for debugging # sys.stderr.write(f"ERROR IN INTERRUPT HANDLER: {type(e).__name__}: {e}\n") # log.error("Error in interrupt handler", exc_info=True) # # Fallback to quitting if the handler itself fails # self.quit() except Exception as e: # Add a broad catch here for debugging sys.stderr.write(f"ERROR IN INTERRUPT HANDLER: {type(e).__name__}: {e}\n") log.error("Error in interrupt handler", exc_info=True) # Fallback to quitting if the handler itself fails self.quit() finally: signal.signal(signal.SIGINT, self.interrupt) # Re-hook Loading Loading @@ -277,9 +295,9 @@ class Sequencer(): self.ob_file}." log.error(msg) self.error.append(msg) self.tpl.aborted = True # Signal the template to stop its current paragraph # self.tpl.abort() # Call template's specific abort logic (e.g., # stop camera) self.tpl.aborted = True # Signal the template to stop its # current paragraph self.tpl.abort() # Call template's # specific abort logic (e.g., stop camera) else: log.warning( "SEQUENCER: No active template to abort or template doesn't support aborting.") Loading
noctua/utils/logger.py +21 −76 Original line number Diff line number Diff line Loading @@ -13,12 +13,9 @@ from datetime import datetime, time # Third-party modules # 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 ..config.constants import ( 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 Loading @@ -42,8 +39,6 @@ class ColorizingFormatter(logging.Formatter): 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 Loading @@ -60,15 +55,15 @@ class ColorizingFormatter(logging.Formatter): color = self.LEVEL_COLORS.get(record.levelno) if color: padded_levelname = f"{record.levelname:<8}" # Adjust padding if different 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 record.levelname in s: colored_levelname = f"{color}{record.levelname}{self.RESET_SEQ}" s = s.replace(record.levelname, colored_levelname, 1) if levelname_str in s: colored_levelname = f"{color}{levelname_str}{self.RESET_SEQ}" s = s.replace(levelname_str, colored_levelname, 1) return s Loading Loading @@ -99,7 +94,7 @@ def setup_logger(name="noctua_logger", log_level=logging.DEBUG): 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 tzinfo=None ) rotating_file_handler = logging.handlers.TimedRotatingFileHandler( Loading @@ -109,50 +104,20 @@ def setup_logger(name="noctua_logger", log_level=logging.DEBUG): 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 utc=True ) 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 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() # 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: Loading @@ -160,40 +125,17 @@ def setup_logger(name="noctua_logger", log_level=logging.DEBUG): 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" # 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 # 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. Loading @@ -208,7 +150,7 @@ def setup_logger(name="noctua_logger", log_level=logging.DEBUG): # 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 # 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)) Loading @@ -226,6 +168,9 @@ from datetime import timedelta # Ensure timedelta is imported for namer log = setup_logger() if __name__ == "__main__": log.debug("This is a debug message from logger.py main.") log.debug("This is a debug message.") log.info("This is an info message.") # ... rest of your test code ... log.warning("This is a warning message.") log.error("This is an error message.") log.critical("This is a critical message.")