Merge pull request #2348 from desihub/kibo_pipe_tweaks

Minor Kibo workflow fixes

py/desispec/scripts/proc_night.py

@@ -756,6 +756,13 @@ def submit_calibrations(cal_etable, ptable, cal_override, calibjobs, int_id,
         arc_prows = []
         for arc_erow in arcs:
             if arc_erow['EXPID'] in processed_cal_expids:
+                matches = np.where([arc_erow['EXPID'] in itterprow['EXPID']
+                                    for itterprow in ptable])[0]
+                if len(matches) == 1:
+                    prow = ptable[matches[0]]
+                    log.info("Found existing arc prow in ptable, " 
+                             + f"including it for psfnight job: {list(prow)}")
+                    arc_prows.append(prow)
                 continue
             prow, int_id = make_exposure_prow(arc_erow, int_id, calibjobs)
             prow, ptable = create_submit_add_and_save(prow, ptable)
@@ -775,6 +782,13 @@ def submit_calibrations(cal_etable, ptable, cal_override, calibjobs, int_id,
         flat_prows = []
         for flat_erow in flats:
             if flat_erow['EXPID'] in processed_cal_expids:
+                matches = np.where([flat_erow['EXPID'] in itterprow['EXPID']
+                                    for itterprow in ptable])[0]
+                if len(matches) == 1:
+                    prow = ptable[matches[0]]
+                    log.info("Found existing flat prow in ptable, " 
+                             + f"including it for nightlyflat job: {list(prow)}")
+                    flat_prows.append(prow)
                 continue
 
             jobdesc = 'flat'

py/desispec/scripts/submit_prod.py

@@ -245,17 +245,22 @@ def submit_production(production_yaml, queue_threshold=4500, dry_run_level=False
     all_nights = sorted(all_nights)
     log.info(f"Processing {all_nights=}")
     for night in sorted(all_nights):
+        ## If proctable exists, assume we've already completed that night
+        if os.path.exists(findfile('proctable', night=night, readonly=True)):
+            skipped_nights.append(night)
+            log.info(f"{night=} already has a proctable, skipping.")
+            continue
+
+        ## If the queue is too full, stop submitting nights
         num_in_queue = check_queue_count(user=user, include_scron=False,
                                          dry_run_level=dry_run_level)
-        ## In Jura the largest night had 115 jobs, to be conservative say 200 by default
+        ## In Jura the largest night had 115 jobs, to be conservative we submit
+        ## up to 4500 jobs (out of a 5000 limit) by default
         if num_in_queue > queue_threshold:
             log.info(f"{num_in_queue} jobs in the queue > {queue_threshold},"
                      + " so stopping the job submissions.")
             break
-        if os.path.exists(findfile('proctable', night=night, readonly=True)):
-            skipped_nights.append(night)
-            log.info(f"{night=} already has a proctable, skipping.")
-            continue
+
         ## We don't expect exposure tables to change during code execution here
         ## but we do expect processing tables to evolve, so clear that cache
         log.info(f"Processing {night=}")

py/desispec/workflow/calibration_selection.py

@@ -110,7 +110,7 @@ def determine_calibrations_to_proc(etable, do_cte_flats=True,
                  + f"more information is known.")
         return etable[[]]
     else:
-        log.error(f"Only found {Counter(exptypes)} calibrations "
+        log.warning(f"Only found {Counter(exptypes)} calibrations "
                   + "and not acquiring new data, so this may be fatal "
                   + "if you aren't using an override file.")
 

py/desispec/workflow/proctable.py

@@ -463,10 +463,21 @@ def read_minimal_tilenight_proctab_cols(nights=None, tileids=None,
     ## Load each relevant processing table file, subselect valid tilenight's and
     ## append to the full set
     ptab_files = sorted(ptab_files)
+    ## Less intrusive logging of files we're reading in
+    if len(ptab_files) > 0:
+        dirname = os.path.dirname(ptab_files[0])
+        shortnames = [fil.replace(dirname+"/", '') for fil in ptab_files]
+    else:
+        dirname = ''
+        shortnames = []
+    log.info(f"Loading the following processing tables for tilenight processing"
+             + f" table cache from directory: {dirname}, filenames: {shortnames}")
+
     ptables = list()
     for ptab_file in ptab_files:
         ## correct way but slower and we don't need multivalue columns
-        t = load_table(tablename=ptab_file, tabletype='proctable')
+        t = load_table(tablename=ptab_file, tabletype='proctable',
+                       suppress_logging=True)
         t = _select_tilenights_from_ptab(t)
 
         ## Need to ensure that the string columns are consistent
@@ -626,10 +637,21 @@ def read_minimal_full_proctab_cols(nights=None, tileids=None,
     ## Load each relevant processing table file, subselect valid tilenight's and
     ## append to the full set
     ptab_files = sorted(ptab_files)
+    ## Less intrusive logging of files we're reading in
+    if len(ptab_files) > 0:
+        dirname = os.path.dirname(ptab_files[0])
+        shortnames = [fil.replace(dirname+"/", '') for fil in ptab_files]
+    else:
+        dirname = ''
+        shortnames = []
+    log.info(f"Loading the following processing tables for full processing "
+             + f"table cache from directory: {dirname}, filenames: {shortnames}")
+
     ptables = list()
     for ptab_file in ptab_files:
         ## correct way but slower and we don't need multivalue columns
-        t = load_table(tablename=ptab_file, tabletype='proctable')
+        t = load_table(tablename=ptab_file, tabletype='proctable',
+                       suppress_logging=True)
 
         ## Need to ensure that the string columns are consistent
         for col in ['PROCCAMWORD']: