Skip to content

lstautorta.merge_DL3

Script to merge DL3 and analyse them.

Usage: $> python merge_DL3.py --input-filter "dl3_LST-1.Run04190.*.fits" --input-directory "./DL3/" --output-dir "./DL3/" --run_id 9864

Functions:

Name Description
setup_logging

Configure logging with one log file per execution.

setup_logging 

setup_logging(output_dir, run_id)

Configure logging with one log file per execution.

If output_dir does not exist, write the log in a fallback location and raise FileNotFoundError so the script crashes immediately.

Source code in src/lstautorta/merge_DL3.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
def setup_logging(output_dir, run_id):
    """
    Configure logging with one log file per execution.

    If output_dir does not exist, write the log in a fallback location
    and raise FileNotFoundError so the script crashes immediately.
    """
    logger = logging.getLogger("merge_DL3")
    logger.setLevel(logging.INFO)
    logger.handlers.clear()

    formatter = logging.Formatter(
        fmt="%(asctime)s | %(levelname)s | %(message)s",
        datefmt="%Y-%m-%d %H:%M:%S",
    )

    # Always keep stdout logging
    stream_handler = logging.StreamHandler(sys.stdout)
    stream_handler.setLevel(logging.INFO)
    stream_handler.setFormatter(formatter)
    logger.addHandler(stream_handler)

    output_path = Path(output_dir)

    if not output_path.exists():
        fallback_log = Path.cwd() / f"merge_DL3_run{run_id}.log"

        file_handler = logging.FileHandler(fallback_log)
        file_handler.setLevel(logging.INFO)
        file_handler.setFormatter(formatter)
        logger.addHandler(file_handler)

        logger.info("============================================================")
        logger.error("Output directory does not exist: %s", output_path)
        logger.error("Fallback log file: %s", fallback_log)
        logger.error("Crashing because output directory is missing.")
        logger.info("============================================================")

        raise FileNotFoundError(f"Output directory does not exist: {output_path}")

    log_dir = output_path / "logs"
    log_dir.mkdir(exist_ok=True)

    log_file = log_dir / f"merge_DL3_run{run_id}.log"

    file_handler = logging.FileHandler(log_file)
    file_handler.setLevel(logging.INFO)
    file_handler.setFormatter(formatter)
    logger.addHandler(file_handler)

    logger.info("============================================================")
    logger.info("Starting merge_DL3.py")
    logger.info("Log file: %s", log_file)

    return logger, log_file