Documentation for Roi.py

Roi

Provides implementation of Regions-of-Interest (ROI) objects that are used for real-time ROI tracking in various controller. Additionally, ROIs can also be loaded from an mzML file and converted into Chemical objects for simulation input.

FrequentistRoiAligner

FrequentistRoiAligner(
    mz_tolerance_absolute=1e-08,
    mz_tolerance_ppm=10,
    rt_tolerance=0.5,
    mz_column_pos=1,
    rt_column_pos=2,
    intensity_column_pos=3,
    min_rt_width=1e-06,
    min_mz_width=1e-06,
    n_categories=1,
)

Bases: RoiAligner

TODO: add docstring comment This class does ...

Source code in vimms/Roi.py

def __init__(
    self,
    mz_tolerance_absolute=1e-8,
    mz_tolerance_ppm=10,
    rt_tolerance=0.5,
    mz_column_pos=1,
    rt_column_pos=2,
    intensity_column_pos=3,
    min_rt_width=0.000001,
    min_mz_width=0.000001,
    n_categories=1,
):
    """
    TODO: ask Ross or Vinny to add comment

    Args:
        mz_tolerance_absolute:
        mz_tolerance_ppm:
        rt_tolerance:
        mz_column_pos:
        rt_column_pos:
        intensity_column_pos:
        min_rt_width:
        min_mz_width:
        n_categories:
    """

    self.mz_tolerance_absolute, self.mz_tolerance_ppm, self.rt_tolerance = (
        mz_tolerance_absolute,
        mz_tolerance_ppm,
        rt_tolerance,
    )
    self.mz_column_pos, self.rt_column_pos, self.intensity_column_pos = (
        mz_column_pos,
        rt_column_pos,
        intensity_column_pos,
    )
    self.min_rt_width, self.min_mz_width = min_rt_width, min_mz_width

    self.n_categories = n_categories
    self.peaksets, self.files_loaded, self.list_of_boxes = [], [], []
    self.sample_names, self.sample_types = [], []
    self.mz_weight, self.rt_weight = 75, 25
    self.peaksets2boxes, self.peaksets2fragintensities = {}, {}
    self.addition_method = None

get_boxes

get_boxes(method='mean')

Converts peaksets to generic boxes in a different way

Parameters:

Name	Type	Description	Default
method			'mean'

Returns: ???

Source code in vimms/Roi.py

def get_boxes(self, method="mean"):
    """
    Converts peaksets to generic boxes in a different way

    Args:
        method:

    Returns: ???

    """
    boxes = super().get_boxes(method)
    categories = np.unique(np.array(self.sample_types))
    enough_categories = (
        min(Counter(self.sample_types).values()) > 1 and len(categories) == self.n_categories
    )
    pvalues = self.get_p_values(enough_categories)
    for i, box in enumerate(boxes):
        box.pvalue = pvalues[i]
    return boxes

get_p_values

get_p_values(enough_catergories)

Parameters:

Name	Type	Description	Default
enough_catergories			required

Returns: ???

Source code in vimms/Roi.py

def get_p_values(self, enough_catergories):
    """

    Args:
        enough_catergories:

    Returns: ???

    """
    # need to match boxes, not base chemicals
    if enough_catergories:
        p_values = []
        # sort X
        X = np.array(self.to_matrix())
        # sort y
        categories = np.unique(np.array(self.sample_types))
        if self.n_categories == 2:  # logistic regression
            x = np.array([1 for i in self.sample_types])
            if "control" in categories:
                control_type = "control"
            else:
                control_type = categories[0]
            x[np.where(np.array(self.sample_types) == control_type)] = 0
            x = sm.add_constant(x)
            for i in range(X.shape[0]):
                y = np.log(X[i, :] + 1)
                model = sm.OLS(y, x)
                p_values.append(model.fit(disp=0).pvalues[1])
        else:  # classification
            pass
    else:
        p_values = [None for ps in self.peaksets]
    return p_values

Roi

Roi(mz, rt, intensity, id=None)

A class to store an ROI (Regions-of-interest). An ROI is a region of consecutive scans that potentially form a chromatographic peak. This is the first step in peak detection but before the region is identified to be a peak or not. This class maintains 3 lists -- mz, rt and intensity. When a new point (mz,rt,intensity) is added, it updates the list and the mean mz which is required.

Constructs a new ROI

Parameters:

Name	Type	Description	Default
mz		the initial m/z of this ROI. Can be either a single value or a list of values.	required
rt		the initial rt of this ROI. Can be either a single value or a list of values.	required
intensity		the initial intensity of this ROI. Can be either a single value or a list of values.	required
id		ID of this ROI if provided	None

Source code in vimms/Roi.py

def __init__(self, mz, rt, intensity, id=None):
    """
    Constructs a new ROI

    Args:
        mz: the initial m/z of this ROI. Can be either a single value
            or a list of values.
        rt: the initial rt  of this ROI. Can be either a single value
            or a list of values.
        intensity: the initial intensity  of this ROI. Can be either
                   a single value or a list of values.
        id: ID of this ROI if provided
    """
    self.id = id
    self.fragmentation_events = []
    self.fragmentation_intensities = []
    self.max_fragmentation_intensity = 0.0
    self.mz_list = [mz]
    self.rt_list = [rt]
    self.intensity_list = [intensity]
    self.n = len(self.mz_list)
    self.mz_sum = sum(self.mz_list)
    self.mean_mz = self.calculate_mean_mz()
    self.min_rt = self.rt_list[0]
    self.max_rt = self.rt_list[-1]
    self.length_in_seconds = self.max_rt - self.min_rt
    self.is_fragmented = False
    self.can_fragment = True
    self.last_frag_rt = None

__getitem__

__getitem__(idx)

Returns a single point in this ROI at the specified index

Parameters:

Name	Type	Description	Default
idx		the index of item to retrieve	required

Returns: a tuple of (rt, m/z, intensity) value

Source code in vimms/Roi.py

def __getitem__(self, idx):
    """
    Returns a single point in this ROI at the
    specified index

    Args:
        idx: the index of item to retrieve

    Returns: a tuple of (rt, m/z, intensity) value

    """
    return list(zip(self.rt_list, self.mz_list, self.intensity_list))[idx]

__lt__

__lt__(other)

