Improve mzML reader parsing for filter strings and missing fields

alpharaw/mzml.py

@@ -8,6 +8,10 @@
     ms_reader_provider,
 )
 
+SAFE_PRECURSOR_MZ = -1.0
+SAFE_ISOLATION_MZ = -1.0
+DEFAULT_ISOLATION_OFFSET = 1.5
+
 
 class MzMLReader(MSData_Base):
     """
@@ -112,10 +116,20 @@ def _import(
         peak_indices = np.empty(len(spec_indices) + 1, np.int64)
         peak_indices[0] = 0
         peak_indices[1:] = np.cumsum(_peak_indices)
+        peak_mz = (
+            np.concatenate(mzs_list)
+            if mzs_list
+            else np.empty(0, dtype=PEAK_MZ_DTYPE)
+        )
+        peak_intensity = (
+            np.concatenate(intens_list)
+            if intens_list
+            else np.empty(0, dtype=PEAK_INTENSITY_DTYPE)
+        )
         ret_dict = {
             "peak_indices": peak_indices,
-            "peak_mz": np.concatenate(mzs_list),
-            "peak_intensity": np.concatenate(intens_list),
+            "peak_mz": peak_mz,
+            "peak_intensity": peak_intensity,
             "rt": np.array(rt_list),
             "precursor_mz": np.array(prec_mz_list),
             "precursor_charge": np.array(charge_list, dtype=np.int8),
@@ -130,6 +144,106 @@ def _import(
         return ret_dict
 
 
+def _parse_nce_from_filter_string(filter_string) -> float:
+    """Parse NCE from Thermo-like filter strings."""
+    if not isinstance(filter_string, str) or not filter_string:
+        return np.nan
+
+    try:
+        if "@hcd" in filter_string:
+            return float(filter_string.split("@hcd")[1].split(" ")[0])
+
+        if "@cid" in filter_string:
+            return float(filter_string.split("@cid")[1].split(" ")[0])
+
+        return np.nan
+    except (ValueError, IndexError):
+        return np.nan
+
+
+def _parse_charge_state(selected_ion: dict | None) -> int:
+    if selected_ion is None:
+        return 0
+
+    charge_state = selected_ion.get("charge state")
+
+    if charge_state is None:
+        return 0
+
+    try:
+        return int(charge_state)
+    except (TypeError, ValueError):
+        return 0
+
+
+def _get_first_precursor(item_dict: dict) -> dict | None:
+    precursor_list = item_dict.get("precursorList")
+    if not isinstance(precursor_list, dict):
+        return None
+
+    precursors = precursor_list.get("precursor")
+    if not isinstance(precursors, list) or not precursors:
+        return None
+
+    precursor = precursors[0]
+    if not isinstance(precursor, dict):
+        return None
+
+    return precursor
+
+
+def _get_first_selected_ion(precursor: dict | None) -> dict | None:
+    if precursor is None:
+        return None
+
+    selected_ion_list = precursor.get("selectedIonList")
+    if not isinstance(selected_ion_list, dict):
+        return None
+
+    selected_ions = selected_ion_list.get("selectedIon")
+    if not isinstance(selected_ions, list) or not selected_ions:
+        return None
+
+    selected_ion = selected_ions[0]
+    if not isinstance(selected_ion, dict):
+        return None
+
+    return selected_ion
+
+
+def _get_isolation_window(precursor: dict | None) -> dict | None:
+    if precursor is None:
+        return None
+
+    isolation_window = precursor.get("isolationWindow")
+    if not isinstance(isolation_window, dict):
+        return None
+
+    return isolation_window
+
+
+def _get_peak_array(item_dict: dict, key: str) -> np.ndarray:
+    peak_array = item_dict.get(key)
+    if peak_array is None:
+        raise KeyError(f"Missing '{key}' in mzML scan payload.")
+    return np.asarray(peak_array)
+
+
+def _get_first_scan_entry(item_dict: dict) -> dict:
+    scan_list = item_dict.get("scanList")
+    if not isinstance(scan_list, dict):
+        raise KeyError("Missing 'scanList' in mzML scan payload.")
+
+    scans = scan_list.get("scan")
+    if not isinstance(scans, list) or not scans:
+        raise KeyError("Missing 'scan' entries in mzML scan payload.")
+
+    scan_entry = scans[0]
+    if not isinstance(scan_entry, dict):
+        raise TypeError("Expected first 'scan' entry to be a dict.")
+    return scan_entry
+
+
 def parse_mzml_entry(item_dict: dict) -> tuple:
     """
     Parse mzml entries from pyteomics extracted items.
