Refactor: Centralized HDF5 data loading

classification/evaluate.py

@@ -14,7 +14,7 @@
 from pathlib import Path
 from typing import Iterator, Tuple, Dict, Any
 from utils.config import DATA_DIR
-from .data_loader import QuenchData, load_quench_events
+from utils.h5_load_data import QuenchData, load_quench_events
 from .logic import classify
 
 
@@ -25,7 +25,7 @@ def run_classification(
     classification_results = {}
 
     for event_id, filename, event_data in events_iterator:
-        label = classify(event_data)
+        label = classify(event_data)  # type: ignore
         classification_results[event_id] = (label, filename)
 
     return classification_results
@@ -82,7 +82,7 @@ def main() -> None:
     events_iterator = load_quench_events(target_files)
     prediction_results = run_classification(events_iterator)
     labeled_file_path = (
-        Path(DATA_DIR) / "quench_data_L0_labeled.h5"
+        Path(DATA_DIR) / "quench_data_L0.h5"
     )  # File path of labeled data to be used for comparison
     compare_classification(prediction_results, labeled_file_path)
 

classification/logic.py

@@ -99,43 +99,40 @@ def classify(event_data: QuenchData) -> QuenchStatus:
 
     return QuenchStatus.other
 
-def compute_suggestion(signal_data, frequency, saved_q_loaded):
-    """Compute the classification suggestion using the classify system written by Norah"""
 
-    # If there is no fault_waveform, we are unable to classify 
+# Compute the classification suggestion using the classify system
+def compute_suggestion(signal_data, frequency, saved_q_loaded):
     if "fault_waveform" not in signal_data:
         return None
 
-    x_fault, y_fault = signal_data["fault_waveform"]    # Split the fault_waveform (time, amplitude) tuple into two separate arrays
+    x_fault, y_fault = signal_data["fault_waveform"]
 
-    # If the forward_power is missing, we can't run the classifier 
     if "forward_power" not in signal_data:
         return None
-    x_fwd, y_fwd = signal_data["forward_power"]     # Split the forward_power (time, amplitude) tuple into two separate arrays
+    x_fwd, y_fwd = signal_data["forward_power"]
 
-    # reverse_power may or may not exist, if missing assign none to the time and amplitude 
     x_rev, y_rev = signal_data.get("reverse_power", (None, None))
 
     try:
-        # Build the QuenchData object 
-        # Convert every array into float for safer math calculations 
         quench_event = QuenchData(
-            fault_time=np.asarray(x_fault, dtype=float),    
-            fault_waveform=np.asarray(y_fault, dtype=float), 
+            fault_time=np.asarray(x_fault, dtype=float),
+            fault_waveform=np.asarray(y_fault, dtype=float),
             forward_power=np.asarray(y_fwd, dtype=float),
             forward_time=np.asarray(x_fwd, dtype=float),
-            reverse_power=np.asarray(y_rev, dtype=float) if y_rev is not None else np.array([]), # Reverse power amplitude if available, else an empty array
-            reverse_time=np.asarray(x_rev, dtype=float) if x_rev is not None else np.array([]), # Reverse time if available, else an empty array
+            reverse_power=np.asarray(y_rev, dtype=float)
+            if y_rev is not None
+            else np.array([]),
+            reverse_time=np.asarray(x_rev, dtype=float)
+            if x_rev is not None
+            else np.array([]),
         )
-       
+
         if frequency is not None:
-            # Convert frequency into numpy no matter what type of data it came in 
             quench_event.frequency = float(np.asarray(frequency).flat[0])
         if saved_q_loaded is not None:
-            # Convert saved_q_loaded into numpy no matter what type of data it came in 
             quench_event.saved_q_loaded = float(np.asarray(saved_q_loaded).flat[0])
 
-        return classify(quench_event)  # Calls classify function which returns a QuenchStatus [real, false, other or cavoty off]
-    except Exception as e :
+        return classify(quench_event)
+    except Exception as e:
         st.error(f"Classification suggestion has failed: {e}")
-        return None
+        return None

generate_plots.py

@@ -3,7 +3,7 @@
 """
 
 from plotter.plot_data import plot_data
-from utils.load_data import load_data
+from utils.h5_load_data import build_plotter_bundle
 
 PLOTS: dict[str, bool] = {
     "box_real_slice_cm": True,
@@ -25,7 +25,7 @@
 
 
 def main():
-    config = load_data()
+    config = build_plotter_bundle()
     plot_data(PLOTS, config)
 
 

interface/add_multipacting_flags.py

@@ -1,61 +1,64 @@
 import os
 import sys
+
 sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 import h5py
-import pandas as pd
-import numpy as np
-from utils.srf_waveforms import parse_h5_event_path
+from utils.label_helpers import parse_h5_event_path
 from utils.quench_data_summary import load_csv
 from utils.h5_reader import find_event_groups
 from utils.label_helpers import norm_cm, norm_cav
 from pathlib import Path
 
 
-
-def add_multipacting_flags(file_path, multipacting_file, flag_attr='Multipacting'):
+def add_multipacting_flags(file_path, multipacting_file, flag_attr="Multipacting"):
     """
     Add a multipacting flag to each events in the h5 file
-    - The function matches the event name in the h5 file with the data in the csv file 
-    - Each event in the h5 file gets a boolean attribute so either true or false 
+    - The function matches the event name in the h5 file with the data in the csv file
+    - Each event in the h5 file gets a boolean attribute so either true or false
     - Match only after ignoring the time (HHMMSS) in the h5 file since the events in the csv file are saved with no time records
     """
-    dataframe = load_csv(multipacting_file) # Load the csv file into a dataframe
+    dataframe = load_csv(multipacting_file)  # Load the csv file into a dataframe
     multipacting_keys = build_multipacting_keys(dataframe)
 
-    matched = 0     # flagged multipacting events 
-    total = 0       # valid processed events 
+    matched = 0  # flagged multipacting events
+    total = 0  # valid processed events
 
-    # read the h5 file 
-    with h5py.File(file_path, 'a') as f :
+    # read the h5 file
+    with h5py.File(file_path, "a") as f:
         event_paths = find_event_groups(f)
 
-        # Loop over each path 
+        # Loop over each path
         for path in event_paths:
-           is_valid, is_multipacting = flag_event(f, path, multipacting_keys, flag_attr)
-           if is_valid:
+            is_valid, is_multipacting = flag_event(
+                f, path, multipacting_keys, flag_attr
+            )
+            if is_valid:
                 total += 1
                 if is_multipacting:
-                    matched += 1        # Increment the multipacting events counter 
+                    matched += 1  # Increment the multipacting events counter
 
-    print(f"Flagged {matched} multipacting events out of {total} valid events in the h5 file.")
+    print(
+        f"Flagged {matched} multipacting events out of {total} valid events in the h5 file."
+    )
     return matched, total
 
 
 # build a set of keys to identify each multipacting event
 def build_multipacting_keys(dataframe):
     multipacting_keys = set()
     for _, row in dataframe.iterrows():
-        key =(
-            norm_cm(row['cm']),     # cryomodule as a string
-            norm_cav(row['cav']),    # cavity as a string 
+        key = (
+            norm_cm(row["cm"]),  # cryomodule as a string
+            norm_cav(row["cav"]),  # cavity as a string
             f"{int(row['year']):04d}",  # year : 4 digits, e.g. 2025
-            f"{int(row['month']):02d}", # month : 2 didgits, e.g. 07
-            f"{int(row['day']):02d}",   # day : 2 digits, e.g. 19
+            f"{int(row['month']):02d}",  # month : 2 didgits, e.g. 07
+            f"{int(row['day']):02d}",  # day : 2 digits, e.g. 19
         )
         multipacting_keys.add(key)
 
     return multipacting_keys
 
+
 def flag_event(f, path, multipacting_keys, flag_attr):
     parsed = parse_h5_event_path(path)
     if not parsed:
@@ -66,19 +69,23 @@ def flag_event(f, path, multipacting_keys, flag_attr):
 
     event_key = (norm_cm(cm), norm_cav(cav), year, month, day)
 
-    is_multipacting = event_key in multipacting_keys        # Check if the event is in the multipcating set 
-    f[path].attrs[flag_attr] = bool(is_multipacting)        # Write the boolean result back to the h5 file
+    is_multipacting = (
+        event_key in multipacting_keys
+    )  # Check if the event is in the multipcating set
+    f[path].attrs[flag_attr] = bool(
+        is_multipacting
+    )  # Write the boolean result back to the h5 file
 
     return True, is_multipacting
 
 
-if __name__ == '__main__':
-
-    ROOT = Path(__file__).resolve().parent.parent  
-    csv_file = ROOT / "config" / "all_mp_dates.csv" # The mp file 
+if __name__ == "__main__":
+    ROOT = Path(__file__).resolve().parent.parent
+    csv_file = ROOT / "config" / "all_mp_dates.csv"  # The mp file
 
     # Change this field:
-    h5_file_path = '/Users/username/directory/data/quench_data_L0.h5' # Your local h5 file path
+    h5_file_path = (
+        "/Users/username/directory/data/quench_data_L0.h5"  # Your local h5 file path
+    )
 
     add_multipacting_flags(file_path=h5_file_path, multipacting_file=str(csv_file))
-