Compares this ROI to other based on the mean m/z value. Used for sorting.

Parameters:

Name	Type	Description	Default
other		the other ROI object for comparison	required

Returns: comparison is done by mean m/z of ROIs

Source code in vimms/Roi.py

def __lt__(self, other):
    """
    Compares this ROI to other based on the mean m/z value.
    Used for sorting.

    Args:
        other: the other ROI object for comparison

    Returns: comparison is done by mean m/z of ROIs

    """
    return self.mean_mz <= other.mean_mz

__repr__

__repr__()

Returns a string representation of this ROI

Source code in vimms/Roi.py

def __repr__(self):
    """
    Returns a string representation of this ROI
    """
    return "ROI with data points=%d fragmentations=%d mz " "(%.4f-%.4f) rt (%.4f-%.4f)" % (
        self.n,
        len(self.fragmentation_events),
        self.mz_list[0],
        self.mz_list[-1],
        self.rt_list[0],
        self.rt_list[-1],
    )

add

add(mz, rt, intensity)

Adds a point to this ROI

Parameters:

Name	Type	Description	Default
mz		the m/z value to add	required
rt		the retention time value to add	required
intensity		the intensity value to add	required

Returns: None

Source code in vimms/Roi.py

def add(self, mz, rt, intensity):
    """
    Adds a point to this ROI

    Args:
        mz: the m/z value to add
        rt: the retention time value to add
        intensity: the intensity value to add

    Returns: None

    """
    self.mz_list.append(mz)
    self.rt_list.append(rt)
    self.intensity_list.append(intensity)
    self.mz_sum += mz
    self.n += 1
    self.mean_mz = self.calculate_mean_mz()
    self.min_rt = self.rt_list[0]
    self.max_rt = self.rt_list[-1]
    self.length_in_seconds = self.max_rt - self.min_rt

add_fragmentation_event

add_fragmentation_event(scan, precursor_intensity)

Stores the fragmentation events (MS2 scan) linked to this ROI

Parameters:

Name	Type	Description	Default
scan		the MS2 scan	required
precursor_intensity		the precursor intensity	required

Returns: None

Source code in vimms/Roi.py

def add_fragmentation_event(self, scan, precursor_intensity):
    """
    Stores the fragmentation events (MS2 scan) linked to this ROI

    Args:
        scan: the MS2 scan
        precursor_intensity: the precursor intensity

    Returns: None

    """
    self.fragmentation_events.append(scan)
    self.fragmentation_intensities.append(precursor_intensity)
    self.max_fragmentation_intensity = max(self.fragmentation_intensities)

estimate_apex

estimate_apex()

Returns the apex retention time

Source code in vimms/Roi.py

def estimate_apex(self):
    """
    Returns the apex retention time
    """
    return self.rt_list[np.argmax(self.intensity_list)]

fragmented

fragmented(rt)

Sets flags to indicate that this ROI can or has been fragmented

Source code in vimms/Roi.py

def fragmented(self, rt):
    """
    Sets flags to indicate that this ROI can or has been fragmented
    """
    self.is_fragmented = True
    self.can_fragment = True
    self.last_frag_rt = rt

get_autocorrelation

get_autocorrelation(lag=1)

Computes auto-correlation of this ROI intensity signal

Source code in vimms/Roi.py

def get_autocorrelation(self, lag=1):
    """
    Computes auto-correlation of this ROI intensity signal
    """
    return pd.Series(self.intensity_list).autocorr(lag=lag)

get_boxes_overlap

get_boxes_overlap(boxes, min_rt_width, min_mz_width, rt_shift=0, mz_shift=0)

TODO: ask Ross or Vinny to add comment

Parameters:

Name	Type	Description	Default
boxes		the boxes to check	required
min_rt_width		minimum RT width of the box	required
min_mz_width		minimum m/z width of the box	required
rt_shift		shift in retention time, if any	0
mz_shift		shift in m/z, if any	0

Returns: ???

Source code in vimms/Roi.py

def get_boxes_overlap(self, boxes, min_rt_width, min_mz_width, rt_shift=0, mz_shift=0):
    """
    TODO: ask Ross or Vinny to add comment

    Args:
        boxes: the boxes to check
        min_rt_width: minimum RT width of the box
        min_mz_width: minimum m/z width of the box
        rt_shift: shift in retention time, if any
        mz_shift: shift in m/z, if any

    Returns: ???

    """

    roi_box = self.to_box(min_rt_width, min_mz_width, rt_shift, mz_shift)
    # print(roi_box)
    overlaps = [roi_box.overlap_2(box) for box in boxes]
    return overlaps

get_last_datum

get_last_datum()

Returns the last (m/z, rt, intensity) point of this ROI

Source code in vimms/Roi.py

def get_last_datum(self):
    """
    Returns the last (m/z, rt, intensity) point of this ROI
    """
    return self.mz_list[-1], self.rt_list[-1], self.intensity_list[-1]

get_max_intensity

get_max_intensity()

Returns the maximum intensity value of this ROI

Source code in vimms/Roi.py

def get_max_intensity(self):
    """
    Returns the maximum intensity value of this ROI
    """
    return max(self.intensity_list)

get_min_intensity

get_min_intensity()

Returns the minimum intensity value of this ROI

Source code in vimms/Roi.py

def get_min_intensity(self):
    """
    Returns the minimum intensity value of this ROI
    """
    return min(self.intensity_list)

get_roi_overlap

get_roi_overlap(boxes, min_rt_width, min_mz_width, rt_shift=0, mz_shift=0)

TODO: ask Ross or Vinny to add comment

Parameters:

Name	Type	Description	Default
boxes		the boxes to check	required
min_rt_width		minimum RT width of the box	required
min_mz_width		minimum m/z width of the box	required
rt_shift		shift in retention time, if any	0
mz_shift		shift in m/z, if any	0

Returns: ???

Source code in vimms/Roi.py

def get_roi_overlap(self, boxes, min_rt_width, min_mz_width, rt_shift=0, mz_shift=0):
    """
    TODO: ask Ross or Vinny to add comment

    Args:
        boxes: the boxes to check
        min_rt_width: minimum RT width of the box
        min_mz_width: minimum m/z width of the box
        rt_shift: shift in retention time, if any
        mz_shift: shift in m/z, if any

    Returns: ???

    """
    roi_box = self.to_box(min_rt_width, min_mz_width, rt_shift, mz_shift)
    overlaps = [roi_box.overlap_3(box) for box in boxes]
    return overlaps

to_box

to_box(
    min_rt_width,
    min_mz_width,
    fixed_rt_dist=None,
    fixed_mz_dist=None,
    rt_shift=0,
    mz_shift=0,
)

Returns a generic box representation of this ROI

Parameters:

Name	Type	Description	Default
min_rt_width		minimum RT width of the box	required
min_mz_width		minimum m/z width of the box	required
fixed_rt_dist		if not set to None, overrides the default rt-width (in seconds)	None
fixed_mz_width		if not set to None, overrides the default mz-width (in ppm) of	required
rt_shift		shift in retention time, if any	0
mz_shift		shift in m/z, if any	0

Returns: a vimms.Box.GenericBox object.

Source code in vimms/Roi.py

def to_box(
    self,
    min_rt_width,
    min_mz_width,
    fixed_rt_dist=None,
    fixed_mz_dist=None,
    rt_shift=0,
    mz_shift=0,
):
    """
    Returns a generic box representation of this ROI

    Args:
        min_rt_width: minimum RT width of the box
        min_mz_width: minimum m/z width of the box
        fixed_rt_dist: if not set to None, overrides the default rt-width (in seconds)
        of the box, setting it to twice the parameter value. If set to None uses the
        maximum rt-range of points belonging to this RoI as the box's rt-width
        fixed_mz_width: if not set to None, overrides the default mz-width (in ppm) of
        the box, setting it twice the parameter value. If set to None uses the
        maximum mz-range of points belonging to this RoI as the box's rt-width
        rt_shift: shift in retention time, if any
        mz_shift: shift in m/z, if any

    Returns: a [vimms.Box.GenericBox][] object.

    """
    min_rt, max_rt = min(self.rt_list), max(self.rt_list)
    min_mz, max_mz = min(self.mz_list), max(self.mz_list)

    min_rt, max_rt = self._set_fixed_bounds_for_box(fixed_rt_dist, min_rt, max_rt)
    dalton_dist = self.mean_mz * fixed_mz_dist / 1e6 if fixed_mz_dist is not None else None
    min_mz, max_mz = self._set_fixed_bounds_for_box(dalton_dist, min_mz, max_mz)

    return GenericBox(
        min_rt + rt_shift,
        max_rt + rt_shift,
        min_mz + mz_shift,
        max_mz + mz_shift,
        min_xwidth=min_rt_width,
        min_ywidth=min_mz_width,
        intensity=self.max_fragmentation_intensity,
        id=self.id,
        roi=self,
    )

to_chromatogram

to_chromatogram()

Converts this ROI to a ViMMS EmpiricalChromatogram object

Source code in vimms/Roi.py

def to_chromatogram(self):
    """
    Converts this ROI to a ViMMS EmpiricalChromatogram object
    """
    if self.n == 0:
        return None
    chrom = EmpiricalChromatogram(
        np.array(self.rt_list), np.array(self.mz_list), np.array(self.intensity_list)
    )
    return chrom

RoiAligner

RoiAligner(
    mz_tolerance_absolute=1e-08,
    mz_tolerance_ppm=10,
    rt_tolerance=0.5,
    mz_column_pos=1,
    rt_column_pos=2,
    intensity_column_pos=3,
    min_rt_width=1e-06,
    min_mz_width=1e-06,
    n_categories=1,
)

A class that aligns multiple ROIs in different samples

TODO: ask Ross or Vinny to add comment

Parameters:

Name	Type	Description	Default
mz_tolerance_absolute			1e-08
mz_tolerance_ppm			10
rt_tolerance			0.5
mz_column_pos			1
rt_column_pos			2
intensity_column_pos			3
min_rt_width			1e-06
min_mz_width			1e-06
n_categories			1

Source code in vimms/Roi.py

def __init__(
    self,
    mz_tolerance_absolute=1e-8,
    mz_tolerance_ppm=10,
    rt_tolerance=0.5,
    mz_column_pos=1,
    rt_column_pos=2,
    intensity_column_pos=3,
    min_rt_width=0.000001,
    min_mz_width=0.000001,
    n_categories=1,
):
    """
    TODO: ask Ross or Vinny to add comment

    Args:
        mz_tolerance_absolute:
        mz_tolerance_ppm:
        rt_tolerance:
        mz_column_pos:
        rt_column_pos:
        intensity_column_pos:
        min_rt_width:
        min_mz_width:
        n_categories:
    """

    self.mz_tolerance_absolute, self.mz_tolerance_ppm, self.rt_tolerance = (
        mz_tolerance_absolute,
        mz_tolerance_ppm,
        rt_tolerance,
    )
    self.mz_column_pos, self.rt_column_pos, self.intensity_column_pos = (
        mz_column_pos,
        rt_column_pos,
        intensity_column_pos,
    )
    self.min_rt_width, self.min_mz_width = min_rt_width, min_mz_width

    self.n_categories = n_categories
    self.peaksets, self.files_loaded, self.list_of_boxes = [], [], []
    self.sample_names, self.sample_types = [], []
    self.mz_weight, self.rt_weight = 75, 25
    self.peaksets2boxes, self.peaksets2fragintensities = {}, {}
    self.addition_method = None

add_picked_peaks

add_picked_peaks(
    mzml_file,
    peak_file,
    sample_name,
    picking_method="mzmine",
    sample_type=None,
    half_isolation_window=0,
    allow_last_overlap=False,
    rt_shifts=None,
    mz_shifts=None,
)

TODO: ask Ross or Vinny to add comment

Parameters:

Name	Type	Description	Default
mzml_file			required
peak_file			required
sample_name			required
picking_method			'mzmine'
sample_type			None
half_isolation_window			0
allow_last_overlap			False
rt_shifts			None
mz_shifts			None

Returns: ???

Source code in vimms/Roi.py