@@ -144,71 +258,48 @@ def parse_mzml_entry(item_dict: dict) -> tuple:
     tuple
         items in tuple format.
     """
-    rt = float(item_dict.get("scanList").get("scan")[0].get("scan start time"))
-    masses = item_dict.get("m/z array")
-    intensities = item_dict.get("intensity array")
+    scan_entry = _get_first_scan_entry(item_dict)
+    rt = float(scan_entry.get("scan start time"))
+    masses = _get_peak_array(item_dict, "m/z array")
+    intensities = _get_peak_array(item_dict, "intensity array")
     ms_level = item_dict.get("ms level")
-    prec_mz = -1.0
-    isolation_lower_mz = -1.0
-    isolation_upper_mz = -1.0
+    prec_mz = SAFE_PRECURSOR_MZ
+    isolation_lower_mz = SAFE_ISOLATION_MZ
+    isolation_upper_mz = SAFE_ISOLATION_MZ
     charge = 0
     nce = 0.0
     if ms_level == 2:
-        try:
-            charge = int(
-                item_dict.get("precursorList")
-                .get("precursor")[0]
-                .get("selectedIonList")
-                .get("selectedIon")[0]
-                .get("charge state")
-            )
-        except TypeError:
-            charge = 0
-        try:
-            charge = int(
-                item_dict.get("precursorList")
-                .get("precursor")[0]
-                .get("selectedIonList")
-                .get("selectedIon")[0]
-                .get("charge state")
-            )
-        except TypeError:
-            charge = 0
-
-        prec_mz = (
-            item_dict.get("precursorList")
-            .get("precursor")[0]
-            .get("selectedIonList")
-            .get("selectedIon")[0]
-            .get("selected ion m/z")
+        precursor = _get_first_precursor(item_dict)
+        selected_ion = _get_first_selected_ion(precursor)
+        charge = _parse_charge_state(selected_ion)
+
+        precursor_mz_value = (
+            None if selected_ion is None else selected_ion.get("selected ion m/z")
         )
         try:
-            iso_window = (
-                item_dict.get("precursorList")
-                .get("precursor")[0]
-                .get("isolationWindow")
+            prec_mz = float(precursor_mz_value)
+        except (TypeError, ValueError):
+            prec_mz = SAFE_PRECURSOR_MZ
+
+        if prec_mz != SAFE_PRECURSOR_MZ:
+            iso_window = _get_isolation_window(precursor)
+            iso_lower = None if iso_window is None else iso_window.get(
+                "isolation window lower offset"
             )
-            iso_lower = float(iso_window.get("isolation window lower offset"))
-            iso_upper = float(iso_window.get("isolation window upper offset"))
-            isolation_upper_mz = prec_mz + iso_upper
-            isolation_lower_mz = prec_mz - iso_lower
-        except TypeError:
-            isolation_upper_mz = prec_mz + 1.5
-            isolation_lower_mz = prec_mz - 1.5
-
-        nce = np.nan
-        try:
-            filter_string = (
-                item_dict.get("scanList").get("scan")[0].get["filter string"]
+            iso_upper = None if iso_window is None else iso_window.get(
+                "isolation window upper offset"
             )
-            if "@hcd" in filter_string:
-                nce = float(filter_string.split("@hcd")[1].split(" ")[0])
-            elif "@cid" in filter_string:
-                nce = float(filter_string.split("@cid")[1].split(" ")[0])
-            else:
-                nce = np.nan
-        except Exception:
-            nce = np.nan
+            try:
+                iso_lower = float(iso_lower)
+                iso_upper = float(iso_upper)
+                isolation_upper_mz = prec_mz + iso_upper
+                isolation_lower_mz = prec_mz - iso_lower
+            except (TypeError, ValueError):
+                isolation_upper_mz = prec_mz + DEFAULT_ISOLATION_OFFSET
+                isolation_lower_mz = prec_mz - DEFAULT_ISOLATION_OFFSET
+
+        filter_string = scan_entry.get("filter string")
+        nce = _parse_nce_from_filter_string(filter_string)
     return (
         rt,
         prec_mz,