Documentation for Common.py

def add_log_file(log_path, level):
    """
    Add path to log file
    Args:
        log_path: filename to output the logging to
        level: the log level

    Returns: None

    """
    logger.add(log_path, level=level)

def chromatogramDensityNormalisation(rts, intensities):
    """
    Definition to standardise the area under a chromatogram to 1.

    Args:
        rts: the RT values of this chromatogram
        intensities: the intensity values of this chromatogram

    Returns: updated intensities that have been standardised so the area is 1.

    """
    assert len(rts) == len(intensities)
    area = 0.0
    for rt_index in range(len(rts) - 1):
        area += ((intensities[rt_index] + intensities[rt_index + 1]) / 2) / (
            rts[rt_index + 1] - rts[rt_index]
        )
    new_intensities = [x * (1 / area) for x in intensities]
    return new_intensities

def create_if_not_exist(out_dir):
    """
    Creates a directory if it doesn't already exist
    Args:
        out_dir: the directory to create, if it doesn't exist

    Returns: None.

    """
    if not pathlib.Path(out_dir).exists():
        logger.info("Created %s" % out_dir)
        pathlib.Path(out_dir).mkdir(parents=True, exist_ok=True)

def download_file(url, out_file=None):
    """
    Download a file from the given URL
    Args:
        url: URL to download
        out_file: filename of output file to save to

    Returns: filename of the out_file

    """
    r = requests.get(url, stream=True)
    total_size = int(r.headers.get("content-length", 0))
    block_size = 1024
    current_size = 0

    if out_file is None:
        out_file = url.rsplit("/", 1)[-1]  # get the last part in url
    logger.info("Downloading %s" % out_file)

    with open(out_file, "wb") as f:
        for data in tqdm(
            r.iter_content(block_size),
            total=math.ceil(total_size // block_size),
            unit="KB",
            unit_scale=True,
        ):
            current_size += len(data)
            f.write(data)
    assert current_size == total_size
    return out_file

def extract_zip_file(in_file, delete=True):
    """
    Extract a zip file
    Args:
        in_file: the input zip file
        delete: whether to delete the input zip file after extracting

    Returns: None

    """
    logger.info("Extracting %s" % in_file)
    with zipfile.ZipFile(file=in_file) as zip_file:
        for file in tqdm(iterable=zip_file.namelist(), total=len(zip_file.namelist())):
            zip_file.extract(member=file)

    if delete:
        logger.info("Deleting %s" % in_file)
        os.remove(in_file)

def find_nearest_index_in_array(array, value):
    """
    Finds index in array where the value is the nearest

    Args:
        array: the array to check
        value: the value to check

    Returns: index in array where the value is the nearest

    """
    idx = (np.abs(array - value)).argmin()
    return idx

def get_dda_scan_param(
    mz,
    intensity,
    precursor_scan_id,
    isolation_width,
    mz_tol,
    rt_tol,
    agc_target=DEFAULT_MS2_AGC_TARGET,
    max_it=DEFAULT_MS2_MAXIT,
    collision_energy=DEFAULT_MS2_COLLISION_ENERGY,
    source_cid_energy=DEFAULT_SOURCE_CID_ENERGY,
    orbitrap_resolution=DEFAULT_MS2_ORBITRAP_RESOLUTION,
    mass_analyser=DEFAULT_MS2_MASS_ANALYSER,
    activation_type=DEFAULT_MS1_ACTIVATION_TYPE,
    isolation_mode=DEFAULT_MS2_ISOLATION_MODE,
    polarity=POSITIVE,
    metadata=None,
    scan_id=None,
):
    """
    Generate the default MS2 scan parameters.

    Args:
        mz: m/z of precursor peak to fragment
        intensity: intensity of precursor peak to fragment
        precursor_scan_id: scan ID of the MS1 scan containing the precursor peak
        isolation_width: isolation width, in Dalton
        mz_tol: m/z tolerance for dynamic exclusion # FIXME: this shouldn't be here
        rt_tol: RT tolerance for dynamic exclusion # FIXME: this shouldn't be here
        agc_target: AGC (automatic gain control) target
        max_it: maximum time to collect ion
        collision_energy: the collision energy to use
        source_cid_energy: source CID energy
        orbitrap_resolution: resolution of the mass-spec (Orbitrap) instrument
        mass_analyser: which mass analyser to use
        activation_type: activation type, either HCD or CID
        isolation_mode: isolation mode, either None, or Quadrupole or IonTrap
        polarity: the polarity value, either POSITIVE or NEGATIVE
        metadata: additional metadata to include in this scan
        scan_id: the scan ID, if specified

    Returns: the parameters of the MS2 scan to create

    """

    dda_scan_params = ScanParameters()
    dda_scan_params.set(ScanParameters.MS_LEVEL, 2)

    assert isinstance(mz, list) == isinstance(intensity, list)

    # create precursor object, assume it's all singly charged
    precursor_charge = +1 if (polarity == POSITIVE) else -1
    if isinstance(mz, list):
        precursor_list = []
        for i, m in enumerate(mz):
            precursor_list.append(
                Precursor(
                    precursor_mz=m,
                    precursor_intensity=intensity[i],
                    precursor_charge=precursor_charge,
                    precursor_scan_id=precursor_scan_id,
                )
            )
        dda_scan_params.set(ScanParameters.PRECURSOR_MZ, precursor_list)

        if isinstance(isolation_width, list):
            assert len(isolation_width) == len(precursor_list)
        else:
            isolation_width = [isolation_width for m in mz]
        dda_scan_params.set(ScanParameters.ISOLATION_WIDTH, isolation_width)

    else:
        precursor = Precursor(
            precursor_mz=mz,
            precursor_intensity=intensity,
            precursor_charge=precursor_charge,
            precursor_scan_id=precursor_scan_id,
        )
        precursor_list = [precursor]
        dda_scan_params.set(ScanParameters.PRECURSOR_MZ, precursor_list)

        # set the full-width isolation width, in Da
        # if mz is not a list, neither should isolation_width be
        assert not isinstance(isolation_width, list)
        isolation_width = [isolation_width]
        dda_scan_params.set(ScanParameters.ISOLATION_WIDTH, isolation_width)

    # define dynamic exclusion parameters
    dda_scan_params.set(ScanParameters.DYNAMIC_EXCLUSION_MZ_TOL, mz_tol)
    dda_scan_params.set(ScanParameters.DYNAMIC_EXCLUSION_RT_TOL, rt_tol)

    # define other fragmentation parameters
    dda_scan_params.set(ScanParameters.COLLISION_ENERGY, collision_energy)
    dda_scan_params.set(ScanParameters.ORBITRAP_RESOLUTION, orbitrap_resolution)
    dda_scan_params.set(ScanParameters.ACTIVATION_TYPE, activation_type)
    dda_scan_params.set(ScanParameters.MASS_ANALYSER, mass_analyser)
    dda_scan_params.set(ScanParameters.ISOLATION_MODE, isolation_mode)
    dda_scan_params.set(ScanParameters.AGC_TARGET, agc_target)
    dda_scan_params.set(ScanParameters.MAX_IT, max_it)
    dda_scan_params.set(ScanParameters.SOURCE_CID_ENERGY, source_cid_energy)
    dda_scan_params.set(ScanParameters.POLARITY, polarity)
    dda_scan_params.set(ScanParameters.FIRST_MASS, DEFAULT_MSN_SCAN_WINDOW[0])
    dda_scan_params.set(ScanParameters.METADATA, metadata)
    dda_scan_params.set(ScanParameters.SCAN_ID, scan_id)

    # dynamically scale the upper mass
    charge = 1
    wiggle_room = 1.1
    max_precursor_mz = max(
        [(p.precursor_mz + isol / 2) for (p, isol) in zip(precursor_list, isolation_width)]
    )
    last_mass = max_precursor_mz * charge * wiggle_room
    dda_scan_params.set(ScanParameters.LAST_MASS, last_mass)
    return dda_scan_params

def get_default_scan_params(
    polarity=POSITIVE,
    agc_target=DEFAULT_MS1_AGC_TARGET,
    max_it=DEFAULT_MS1_MAXIT,
    collision_energy=DEFAULT_MS1_COLLISION_ENERGY,
    source_cid_energy=DEFAULT_SOURCE_CID_ENERGY,
    orbitrap_resolution=DEFAULT_MS1_ORBITRAP_RESOLUTION,
    default_ms1_scan_window=DEFAULT_MS1_SCAN_WINDOW,
    mass_analyser=DEFAULT_MS1_MASS_ANALYSER,
    activation_type=DEFAULT_MS1_ACTIVATION_TYPE,
    isolation_mode=DEFAULT_MS1_ISOLATION_MODE,
    metadata=None,
    scan_id=None,
):
    """
    Generate the default MS1 scan parameters.

    Args:
        polarity: the polarity value, either POSITIVE or NEGATIVE
        agc_target: AGC (automatic gain control) target
        max_it: maximum time to collect ion
        collision_energy: the collision energy to use
        source_cid_energy: source CID energy
        orbitrap_resolution: resolution of the mass-spec (Orbitrap) instrument
        default_ms1_scan_window: the default MS1 scan window
        mass_analyser: which mass analyser to use
        activation_type: activation type, either HCD or CID
        isolation_mode: isolation mode, either None, or Quadrupole or IonTrap
        metadata: additional metadata to include in this scan
        scan_id: the scan ID, if specified

    Returns: the parameters of the MS1 scan to create

    """
    default_scan_params = ScanParameters()
    default_scan_params.set(ScanParameters.MS_LEVEL, 1)
    default_scan_params.set(ScanParameters.ISOLATION_WINDOWS, [[default_ms1_scan_window]])
    default_scan_params.set(ScanParameters.ISOLATION_WIDTH, DEFAULT_ISOLATION_WIDTH)
    default_scan_params.set(ScanParameters.COLLISION_ENERGY, collision_energy)
    default_scan_params.set(ScanParameters.ORBITRAP_RESOLUTION, orbitrap_resolution)
    default_scan_params.set(ScanParameters.ACTIVATION_TYPE, activation_type)
    default_scan_params.set(ScanParameters.MASS_ANALYSER, mass_analyser)
    default_scan_params.set(ScanParameters.ISOLATION_MODE, isolation_mode)
    default_scan_params.set(ScanParameters.AGC_TARGET, agc_target)
    default_scan_params.set(ScanParameters.MAX_IT, max_it)
    default_scan_params.set(ScanParameters.SOURCE_CID_ENERGY, source_cid_energy)
    default_scan_params.set(ScanParameters.POLARITY, polarity)
    default_scan_params.set(ScanParameters.FIRST_MASS, default_ms1_scan_window[0])
    default_scan_params.set(ScanParameters.LAST_MASS, default_ms1_scan_window[1])
    default_scan_params.set(ScanParameters.METADATA, metadata)
    default_scan_params.set(ScanParameters.SCAN_ID, scan_id)
    return default_scan_params

def get_rt(spectrum):
    """
    Extracts RT value from a pymzml spectrum object

    Args:
        spectrum: a pymzml spectrum object

    Returns: the retention time (in seconds)

    """
    rt, units = spectrum.scan_time
    if units == "minute":
        rt *= 60.0
    return rt

def load_obj(filename):
    """
    Load saved object from file

    Args:
        filename: The filename to load. Should be saved using the `save_obj` method.

    Returns: the loaded object

    """
    try:
        with gzip.GzipFile(filename, "rb") as f:
            return pickle.load(f)
    except OSError:
        logger.warning(
            "Old, invalid or missing pickle in %s. " "Please regenerate this file." % filename
        )
        raise

def save_obj(obj, filename):
    """
    Save object to file. This is useful for storing simulation results and other objects.

    If the directory containing the specified filename doesn't exist, it will be created first.
    The object will be saved using gzip + pickle (highest protocol).

    Args:
        obj: the Python object to save
        filename: the output filename to use

    Returns: None
    """

    # workaround for
    # TypeError: can't pickle _thread.lock objects
    # when trying to pickle a progress bar
    if hasattr(obj, "bar"):
        obj.bar = None

    out_dir = os.path.dirname(filename)
    create_if_not_exist(out_dir)
    logger.info("Saving %s to %s" % (type(obj), filename))
    with gzip.GzipFile(filename, "w") as f:
        pickle.dump(obj, f, protocol=pickle.HIGHEST_PROTOCOL)

def set_log_level(level, remove_id=None):
    """
    Set the logging level of the default logger
    Args:
        level: the new level to set
        remove_id: ID of previous log handler to be removed

    Returns: the new log handler after setting the log level

    """
    if remove_id is None:
        logger.remove()  # remove all previously set handlers
    else:
        try:
            logger.remove(remove_id)  # remove previously set handler by id
        except ValueError:  # can't find the handler
            pass

    # add new handler at the desired log level
    new_handler_id = logger.add(sys.stderr, level=level)
    return new_handler_id

def set_log_level_debug(remove_id=None):
    """
    Set log level to DEBUG
    Args:
        remove_id: ID of previous log handler to be removed

    Returns: None

    """
    return set_log_level(logging.DEBUG, remove_id=remove_id)

def set_log_level_info(remove_id=None):
    """
    Set log level to INFO
    Args:
        remove_id: ID of previous log handler to be removed

    Returns: None

    """
    return set_log_level(logging.INFO, remove_id=remove_id)

def set_log_level_warning(remove_id=None):
    """
    Set log level to WARNING
    Args:
        remove_id: ID of previous log handler to be removed

    Returns: None

    """
    return set_log_level(logging.WARNING, remove_id=remove_id)

def uniform_list(N, min_val, max_val):
    """
    Generates a list of N uniformly random values from min_val to max_val
    Args:
        N: the number of items to generate
        min_val: the minimum range
        max_val: the maximum range

    Returns: a list of N values

    """
    return list(np.random.rand(N) * (max_val - min_val) + min_val)