def add_picked_peaks(
    self,
    mzml_file,
    peak_file,
    sample_name,
    picking_method="mzmine",
    sample_type=None,
    half_isolation_window=0,
    allow_last_overlap=False,
    rt_shifts=None,
    mz_shifts=None,
):
    """
    TODO: ask Ross or Vinny to add comment

    Args:
        mzml_file:
        peak_file:
        sample_name:
        picking_method:
        sample_type:
        half_isolation_window:
        allow_last_overlap:
        rt_shifts:
        mz_shifts:

    Returns: ???

    """
    self.sample_names.append(sample_name)
    self.sample_types.append(sample_type)

    these_peaks, frag_intensities = [], []
    temp_boxes = self.load_boxes(peak_file, picking_method)
    temp_boxes = update_picked_boxes(temp_boxes, rt_shifts, mz_shifts)
    self.list_of_boxes.append(temp_boxes)

    mzml = path_or_mzml(mzml_file)
    scans2boxes, boxes2scans = map_boxes_to_scans(
        mzml,
        temp_boxes,
        half_isolation_window=half_isolation_window,
        allow_last_overlap=allow_last_overlap,
    )
    precursor_intensities, scores = get_precursor_intensities(boxes2scans, temp_boxes, "max")

    for i, box in enumerate(temp_boxes):
        source_id = f"{sample_name}_{i}"
        peak_mz = box.mz
        peak_rt = box.rt_in_seconds
        these_peaks.append(Peak(peak_mz, peak_rt, box.height, sample_name, source_id))
        frag_intensities.append(precursor_intensities[i])

    self._align(these_peaks, temp_boxes, frag_intensities, sample_name)

add_sample

add_sample(rois, sample_name, sample_type=None, rt_shifts=None, mz_shifts=None)

TODO: ask Ross or Vinny to add comment

Parameters:

Name	Type	Description	Default
rois			required
sample_name			required
sample_type			None
rt_shifts			None
mz_shifts			None

Returns: ???

Source code in vimms/Roi.py

def add_sample(self, rois, sample_name, sample_type=None, rt_shifts=None, mz_shifts=None):
    """
    TODO: ask Ross or Vinny to add comment

    Args:
        rois:
        sample_name:
        sample_type:
        rt_shifts:
        mz_shifts:

    Returns: ???

    """
    self.sample_names.append(sample_name)
    self.sample_types.append(sample_type)

    these_peaks, frag_intensities, temp_boxes = [], [], []
    for i, roi in enumerate(rois):
        source_id = f"{sample_name}_{i}"
        peak_mz = roi.mean_mz
        peak_rt = roi.estimate_apex()
        peak_intensity = roi.get_max_intensity()
        these_peaks.append(Peak(peak_mz, peak_rt, peak_intensity, sample_name, source_id))
        frag_intensities.append(roi.max_fragmentation_intensity)
        rt_shift = 0 if rt_shifts is None else rt_shifts[i]
        mz_shift = 0 if mz_shifts is None else mz_shifts[i]
        temp_boxes.append(roi.to_box(self.min_rt_width, self.min_mz_width, rt_shift, mz_shift))

    # do alignment, adding the peaks and boxes, and recalculating max
    # frag intensity
    self._align(these_peaks, temp_boxes, frag_intensities, sample_name)

get_boxes

get_boxes(method='mean')

Converts peaksets to generic boxes

Parameters:

Name	Type	Description	Default
method		which method to use for the conversion	'mean'

Returns: a list of vimms.Box.GenericBox objects.

Source code in vimms/Roi.py

def get_boxes(self, method="mean"):
    """
    Converts peaksets to generic boxes

    Args:
        method: which method to use for the conversion

    Returns: a list of [vimms.Box.GenericBox][] objects.

    """
    if method == "max":
        f1, f2 = min, max
    else:
        f1 = f2 = mean

    boxes = []
    for ps in self.peaksets:
        box_list = self.peaksets2boxes[ps]
        x1 = f1(b.pt1.x for b in box_list)
        x2 = f2(b.pt2.x for b in box_list)
        y1 = f1(b.pt1.y for b in box_list)
        y2 = f2(b.pt2.y for b in box_list)
        intensity = max(self.peaksets2fragintensities[ps])
        boxes.append(
            GenericBox(
                x1,
                x2,
                y1,
                y2,
                intensity=intensity,
                min_xwidth=self.min_rt_width,
                min_ywidth=self.min_mz_width,
            )
        )
    return boxes

get_max_frag_intensities

get_max_frag_intensities()

Returns the maximum fragmentation intensities of peaksets

Source code in vimms/Roi.py

def get_max_frag_intensities(self):
    """
    Returns the maximum fragmentation intensities of peaksets
    """
    return [max(self.peaksets2fragintensities[ps]) for ps in self.peaksets]

to_matrix

to_matrix()

Converts aligned peaksets to nicely formatted intensity matrix (rows: peaksets, columns: files)

Source code in vimms/Roi.py

def to_matrix(self):
    """
    Converts aligned peaksets to nicely formatted intensity matrix
    (rows: peaksets, columns: files)
    """
    return np.array(
        [[ps.get_intensity(fname) for fname in self.files_loaded] for ps in self.peaksets],
        dtype=np.double,
    )

RoiBuilder

RoiBuilder(roi_params, smartroi_params=None)

A class to construct ROIs. This can be used in real-time to track ROIs in a controller, or for extracting ROIs from an mzML file.

Initialises an ROI Builder object.

Parameters:

Name	Type	Description	Default
roi_params		parameters for ROI building, as defined in vimms.Roi.RoiBuilderParams.	required
smartroi_params		other SmartROI parameters, as defined in vimms.Roi.SmartRoiParams.	None
grid		a grid object, if available	required

Source code in vimms/Roi.py

def __init__(self, roi_params, smartroi_params=None):
    """
    Initialises an ROI Builder object.

    Args:
        roi_params: parameters for ROI building, as defined in [vimms.Roi.RoiBuilderParams][].
        smartroi_params: other SmartROI parameters, as defined in
                         [vimms.Roi.SmartRoiParams][].
        grid: a grid object, if available
    """
    self.roi_params = roi_params
    self.smartroi_params = smartroi_params

    self.roi_type = ROI_TYPE_NORMAL
    if self.smartroi_params is not None:
        self.roi_type = ROI_TYPE_SMART
    assert self.roi_type in [ROI_TYPE_NORMAL, ROI_TYPE_SMART]

    # Create ROI
    self.live_roi = []
    self.dead_roi = []
    self.junk_roi = []

    # fragmentation to Roi dictionaries
    self.frag_roi_dicts = []  # scan_id, roi_id, precursor_intensity
    self.roi_id_counter = 0

    # count how many times an ROI is not grown
    self.skipped_roi_count = defaultdict(int)

