corems.mass_spectra.input.brukerSolarix_utils

 1"""Utility functions for Bruker data processing."""
 2
 3from pathlib import Path
 4from s3path import S3Path
 5
 6
 7def get_scan_attributes(scan_attr, imaging_info_attr) -> dict:
 8    """
 9    Get the scan attributes from the scan.xml or ImagingInfo.xml file.
10    
11    Parameters
12    ----------
13    d_directory_location : str, Path, or S3Path
14        Directory containing the XML files
15        
16    Returns
17    -------
18    dict
19        Dictionary containing the scan number as key and a tuple of retention time, TIC, 
20        and optionally maxpeak and spotname as values.
21
22
23    TODO: We need to reformat the dictionary to actually include keys and values so it is self-descriptive.
24    TODO: This will break the code, so a new version is needed. 
25    TODO: Will need to make sure theres tests which capture this change.
26    
27    """
28    from bs4 import BeautifulSoup
29    
30    scan_xml_exists = scan_attr.exists()
31    imaging_info_exists = imaging_info_attr.exists()
32
33    if scan_xml_exists:
34        try:
35            soup = BeautifulSoup(scan_attr.open(), "xml")
36            list_rt = [float(rt.text) for rt in soup.find_all("minutes")]
37            list_tic = [float(tic.text) for tic in soup.find_all("tic")]
38            list_scan = [int(scan.text) for scan in soup.find_all("count")]
39            
40            # Check if maxpeak exists (more comprehensive version)
41            # TODO: Enable this, but it could break code so a new version is needed
42            enable_maxpeak = False
43            if enable_maxpeak:
44                maxpeak_elements = soup.find_all("maxpeak")
45                if maxpeak_elements:
46                    list_maxpeak = [float(maxpeak.text) for maxpeak in maxpeak_elements]
47                    dict_scan_rt_tic = dict(zip(list_scan, zip(list_rt, list_tic, list_maxpeak)))
48                else:
49                    dict_scan_rt_tic = dict(zip(list_scan, zip(list_rt, list_tic)))
50
51            dict_scan_rt_tic = dict(zip(list_scan, zip(list_rt, list_tic)))
52
53            return dict_scan_rt_tic
54        except Exception as e:
55            raise FileNotFoundError(f"Error reading scan.xml: {e}")
56    elif imaging_info_exists:
57        try:
58            soup = BeautifulSoup(imaging_info_attr.open(), "xml")
59            list_rt = [float(rt.text) for rt in soup.find_all("minutes")]
60            list_tic = [float(tic.text) for tic in soup.find_all("tic")]
61            list_maxpeak = [float(maxpeak.text) for maxpeak in soup.find_all("maxpeak")]
62            list_scan = [int(scan.find("count").text) for scan in soup.find_all("scan")]
63            list_spotname = [
64                scan.find("spotName").text for scan in soup.find_all("scan")
65            ]
66            dict_scan_rt_tic = dict(zip(list_scan, zip(list_rt, list_tic, list_maxpeak, list_spotname)))
67            return dict_scan_rt_tic
68        except Exception as e:
69            raise FileNotFoundError(f"Error reading ImagingInfo.xml: {e}")
70    else:
71        raise FileNotFoundError(
72            "Dataset does not contain a 'scan.xml' or 'ImagingInfo.xml' file."
73        )