Add fix integration window for the noise runs

straxen/config/protocols.py

@@ -222,7 +222,12 @@ def objects_to_array(objects: list):
 
 @URLConfig.register("run_doc")
 def read_rundoc(path, run_id=None, default=None):
-    """Read a path from the rundoc."""
+    """Read a path from the json rundoc metadata.
+
+    :param path: keys of json rundoc metada.
+        e.g. `comments` for reading comments for a specific run
+
+    """
     if run_id is None:
         raise ValueError("rundoc protocol: missing run_id.")
     runs = xent_collection()

straxen/plugins/led_cal/led_calibration.py

@@ -38,14 +38,24 @@ class LEDCalibration(strax.Plugin):
           from the signal one.
     """
 
-    __version__ = "0.3.0"
+    __version__ = "0.3.1"
 
     depends_on = "raw_records"
     data_kind = "led_cal"
     compressor = "zstd"
     parallel = "process"
     rechunk_on_save = False
 
+    run_doc = straxen.URLConfig(
+        default="run_doc://comments?run_id=plugin.run_id",
+        infer_type=False,
+        help=(
+            "Comments dictionary from the run metadata. "
+            "It is used for discriminate between LED on and LED "
+            "off runs."
+        ),
+    )
+
     led_cal_record_length = straxen.URLConfig(
         default=160, infer_type=False, help="Length (samples) of one record without 0 padding."
     )
@@ -65,6 +75,16 @@ class LEDCalibration(strax.Plugin):
         ),
     )
 
+    fixed_position = straxen.URLConfig(
+        default=88,
+        infer_type=False,
+        help=(
+            "Fixed ADC sample upon which the integration window is defined. "
+            "This is used as default when no hits or less than a certain amount "
+            "are identified."
+        ),
+    )
+
     led_hit_extension = straxen.URLConfig(
         default=(-8, 32),
         infer_type=False,
@@ -133,6 +153,9 @@ def compute(self, raw_records):
         This is used for the different ligh levels. As default value all the PMTs are considered.
 
         """
+
+        self.is_led_on = is_the_led_on(self.run_doc)
+
         mask = np.where(np.in1d(raw_records["channel"], self.channel_list))[0]
         raw_records_active_channels = raw_records[mask]
         records = get_records(
@@ -146,6 +169,8 @@ def compute(self, raw_records):
         led_windows, triggered = get_led_windows(
             records,
             self.minimum_led_position,
+            self.fixed_position,
+            self.is_led_on,
             self.led_hit_extension,
             self.led_cal_hit_min_height_over_noise,
             self.led_cal_record_length,
@@ -166,6 +191,34 @@ def compute(self, raw_records):
         return temp
 
 
+def is_the_led_on(run_doc):
+    """Utilizing the run database metadata to determine whether the run ID corresponds to LED on or
+    LED off runs.
+
+    The LED off, or noise runs, are identified by having 'Gain_calibration_step3' or
+    'SPE_calibration_step0' in the comment.
+
+    """
+    # Check if run_doc is a list with a dictionary
+    if isinstance(run_doc, list) and len(run_doc) == 1 and isinstance(run_doc[0], dict):
+        # Extract the dictionary
+        doc = run_doc[0]
+
+        # Check if the required keys are present
+        required_keys = {"user", "date", "comment"}
+        if all(key in doc for key in required_keys):
+            # Check if 'comment' contains the specified strings
+            comment = doc["comment"]
+            if "Gain_calibration_step3" in comment or "SPE_calibration_step0" in comment:
+                return False
+            else:
+                return True
+        else:
+            raise ValueError("The dictionary does not contain the required keys.")
+    else:
+        raise ValueError("The input is not a list with a single dictionary.")
+
+
 def get_records(raw_records, baseline_window, led_cal_record_length):
     """Determine baseline as the average of the first baseline_samples of each pulse.
 
@@ -217,6 +270,8 @@ def get_records(raw_records, baseline_window, led_cal_record_length):
 def get_led_windows(
     records,
     minimum_led_position,
+    fixed_position,
+    is_led_on,
     led_hit_extension,
     hit_min_height_over_noise,
     record_length,
@@ -228,6 +283,11 @@ def get_led_windows(
     :param records: Array of the records to search for LED hits.
     :param minimum_led_position: The minimum simple index of the LED hits. Hits before this sample
         are ignored.
+    : is_led_on: Fetch from the run database. It is used for discriminate between LED on
+        and LED off runs.
+    :param fixed_position: Fixed ADC sample upon which the integration window is defined. Used
+        if no hits are identified :minimum_hits_requirement: Minimum hits requirement for the
+        dynamic window. Used to discriminate between led on and led off runs
     :param led_hit_extension: The integration window around the first hit found to use. A tuple of
         form (samples_before, samples_after) the first LED hit.
     :param hit_min_amplitude: Minimum amplitude of the signal to be considered a hit.
@@ -260,10 +320,14 @@ def get_led_windows(
     ):
         hits.sort(order=["record_i", "time"])
 
-    if len(hits) == 0:  # This really should not be the case. But in case it is:
-        default_hit_position = minimum_led_position
+    # If there are not hits in the chunk or if
+    # the run is a nosie run, with LED off,
+    # the integration window is defined beased on a
+    # hard-coded ADC sample
+    if (not is_led_on) or (len(hits) == 0):
+        default_hit_position = fixed_position
     else:
-        default_hit_position, _ = sps.mode(hits["left"])
+        default_hit_position, mode_count = sps.mode(hits["left"])
 
         if isinstance(default_hit_position, np.ndarray):
             default_hit_position = default_hit_position[0]