add_scan_to_roi

add_scan_to_roi(scan)

Stores the information on which scans and frag events are associated to this ROI

Source code in vimms/Roi.py

def add_scan_to_roi(self, scan):
    """
    Stores the information on which scans and frag events are associated
    to this ROI
    """
    frag_event_ids = np.array([event["scan_id"] for event in self.frag_roi_dicts])
    which_event = np.where(frag_event_ids == scan.scan_id)[0]
    live_roi_ids = np.array([roi.id for roi in self.live_roi])
    which_roi = np.where(live_roi_ids == self.frag_roi_dicts[which_event[0]]["roi_id"])[0]
    if len(which_roi) > 0:
        self.live_roi[which_roi[0]].add_fragmentation_event(
            scan, self.frag_roi_dicts[which_event[0]]["precursor_intensity"]
        )
        del self.frag_roi_dicts[which_event[0]]
    else:
        pass  # hopefully shouldnt happen

get_good_rois

get_good_rois()

Returns all ROIs above filtering criteria

Source code in vimms/Roi.py

def get_good_rois(self):
    """
    Returns all ROIs above filtering criteria
    """
    # length check
    filtered_roi = [
        roi
        for roi in self.live_roi
        if roi.get_num_unique_scans() >= self.roi_params.min_roi_length
    ]

    # intensity check:
    # Keep only the ROIs that can be fragmented above
    # at_least_one_point_above threshold.
    all_roi = filtered_roi + self.dead_roi
    if self.roi_params.at_least_one_point_above > 0:
        keep = []
        for roi in all_roi:
            if any(it > self.roi_params.at_least_one_point_above for it in roi.intensity_list):
                keep.append(roi)
    else:
        keep = all_roi
    return keep

get_mz_intensity

get_mz_intensity(i)

Returns the (m/z, intensity, ROI ID) value of point at position i in this ROI

Parameters:

Name	Type	Description	Default
i		the index of point to return	required

Returns: a tuple of (mz, intensity, roi ID)

Source code in vimms/Roi.py

def get_mz_intensity(self, i):
    """
    Returns the (m/z, intensity, ROI ID) value of point at position i in
    this ROI

    Args:
        i: the index of point to return

    Returns: a tuple of (mz, intensity, roi ID)

    """
    mz = self.current_roi_mzs[i]
    intensity = self.current_roi_intensities[i]
    roi_id = self.current_roi_ids[i]
    return mz, intensity, roi_id

get_rois

get_rois()

Returns all ROIs

Source code in vimms/Roi.py

def get_rois(self):
    """
    Returns all ROIs
    """
    return self.live_roi + self.dead_roi

set_fragmented

set_fragmented(current_task_id, i, roi_id, rt, intensity)

Updates this ROI to indicate that it has been fragmented

Parameters:

Name	Type	Description	Default
current_task_id		the current task ID	required
i		index of fragmented ROI in the live ROI list	required
roi_id		the ID of ROI	required
rt		time of fragmentation	required
intensity		intensity at fragmentation	required

Returns: None

Source code in vimms/Roi.py

def set_fragmented(self, current_task_id, i, roi_id, rt, intensity):
    """
    Updates this ROI to indicate that it has been fragmented

    Args:
        current_task_id:  the current task ID
        i: index of fragmented ROI in the live ROI list
        roi_id: the ID of ROI
        rt: time of fragmentation
        intensity: intensity at fragmentation

    Returns: None

    """
    self.live_roi[i].fragmented(rt)

    # Add information on which scan has fragmented this ROI
    self.frag_roi_dicts.append(
        {"scan_id": current_task_id, "roi_id": roi_id, "precursor_intensity": intensity}
    )

    # need to track for intensity non-overlap
    self.live_roi[i].max_fragmentation_intensity = max(
        self.live_roi[i].max_fragmentation_intensity, intensity
    )

update_roi

update_roi(new_scan)

Updates ROI in real-time based on incoming scans

Parameters:

Name	Type	Description	Default
new_scan		a newly arriving Scan object	required

Returns: None

Source code in vimms/Roi.py

def update_roi(self, new_scan):
    """
    Updates ROI in real-time based on incoming scans

    Args:
        new_scan: a newly arriving Scan object

    Returns: None

    """
    if new_scan.ms_level == 1:

        # Sort ROIs in live_roi according to their m/z values.
        # Ensure that the live roi fragmented flags and the last RT
        # are also consistent with the sorting order.
        self.live_roi.sort()

        # Current scan retention time of the MS1 scan is the RT of all
        # points in this scan
        current_rt = new_scan.rt

        # check that there's no ROI with skip_count > self.max_skips_allowed
        skip_counts = np.array(list(self.skipped_roi_count.values()))
        assert np.sum(skip_counts > self.roi_params.max_gaps_allowed) == 0

        # The set of ROIs that are not grown yet.
        # Initially all currently live ROIs are included, and they're
        # removed once grown.
        not_grew = set(self.live_roi)

        # For every (mz, intensity) in scan ..
        for idx in range(len(new_scan.intensities)):
            current_intensity = new_scan.intensities[idx]
            current_mz = new_scan.mzs[idx]

            if current_intensity >= self.roi_params.min_roi_intensity:

                # Create a dummy ROI object to represent the current m/z
                # value. This produces either a normal ROI or smart ROI
                # object, depending on self.roi_type
                roi = self._get_roi_obj(current_mz, 0, 0, None)

                # Match dummy ROI to currently live ROIs based on mean
                # m/z values. If no match, then return None
                match_roi = self._match(roi, self.live_roi, self.roi_params.mz_tol)
                if match_roi:

                    # Got a match, so we grow this ROI
                    match_roi.add(current_mz, current_rt, current_intensity)

                    # ROI has been matched and can be removed from not_grew
                    if match_roi in not_grew:
                        not_grew.remove(match_roi)

                        # this ROI has been grown so delete from skip count if possible
                        try:
                            del self.skipped_roi_count[match_roi]
                        except KeyError:
                            pass

                else:

                    # No match, so create a new ROI and insert it in the
                    # right place in the sorted list
                    new_roi = self._get_roi_obj(
                        current_mz, current_rt, current_intensity, self.roi_id_counter
                    )
                    self.roi_id_counter += 1
                    bisect.insort_right(self.live_roi, new_roi)

        # Separate the ROIs that have not been grown into dead or junk ROIs
        # Dead ROIs are longer than self.min_roi_length but they haven't
        # been grown. Junk ROIs are too short and not grown.
        for roi in not_grew:

            # if too much gaps ...
            self.skipped_roi_count[roi] += 1
            if self.skipped_roi_count[roi] > self.roi_params.max_gaps_allowed:

                # then set the roi to either dead or junk (depending on length)
                if roi.get_num_unique_scans() >= self.roi_params.min_roi_length:
                    self.dead_roi.append(roi)
                else:
                    self.junk_roi.append(roi)

                # Remove not-grown ROI from the list of live ROIs
                pos = self.live_roi.index(roi)
                del self.live_roi[pos]

                # this ROI is either dead or junk, so delete from skip count
                # as we don't need to track it anymore
                del self.skipped_roi_count[roi]

        self.current_roi_ids = [roi.id for roi in self.live_roi]
        self.current_roi_mzs = [roi.mz_list[-1] for roi in self.live_roi]
        self.current_roi_intensities = [roi.intensity_list[-1] for roi in self.live_roi]
        self.current_roi_length = np.array(
            [roi.get_num_unique_scans() for roi in self.live_roi]
        )

RoiBuilderParams

RoiBuilderParams(
    mz_tol=10,
    min_roi_length=0,
    min_roi_intensity=0,
    at_least_one_point_above=0,
    start_rt=0,
    stop_rt=100000.0,
    max_gaps_allowed=0,
)

A parameter object that stores various settings required for ROIBuilder

Initialises an RoiBuilderParams object

Parameters:

Name	Type	Description	Default
mz_tol		m/z tolerance	10
min_roi_length		minimum ROI length	0
min_roi_intensity		minimum intensity to be included for ROI building	0
at_least_one_point_above		keep only the ROIs containing at least one point above this threshold	0
start_rt		start RT of scans to be included for ROI building	0
stop_rt		end RT of scans to be included for ROI building	100000.0
max_gaps_allowed		the number of gaps allowed in successive scans when building ROIs	0

Source code in vimms/Roi.py

def __init__(
    self,
    mz_tol=10,
    min_roi_length=0,
    min_roi_intensity=0,
    at_least_one_point_above=0,
    start_rt=0,
    stop_rt=1e5,
    max_gaps_allowed=0,
):
    """
    Initialises an RoiBuilderParams object

    Args:
        mz_tol: m/z tolerance
        min_roi_length: minimum ROI length
        min_roi_intensity: minimum intensity to be included for ROI building
        at_least_one_point_above: keep only the ROIs containing at least one point
                                  above this threshold
        start_rt: start RT of scans to be included for ROI building
        stop_rt: end RT of scans to be included for ROI building
        max_gaps_allowed: the number of gaps allowed in successive scans
                          when building ROIs
    """
    if mz_tol < 0.1:
        logger.warning(f"Is your m/z tolerance correct? mz_tol={mz_tol}")

    self.mz_tol = mz_tol
    self.min_roi_length = min_roi_length
    self.min_roi_intensity = min_roi_intensity
    self.at_least_one_point_above = at_least_one_point_above
    self.start_rt = start_rt
    self.stop_rt = stop_rt
    self.max_gaps_allowed = max_gaps_allowed

SmartRoi

SmartRoi(mz, rt, intensity, smartroi_params, id=None)

Bases: Roi

A smarter ROI class that can track the states in which it should be fragmented.

SmartROI is described further in the following paper: - Davies, Vinny, et al. "Rapid Development of Improved Data-Dependent Acquisition Strategies." Analytical chemistry 93.14 (2021): 5676-5683.

Constructs a new Smart ROI object

Parameters:

Name	Type	Description	Default
mz		the initial m/z of this ROI. Can be either a single value or a list of values.	required
rt		the initial rt of this ROI. Can be either a single value or a list of values.	required
intensity		the initial intensity of this ROI. Can be either a single value or a list of values.	required
smartroi_params		other SmartROI parameters, as defined in vimms.Roi.SmartRoiParams.	required
id		the ID of this ROI	None

Source code in vimms/Roi.py

def __init__(self, mz, rt, intensity, smartroi_params, id=None):
    """
    Constructs a new Smart ROI object

    Args:
        mz: the initial m/z of this ROI.
            Can be either a single value or a list of values.
        rt: the initial rt  of this ROI.
            Can be either a single value or a list of values.
        intensity: the initial intensity  of this ROI.
                   Can be either a single value or a list of values.
        smartroi_params: other SmartROI parameters, as defined in
                         [vimms.Roi.SmartRoiParams][].
        id: the ID of this ROI
    """
    super().__init__(mz, rt, intensity, id=id)
    self.max_at_frag = 0.0
    self.max_since_last_frag = 0.0
    self.params = smartroi_params

    if self.params.initial_length_seconds > 0:
        self.status = SmartRoi.INITIAL_WAITING
        self.can_fragment = False
    else:
        self.status = SmartRoi.CAN_FRAGMENT
        self.can_fragment = True

    self.min_frag_intensity = None

add

add(mz, rt, intensity)

Adds a point to this SmartROI

Parameters:

Name	Type	Description	Default
mz		the m/z value to add	required
rt		the retention time value to add	required
intensity		the intensity value to add	required

Returns: None

Source code in vimms/Roi.py

def add(self, mz, rt, intensity):
    """
    Adds a point to this SmartROI

    Args:
        mz: the m/z value to add
        rt: the retention time value to add
        intensity: the intensity value to add

    Returns: None

    """
    super().add(mz, rt, intensity)
    if intensity > self.max_since_last_frag:
        self.max_since_last_frag = intensity

    if self.status == SmartRoi.INITIAL_WAITING:
        if self.length_in_seconds >= self.params.initial_length_seconds:
            self.status = SmartRoi.CAN_FRAGMENT
            self.can_fragment = True

    elif self.status == SmartRoi.AFTER_FRAGMENT:
        self.set_smartroi_rules()

fragmented

fragmented(rt)

Sets this SmartROI as having been fragmented

Source code in vimms/Roi.py

