Add gain selection to OSA workflow

src/osa/configs/sequencer.cfg

@@ -28,6 +28,7 @@ HIGH_LEVEL_DIR: %(OSA_DIR)s/HighLevel
 LONGTERM_DIR: %(OSA_DIR)s/DL1DataCheck_LongTerm
 MERGED_SUMMARY: %(OSA_DIR)s/Catalog/merged_RunSummary.ecsv
 SEQUENCER_WEB_DIR: %(OSA_DIR)s/SequencerWeb
+GAIN_SELECTION_FLAG_DIR: %(OSA_DIR)s/GainSel
 
 # To be set by the user. Using PROD-ID will overcome the automatic
 # fetching of lstchain version. Otherwise leave it empty (and without the colon symbol).
@@ -138,6 +139,7 @@ WORKFLOWPREFIX: Workflow
 GRAPHSUFFIX: .dot
 SVGSUFFIX: .svg
 end_of_activity: NightFinished.txt
+gain_selection_check: GainSelFinished.txt
 
 [OUTPUT]
 # REPORTWIDTH is the width in characters of the heading frame for the output

src/osa/conftest.py

@@ -48,6 +48,13 @@ def monitoring_dir(base_test_dir):
     return monitoring_dir
 
 
+@pytest.fixture(scope="session")
+def osa_dir(base_test_dir):
+    osa_dir = base_test_dir / "OSA"
+    osa_dir.mkdir(parents=True, exist_ok=True)
+    return osa_dir
+
+
 @pytest.fixture(scope="session")
 def run_summary_dir(monitoring_dir):
     summary_dir = monitoring_dir / "RunSummary"
@@ -436,6 +443,8 @@ def sequence_file_list(
     drs4_time_calibration_files,
     systematic_correction_files,
     r0_data,
+    gain_selection_flag_file,
+    merged_run_summary,
 ):
     for r0_file in r0_data:
         assert r0_file.exists()
@@ -448,6 +457,8 @@ def sequence_file_list(
 
     assert run_summary_file.exists()
     assert run_catalog.exists()
+    assert gain_selection_flag_file.exists()
+    assert merged_run_summary.exists()
 
     run_program("sequencer", "-d", "2020-01-17", "--no-submit", "-t", "LST1")
     # First sequence in the list corresponds to the calibration run 1809
@@ -548,11 +559,9 @@ def run_catalog(run_catalog_dir):
 
 
 @pytest.fixture(scope="session")
-def database(base_test_dir):
+def database(osa_dir):
     import sqlite3
 
-    osa_dir = base_test_dir / "OSA"
-    osa_dir.mkdir(parents=True, exist_ok=True)
     db_file = osa_dir / "osa.db"
     with sqlite3.connect(db_file) as connection:
         cursor = connection.cursor()
@@ -562,3 +571,13 @@ def database(base_test_dir):
         )
         cursor.connection.commit()
         yield cursor
+
+
+@pytest.fixture(scope="session")
+def gain_selection_flag_file(osa_dir):
+
+    GainSel_dir = osa_dir / "GainSel" / "20200117"
+    GainSel_dir.mkdir(parents=True, exist_ok=True)
+    file = GainSel_dir / "GainSelFinished.txt"
+    file.touch()
+    return file

src/osa/scripts/gain_selection.py

@@ -14,9 +14,11 @@
 from lstchain.paths import run_info_from_filename, parse_r0_filename
 
 from osa.scripts.reprocessing import get_list_of_dates, check_job_status_and_wait
-from osa.utils.utils import wait_for_daytime
+from osa.utils.utils import wait_for_daytime, date_to_dir
 from osa.utils.logging import myLogger
 from osa.job import get_sacct_output, FORMAT_SLURM
+from osa.configs import options
+from osa.configs.config import cfg
 
 log = myLogger(logging.getLogger(__name__))
 
@@ -150,6 +152,20 @@
     return StringIO(sp.check_output(sacct_cmd).decode())
 
 
+def GainSel_flag_file() -> Path:
+    filename = cfg.get("LSTOSA", "gain_selection_check")
+    date = date_to_dir(options.date)
+    GainSel_dir = Path(cfg.get(options.tel_id, "GAIN_SELECTION_FLAG_DIR"))
+    flagfile = GainSel_dir / date / filename
+    return flagfile.resolve()
+
+
+def GainSel_finished() -> bool:
+    """Check if gain selection finished successfully."""
+    flagfile = GainSel_flag_file()
+    return flagfile.exists()
+
+
 def check_failed_jobs(date: str, output_basedir: Path = None):
     """Search for failed jobs in the log directory."""
     failed_jobs = []
@@ -165,51 +181,56 @@
             log.warning(f"Job {job} did not finish successfully")
             failed_jobs.append(job)
 
-    if not failed_jobs:
-        log.info(f"{date}: all jobs finished successfully")
-    else:
+    if failed_jobs:
         log.warning(f"{date}: some jobs did not finish successfully")
 
+    else:
+        log.info(f"{date}: all jobs finished successfully")
 
-    run_summary_dir = Path("/fefs/aswg/data/real/monitoring/RunSummary")
-    run_summary_file = run_summary_dir / f"RunSummary_{date}.ecsv"
-    summary_table = Table.read(run_summary_file)
-    runs = summary_table["run_id"]
-    missing_runs = []
-
-    r0_files = glob.glob(f"/fefs/aswg/data/real/R0/{date}/LST-1.?.Run?????.????.fits.fz")
-    r0g_files = glob.glob(f"/fefs/aswg/data/real/R0G/{date}/LST-1.?.Run?????.????.fits.fz")
-    all_r0_runs = [parse_r0_filename(i).run for i in r0_files]
-    all_r0g_runs = [parse_r0_filename(i).run for i in r0g_files]
-
-    for run in all_r0_runs:
-        if run not in runs:
-            if run not in all_r0g_runs:
-                missing_runs.append(run)
-
-    missing_runs.sort()
-    if missing_runs:
-        log.info(
-            f"Some runs are missing. Copying R0 files of runs {pd.Series(missing_runs).unique()} "
-            f"directly to /fefs/aswg/data/real/R0G/{date}"
-        )
-
-        for run in missing_runs:
-            output_dir = Path(f"/fefs/aswg/data/real/R0G/{date}/")
-            files = glob.glob(f"/fefs/aswg/data/real/R0/{date}/LST-1.?.Run{run:05d}.????.fits.fz")
-            for file in files:
-                sp.run(["cp", file, output_dir])
 
+        run_summary_dir = Path("/fefs/aswg/data/real/monitoring/RunSummary")
+        run_summary_file = run_summary_dir / f"RunSummary_{date}.ecsv"
+        summary_table = Table.read(run_summary_file)
+        runs = summary_table["run_id"]
+        missing_runs = []
+
+        r0_files = glob.glob(f"/fefs/aswg/data/real/R0/{date}/LST-1.?.Run?????.????.fits.fz")
+        r0g_files = glob.glob(f"/fefs/aswg/data/real/R0G/{date}/LST-1.?.Run?????.????.fits.fz")
+        all_r0_runs = [parse_r0_filename(i).run for i in r0_files]
+        all_r0g_runs = [parse_r0_filename(i).run for i in r0g_files]
+
+        for run in all_r0_runs:
+            if run not in runs:
+                if run not in all_r0g_runs:
+                    missing_runs.append(run)
+
+        missing_runs.sort()
+        if missing_runs:
+            log.info(
+                f"Some runs are missing. Copying R0 files of runs {pd.Series(missing_runs).unique()} "
+                f"directly to /fefs/aswg/data/real/R0G/{date}"
+            )
+
+            for run in missing_runs:
+                output_dir = Path(f"/fefs/aswg/data/real/R0G/{date}/")
+                files = glob.glob(f"/fefs/aswg/data/real/R0/{date}/LST-1.?.Run{run:05d}.????.fits.fz")
+                for file in files:
+                    sp.run(["cp", file, output_dir])
+
+        flagfile = GainSel_flag_file()
+        flagfile.touch()
 
 
 @click.command()
 @click.option("--check", is_flag=True, default=False, help="Check for failed jobs.")
-@click.argument("dates-file", type=click.Path(exists=True, path_type=Path))
-@click.argument("output-basedir", type=click.Path(path_type=Path))
+@click.option("-d", "--date", type=str)
+@click.option("-l", "--dates-file", type=click.Path(exists=True, path_type=Path))
+@click.option("-o", "--output-basedir", type=click.Path(path_type=Path))
 @click.option("-s", "--start-time", type=int, default=10, help="Time to (re)start gain selection in HH format.")
 @click.option("-e", "--end-time", type=int, default=18, help="Time to stop gain selection in HH format.")
 def main(
-    dates_file: Path = None, 
+    date: str = None,
+    dates_file: Path = None,
     output_basedir: Path = None, 
     check: bool = False, 
     start_time: int = 10, 
@@ -222,15 +243,23 @@
     """
     log.setLevel(logging.INFO)
 
-    list_of_dates = get_list_of_dates(dates_file)
-
-    if check:
-        for date in list_of_dates:
+    if date:
+        if check:
             check_failed_jobs(date, output_basedir)
-    else:
-        for date in list_of_dates:
+        else:
             apply_gain_selection(date, start_time, end_time, output_basedir)
-        log.info("Done! No more dates to process.")
+
+
+    elif dates_file:
+        list_of_dates = get_list_of_dates(dates_file)
+
+        if check:
+            for date in list_of_dates:
+                check_failed_jobs(date, output_basedir)
+        else:
+            for date in list_of_dates:
+                apply_gain_selection(date, start_time, end_time, output_basedir)
+            log.info("Done! No more dates to process.")
 
 
 if __name__ == "__main__":

src/osa/scripts/sequencer.py

@@ -7,6 +7,7 @@
 
 import logging
 import os
+import sys
 from decimal import Decimal
 
 from osa import osadb
@@ -22,12 +23,14 @@
     run_squeue,
 )
 from osa.nightsummary.extract import build_sequences
+from osa.nightsummary.nightsummary import run_summary_table
 from osa.paths import analysis_path
 from osa.report import start
 from osa.utils.cliopts import sequencer_cli_parsing
 from osa.utils.logging import myLogger
 from osa.utils.utils import is_day_closed, gettag, date_to_iso
 from osa.veto import get_closed_list, get_veto_list
+from osa.scripts.gain_selection import GainSel_finished
 
 __all__ = [
     "single_process",
@@ -90,6 +93,18 @@
     if not options.simulate:
         os.makedirs(options.log_directory, exist_ok=True)
 
+    summary_table = run_summary_table(options.date)
+    if len(summary_table) == 0:
+        log.warning("No runs found for this date. Nothing to do. Exiting.")
+        sys.exit(0)
+
+    if not options.no_gainsel and not GainSel_finished():
+        log.info(
+            f"Gain selection did not finish successfully for date {options.date}."
+            "Try again later, once gain selection has finished."
+        )
+        sys.exit()
+
     if is_day_closed():
         log.info(f"Date {date_to_iso(options.date)} is already closed for {options.tel_id}")
         return sequence_list

src/osa/scripts/tests/test_osa_scripts.py

@@ -126,9 +126,11 @@ def test_simulated_sequencer(
     run_catalog,
     r0_data,
     merged_run_summary,
+    gain_selection_flag_file,
 ):
     assert run_summary_file.exists()
     assert run_catalog.exists()
+    assert gain_selection_flag_file.exists()
 
     for r0_file in r0_data:
         assert r0_file.exists()

src/osa/utils/cliopts.py

@@ -274,6 +274,12 @@ def sequencer_argparser():
         default=False,
         help="Do not produce DL2 files (default False)",
     )
+    parser.add_argument(
+        "--no-gainsel",
+        action="store_true",
+        default=False,
+        help="Do not check if the gain selection finished correctly (default False)",
+    )
     parser.add_argument(
         "tel_id",
         choices=["ST", "LST1", "LST2", "all"],
@@ -292,6 +298,7 @@ def sequencer_cli_parsing():
     options.no_submit = opts.no_submit
     options.no_calib = opts.no_calib
     options.no_dl2 = opts.no_dl2
+    options.no_gainsel = opts.no_gainsel
 
     log.debug(f"the options are {opts}")