def fragmented(self, rt):
    """
    Sets this SmartROI as having been fragmented
    """
    self.is_fragmented = True
    self.can_fragment = False
    self.last_frag_rt = rt
    self.fragmented_index = len(self.mz_list) - 1
    self.max_at_frag = self.intensity_list[self.fragmented_index]
    self.max_since_last_frag = self.max_at_frag
    self.status = SmartRoi.AFTER_FRAGMENT

get_can_fragment

get_can_fragment()

Returns the status of whether this SmartROI can be fragmented

Source code in vimms/Roi.py

def get_can_fragment(self):
    """
    Returns the status of whether this SmartROI can be fragmented
    """
    return self.can_fragment

get_status

get_status()

Returns the current status of this SmartROI

Source code in vimms/Roi.py

def get_status(self):
    """
    Returns the current status of this SmartROI
    """
    if self.status == 0:
        return "INITIAL_WAITING"
    elif self.status == 1:
        return "CAN_FRAGMENT"
    elif self.status == 2:
        return "AFTER_FRAGMENT"
    elif self.status == 3:
        return "POST_PEAK"

set_can_fragment

set_can_fragment(status)

Sets the status of this SmartROI

Parameters:

Name	Type	Description	Default
status		True if this SmartROI can be fragmented again, False otherwise	required

Returns: None

Source code in vimms/Roi.py

def set_can_fragment(self, status):
    """
    Sets the status of this SmartROI

    Args:
        status: True if this SmartROI can be fragmented again,
                False otherwise

    Returns: None

    """
    self.can_fragment = status

SmartRoiParams

SmartRoiParams(
    initial_length_seconds=5,
    reset_length_seconds=1000000.0,
    intensity_increase_factor=10,
    dew=15,
    drop_perc=0.1 / 100,
)

A parameter object that stores various settings required for SmartRoi

Initialises a SmartRoiParams object

Parameters:

Name	Type	Description	Default
initial_length_seconds		the initial length (in seconds) before this ROI can be fragmented	5
reset_length_seconds		the length (in seconds) before this ROI can be fragmented again (CAN_FRAGMENT)	1000000.0
intensity_increase_factor		a factor of which the intensity should increase from the minimum since last fragmentation before this ROI can be fragmented again	10
dew		dynamic exclusion window (DEW) in seconds before this ROI can be fragmented again	15
drop_perc		percentage drop in intensity since last fragmentation before this ROI can be fragmented again	0.1 / 100

Source code in vimms/Roi.py

def __init__(
    self,
    initial_length_seconds=5,
    reset_length_seconds=1e6,
    intensity_increase_factor=10,
    dew=15,
    drop_perc=0.1 / 100,
):
    """
    Initialises a SmartRoiParams object

    Args:
        initial_length_seconds: the initial length (in seconds) before
                                this ROI can be fragmented
        reset_length_seconds: the length (in seconds) before this ROI
                              can be fragmented again (CAN_FRAGMENT)
        intensity_increase_factor: a factor of which the intensity
                                   should increase from the **minimum** since
                                   last fragmentation before this ROI can be fragmented again
        dew: dynamic exclusion window (DEW) in seconds before this ROI
             can be fragmented again
        drop_perc: percentage drop in intensity since last fragmentation
                   before this ROI can be fragmented again
    """
    self.initial_length_seconds = initial_length_seconds
    self.reset_length_seconds = reset_length_seconds
    self.intensity_increase_factor = intensity_increase_factor
    self.drop_perc = drop_perc
    self.dew = dew

cosine_score

cosine_score(u, v)

Computes the cosine similarity between two vectors

Parameters:

Name	Type	Description	Default
u		first vector	required
v		second vector	required

Returns: the cosine similarity

Source code in vimms/Roi.py

def cosine_score(u, v):
    """
    Computes the cosine similarity between two vectors

    Args:
        u: first vector
        v: second vector

    Returns: the cosine similarity

    """
    numerator = (u * v).sum()
    denominator = np.sqrt((u * u).sum()) * np.sqrt((v * v).sum())
    return numerator / denominator

greedy_roi_cluster

greedy_roi_cluster(roi_list, corr_thresh=0.75, corr_type='cosine')

Performs a greedy clustering of ROIs

Parameters:

Name	Type	Description	Default
roi_list		a list of ROIs to cluster	required
corr_thresh		the threshold on correlation for clustering	0.75
corr_type		correlation type, currently unused	'cosine'

Returns: a list of lists of ROIs, each member list is a cluster.

Source code in vimms/Roi.py

def greedy_roi_cluster(roi_list, corr_thresh=0.75, corr_type="cosine"):
    """
    Performs a greedy clustering of ROIs

    Args:
        roi_list: a list of ROIs to cluster
        corr_thresh: the threshold on correlation for clustering
        corr_type: correlation type, currently unused

    Returns: a list of lists of ROIs, each member list is a cluster.

    """
    # sort in descending intensity
    roi_list_copy = [r for r in roi_list]
    roi_list_copy.sort(key=lambda x: max(x.intensity_list), reverse=True)
    roi_clusters = []
    while len(roi_list_copy) > 0:
        roi_clusters.append([roi_list_copy[0]])
        remove_idx = [0]
        if len(roi_list_copy) > 1:
            for i, r in enumerate(roi_list_copy[1:]):
                corr = roi_correlation(roi_list_copy[0], r)
                if corr > corr_thresh:
                    roi_clusters[-1].append(r)
                    remove_idx.append(i + 1)
        remove_idx.sort(reverse=True)
        for r in remove_idx:
            del roi_list_copy[r]

    return roi_clusters

make_roi

make_roi(input_file, roi_params)

Make ROIs from an input file

Parameters:

Name	Type	Description	Default
input_file		input mzML file	required
roi_params		a RoiBuilderParams object	required

the list of good ROI objects that have been filtered according to

Type	Description
	certain criteria.

Source code in vimms/Roi.py

def make_roi(input_file, roi_params):
    """
    Make ROIs from an input file

    Args:
        input_file: input mzML file
        roi_params: a RoiBuilderParams object

    Returns: the list of good ROI objects that have been filtered according to
             certain criteria.

    """

    from vimms.MassSpec import Scan  # import in fn. body to avoid circular import

    run = pymzml.run.Reader(
        input_file,
        MS1_Precision=5e-6,
        extraAccessions=[("MS:1000016", ["value", "unitName"])],
        obo_version="4.0.1",
    )

    scan_id = 0
    roi_builder = RoiBuilder(roi_params)
    for i, spectrum in enumerate(run):
        ms_level = 1
        if spectrum["ms level"] == ms_level:
            current_ms1_scan_rt, units = spectrum.scan_time

            # check that ms1 scan (in seconds) is within bound
            if units == "minute":
                current_ms1_scan_rt *= 60.0
            if current_ms1_scan_rt < roi_params.start_rt:
                continue
            if current_ms1_scan_rt > roi_params.stop_rt:
                break

            # get the raw peak data from spectrum
            mzs, intensities = spectrum_to_arrays(spectrum)

            # update the ROI construction based on the new scan
            scan = Scan(scan_id, mzs, intensities, ms_level, current_ms1_scan_rt)
            roi_builder.update_roi(scan)
            scan_id += 1

    good_roi = roi_builder.get_good_rois()
    return good_roi

plot_roi

plot_roi(roi, statuses=None, log=False)

Plots an ROI

Parameters:

Name	Type	Description	Default
roi		the ROI to plot	required
statuses		flags for coloring	None
log		whether to log the intensity (defaults False)	False

Returns: None

Source code in vimms/Roi.py

def plot_roi(roi, statuses=None, log=False):
    """
    Plots an ROI

    Args:
        roi: the ROI to plot
        statuses: flags for coloring
        log: whether to log the intensity (defaults False)

    Returns: None

    """
    if log:
        intensities = np.log(roi.intensity_list)
        plt.ylabel("Log Intensity")
    else:
        intensities = roi.intensity_list
        plt.ylabel("Intensity")
    if statuses is not None:
        colours = []
        for s in statuses:
            if s == "Noise":
                colours.append("red")
            elif s == "Increase":
                colours.append("blue")
            elif s == "Decrease":
                colours.append("yellow")
            else:
                colours.append("green")
        plt.scatter(roi.rt_list, intensities, color=colours)
    else:
        plt.scatter(roi.rt_list, intensities)
    plt.xlabel("RT")
    plt.show()

roi_correlation

roi_correlation(roi1, roi2, min_rt_point_overlap=5, method='pearson')

Computes the correlation between two ROI objects

Parameters:

Name	Type	Description	Default
roi1		first ROI	required
roi2		second ROI	required
min_rt_point_overlap		minimum points that overlap in RT, currently unused	5
method		if 'pearson' then Peason's correlation is used, otherwise the cosine score is used	'pearson'

Returns: the correlation between the two ROIs.

Source code in vimms/Roi.py

def roi_correlation(roi1, roi2, min_rt_point_overlap=5, method="pearson"):
    """
    Computes the correlation between two ROI objects

    Args:
        roi1: first ROI
        roi2: second ROI
        min_rt_point_overlap: minimum points that overlap in RT,
                              currently unused
        method: if 'pearson' then Peason's correlation is
                used, otherwise the cosine score is used

    Returns: the correlation between the two ROIs.

    """
    # flip around so that roi1 starts earlier (or equal)
    if roi2.rt_list[0] < roi1.rt_list[0]:
        temp = roi2
        roi2 = roi1
        roi1 = temp

    # check that they meet the min_rt_point overlap
    if roi1.rt_list[-1] < roi2.rt_list[0]:
        # no overlap at all
        return 0.0

    # find the position of the first element in roi2 in roi1
    pos = roi1.rt_list.index(roi2.rt_list[0])

    # print roi1.rt_list
    # print roi2.rt_list
    # print pos

    total_length = max([len(roi1.rt_list), len(roi2.rt_list) + pos])
    # print total_length

    r1 = np.zeros((total_length), np.double)
    r2 = np.zeros_like(r1)

    r1[: len(roi1.rt_list)] = roi1.intensity_list
    r2[pos: pos + len(roi2.rt_list)] = roi2.intensity_list

    if method == "pearson":
        r, _ = pearsonr(r1, r2)
    else:
        r = cosine_score(r1, r2)

    return r

spectrum_to_arrays

spectrum_to_arrays(spectrum)

Converts pymzml spectrum to parallel arrays

Parameters:

Name	Type	Description	Default
spectrum		a pymzml spectrum object	required

a tuple (mzs, intensities) where mzs and intensities are numpy

Type	Description
	arrays of m/z and intensity values

Source code in vimms/Roi.py

def spectrum_to_arrays(spectrum):
    """
    Converts pymzml spectrum to parallel arrays

    Args:
        spectrum: a pymzml spectrum object

    Returns: a tuple (mzs, intensities) where mzs and intensities are numpy
             arrays of m/z and intensity values

    """
    mzs = []
    intensities = []
    for mz, intensity in spectrum.peaks("raw"):
        mzs.append(mz)
        intensities.append(intensity)
    mzs = np.array(mzs)
    intensities = np.array(intensities)
    return mzs, intensities

update_picked_boxes

update_picked_boxes(picked_boxes, rt_shifts, mz_shifts)

Updates picked boxes ??

Parameters:

Name	Type	Description	Default
picked_boxes			required
rt_shifts			required
mz_shifts			required

Returns: ???

Source code in vimms/Roi.py

def update_picked_boxes(picked_boxes, rt_shifts, mz_shifts):
    """
    Updates picked boxes ??

    Args:
        picked_boxes:
        rt_shifts:
        mz_shifts:

    Returns: ???

    """
    if rt_shifts is None and mz_shifts is None:
        return picked_boxes

    new_boxes = copy.deepcopy(picked_boxes)
    for box, sec_shift, mz_shift in zip(new_boxes, rt_shifts, mz_shifts):
        if rt_shifts is not None:
            sec_shift = float(sec_shift)
            min_shift = sec_shift / 60.0

            box.rt += min_shift
            box.rt_in_minutes += min_shift
            box.rt_in_seconds += sec_shift
            box.rt_range = [r + min_shift for r in box.rt_range]
            box.rt_range_in_seconds = [r + sec_shift for r in box.rt_range_in_seconds]

        if mz_shifts is not None:
            mz_shift = float(mz_shift)
            box.mz += mz_shift
            box.mz_range = [r + mz_shift for r in box.mz_range]