FETCH_STUBS = True
ADDUCTS = ('','[M]+','[M+H]+','[M+H]2+','[M+2H]2+','[M+H-H2O]2+','[M+K]2+','[M+NH4]+','[M+Na]+','[M+H-H2O]+','[M-H]-','[M-2H]-','[M-H+Cl]-','[M-2H]2-','[M+Cl]-','[2M+H]+','[2M-H]-','[M-H+Na]+','[M+K]+','[M+2Na]2+','[M-e]+','[M+acetate]-','[M+formate]-','[M-H+Cl]2-','[M-H+2Na]+')
POLARITY = ('positive', 'negative', 'alternating')
FRAGMENTATION_TECHNIQUE = ('hcd','cid','etd','ecd','irmpd')
ID_GRADES: Dict[str, IdentificationGrade] = dict()
    FETCH_STUBS = False
        FETCH_STUBS = True
    FETCH_STUBS = True
def retrieve(object_type, **kwargs):
def remove(object_type, **kwargs):
def remove_objects(objects, all_versions=True, **kwargs):
def store(objects, **kwargs):
    def __init__(self, **kwargs):
    def _update(self, override_user=False):
    def clone(self, recursive=False):
    def show_diff(self, unique_id=None):
    def _on_update(self, change):
    def __str__(self):
    def __repr__(self):
    def __getattribute__(self, name):
    def validate(self, obj, value):
def find_invalid_runs(**kwargs):
def to_dataframe(objects):
class MetatlasObject(HasTraits):
class Method(MetatlasObject):
class Sample(MetatlasObject):
class LcmsRun(MetatlasObject):
class FunctionalSet(MetatlasObject):
class Compound(MetatlasObject):
class Reference(MetatlasObject):
class IdentificationGrade(MetatlasObject):
class _IdGradeTrait(MetInstance):
class Group(MetatlasObject):
class MzIntensityPair(MetatlasObject):
class FragmentationReference(Reference):
class RtReference(Reference):
class MzReference(Reference):
class IntensityReference(Reference):
class CompoundIdentification(MetatlasObject):
class Atlas(MetatlasObject):
class MZMineTask(MetatlasObject): def log_errors(func=None, *, output_context=None):
    """Define function decorator to log and re-raise all exceptions"""
    if func is None:
        return partial(log_errors, output_context=output_context)

    @wraps(func)
    def wrapper(*args, **kwargs):
        # pylint: disable=broad-except
        try:
            return func(*args, **kwargs)
        except Exception as err:
            if output_context:
                with output_context:
                    logger.exception(err)
                    raise Exception from err
            else:
                logger.exception(err)
                raise Exception from err

    return wrapper def create_msms_dataframe(df):
def compare_EIC_to_BPC_for_file(metatlas_dataset,file_index,yscale = 'linear'):
def get_data_for_atlas_df_and_file(input_tuple):
def get_bpc(filename,dataset='ms1_pos',integration='bpc'):
def df_container_from_metatlas_file(my_file):
def fast_nearest_interp(xi, x, y):
def remove_ms1_data_not_in_atlas(atlas_df, data):
def extract(data, ids, default=None):
def set_nested(data, ids, value):
def make_atlas_df(atlas):
def transfer_identification_data_to_atlas(data, atlas, ids_list=None):
def get_data_for_mzrt(row, data_df_pos, data_df_neg, extra_time=0.5, use_mz='mz', extra_mz=0.0):
def get_ms1_summary(row):
def get_ms2_data(row):
def prefilter_ms1_dataframe_with_boundaries(data_df, rt_max, rt_min, mz_min, mz_max, extra_time=0.5, extra_mz=0.01):
def get_ms1_eic(row):
def retrieve_most_intense_msms_scan(data):
def get_data_for_atlas_and_lcmsrun(atlas_df, df_container, extra_time, extra_mz):
    def get_feature_data(atlas_df, pos_df, neg_df, use_mz='mz'):
def get_ms2_dict(ms2_feature_data_df):
def get_ms1_summary_data(ms1_feature_data_df):
def get_eic_data(ms1_feature_data_df):
def get_unique_scan_data(data):
def get_non_redundant_precursor_list(prt,pmz,rt_cutoff,mz_cutoff):
def organize_msms_scan_data(data,list_of_prt,list_of_pmz,list_of_pintensity):
def get_data_for_a_compound(mz_ref,rt_ref,what_to_get,h5file,extra_time):
def get_dill_data(fname):
def get_group_names(data):
def get_group_shortnames(data):
def get_file_names(data,full_path=False):
def get_compound_names(data,use_labels=False):
def make_data_sources_tables(groups, myatlas, output_loc, polarity=None, overwrite=True):
    D = defaultdict(list)
    D = {k:v for k,v in D.items() if len(v)>1} def make_dir_for(file_path):
def check_existing_file(file_path, overwrite=False):
def export_dataframe(dataframe, file_path, description, overwrite=False, **kwargs):
def raise_on_diff(dataframe, file_path, description, **kwargs):
def export_dataframe_die_on_diff(dataframe, file_path, description, overwrite=False, **kwargs): def extract(data, ids, default=None):
    """
    inputs:
        data: hierarchical data structure consisting of lists, dicts, and objects with attributes.
        ids: a list of idices, key names, and attribute names
        default: optional object
    output:
        the value stored at the location indicated by the ids list or default if the location
        does not exist.

    Strings in ids are first tried as key name and if no such key name exists, then they are
    tried as attribute names. To designate that a member of ids should be used as an attribute
    and not a key name, make it a tuple with the attribute name string as the first member, such
    as: ('attribute_name',). If you want to make it more explict to the reader, you can add a
    second member to the tuple, which will not be used, such as ('attribute_name', 'as attribute')
    """
    if len(ids) == 0:
        return data
    try:
        if isinstance(ids[0], tuple):
            sub_data = getattr(data, ids[0][0])
        else:
            try:
                sub_data = data[ids[0]]
            except TypeError:
                sub_data = getattr(data, ids[0])
    except (AttributeError, IndexError, KeyError):
        return default
    else:
        return extract(sub_data, ids[1:], default) def save_compound_eic_pdf(
    data: MetatlasDataset,
    compound_idx: int,
    file_name: str,
    overwrite: bool = False,
    sharey: bool = True,
    max_plots_per_page: int = 30,
) -> None:
    """Generate a PDF of EIC plots for compound_idx within data"""
    file_order_eics = []
    for sample in data:
        compound = sample[compound_idx]
        fields = Path(compound["lcmsrun"].name).stem.split("_")
        title = f"{fields[0]}_{fields[11]}_{fields[13]}_{fields[15]}"
        group_name = compound["group"].short_name
        file_order_eics.append(CompoundEic(title, group_name, compound))
    eics = sorted(file_order_eics, key=lambda x: (x.group_name, x.title))
    compound_name = data[0][compound_idx]["identification"].compound[0].name
    with plot_set.PlotSet(eics, max_plots_per_page, sharey) as plots:
        plots.save_pdf(file_name, title=f"EICs for {compound_name}", overwrite=overwrite) def save_sample_tic_pdf(
    data: MetatlasDataset,
    polarity: str,
    file_name: str,
    overwrite: bool = False,
    sharey: bool = True,
    x_min: Optional[float] = None,
    x_max: Optional[float] = None,
    y_min: Optional[float] = None,
    y_max: Optional[float] = None,
    max_plots_per_page: int = 30,
) -> None:
    """Generate a PDF of TIC plots for samples within data"""
    assert polarity in ["positive", "negative"]
    file_order_tics = []
    for sample in data:
        fields = Path(sample[0]["lcmsrun"].name).stem.split("_")
        title = f"{fields[0]}_{fields[11]}_{fields[13]}_{fields[15]}"
        group_name = sample[0]["group"].short_name
        h5_file_name = sample[0]["lcmsrun"].hdf5_file
        file_order_tics.append(Tic(title, group_name, h5_file_name, polarity))
    tics = sorted(file_order_tics, key=lambda x: (x.group_name, x.title))
    with plot_set.PlotSet(tics, max_plots_per_page, x_min, x_max, y_min, y_max, sharey) as plots:
        plots.save_pdf(file_name, title="TICs", overwrite=overwrite) def pdf_with_text(text: str, file_name: str, size: int = 24):
    """saves a pdf file with a text message"""
    with PdfPages(file_name) as pdf:
        firstPage = plt.figure(figsize=(11.0, 8.5))
        firstPage.clf()
        firstPage.text(0.5, 0.5, text, transform=firstPage.transFigure, size=size, ha="center")
        pdf.savefig()
        plt.close() def parallel_process(function, data, max_cpus, unit="it"): def or_default(none_or_value, default):
    """
    inputs:
        none_or_value: variable to test
        default: value to return if none_or_value is None
    """
    return none_or_value if none_or_value is not None else default
import logging
import os
import os.path
import warnings
import pandas as pd
import dill
import numpy as np
import json
import matplotlib.pyplot as plt
import ipywidgets as widgets
import getpass
import matplotlib.patches
import gspread
from copy import deepcopy
from typing import Callable, List, Optional, Sized
from enum import Enum
from metatlas.datastructures import metatlas_objects as metob
from metatlas.tools.logging import log_errors
from metatlas.io import metatlas_get_data_helper_fun as ma_data
from metatlas.io import write_utils
from metatlas.io.metatlas_get_data_helper_fun import extract
from metatlas.plots.compound_eic import save_compound_eic_pdf
from metatlas.plots.tic import save_sample_tic_pdf
from metatlas.plots.utils import pdf_with_text
from metatlas.tools import fastanalysis
from metatlas.tools import parallel
from metatlas.tools import spectralprocessing as sp
from tqdm.notebook import tqdm
from textwrap import fill, TextWrapper
from rdkit import Chem
from rdkit.Chem import Draw, rdDepictor
from rdkit.Chem.Draw import rdMolDraw2D
from itertools import cycle
from ipywidgets import interact
from IPython.display import display, HTML
from IPython import get_ipython
from ast import literal_eval
from datetime import datetime
from matplotlib.widgets import Slider, RadioButtons
from oauth2client.service_account import ServiceAccountCredentials
from metatlas.tools.util import or_default
from functools import reduce
from io import StringIO
    mz_theoretical = data[file_idx][compound_idx]['identification'].mz_references[0].mz
    ms1_df = get_ms1_df(data[file_idx][compound_idx])
    mz_measured = get_mz_centroid(ms1_df)
    rt_ms1 = get_rt_peak(ms1_df)

    delta_mz = abs(mz_theoretical - mz_measured)
    delta_ppm = delta_mz / mz_theoretical * 1e6
    mz_header = ["m/z theoretical = %5.4f" % mz_theoretical,
                 "m/z measured = %5.4f" % mz_measured,
                 "ppm diff = %3.2f" % delta_ppm]
    rt_header = ["RT theoretical = %3.2f" % rt_theoretical,
                 "RT MS1 measured = %3.2f" % rt_ms1]
    if not hits.empty:
        mz_header.insert(0, "precursor m/z = %5.4f" % mz_precursor)
        rt_header.append("RT MS2 measured = %3.2f" % rt_ms2)
    return (', '.join(mz_header), ', '.join(rt_header),
            get_similar_compounds_header(similar_compounds, compound_idx))


def get_similar_compounds_header(similar_compounds, compound_idx):
    """
    inputs:
        similar_compounds: the output from get_similar_compounds()
        compound_index: index of current compound being considered
    returns:
    """
    if len(similar_compounds) < 2:
        return ''
    joined = '; '.join([_similar_compound_to_str(compound, compound_idx) for compound in similar_compounds])
    return f"Similar Compounds = {joined}"


def _similar_compound_to_str(sdict, compound_idx):
    """
    inputs:
        sdict: a dict returned from get_similar_compounds()
        compound_index: index of current compound being considered
    returns:
        string with only non-breaking spaces or '*' if dict represents current compound
    """
    if sdict['index'] == compound_idx:
        return '*'
    return f"{sdict['index']}, {sdict['label']} {{RT-{sdict['rt'].rt_peak:.2f}}}".replace(' ', '\xa0')


def get_msms_plot_data(hits, hit_ctr):
    """
    inputs:
        hits: dataframe of msms hits
        hit_ctr: index of current hit in hits
    returns:
        tuple: (hit_ref_id, hit_score, hit_query, hit_ref)
    """
    if hits.empty:
        hit_ref_id = "N/A"
        hit_score = np.nan
        hit_query = np.full((2, 2, ), np.nan)
        hit_ref = hit_query
    else:
        hit_ref_id = hits.index.get_level_values('id')[hit_ctr]
        hit_score = hits['score'].iloc[hit_ctr]
        hit_query = hits['msv_query_aligned'][hit_ctr:hit_ctr+1].iloc[0]
        hit_ref = hits['msv_ref_aligned'][hit_ctr:hit_ctr+1].iloc[0]
    return (hit_ref_id, hit_score, hit_query, hit_ref)


def get_hit_metadata(data, hits, file_names, hit_ctr, compound_idx):
    """
    returns a tuple containing:
        file name (without path) of the hit or the string 'None'
        compound object for the hit or None
    """
    if not hits.empty:
        hit_file_name = hits.index.get_level_values('file_name')[hit_ctr]
        return (hit_file_name, data[int(file_names.index(hit_file_name))][compound_idx])
    file_idx = file_with_max_ms1_intensity(data, compound_idx, limit_to_rt_range=True)[0]
    if file_idx:
        return (os.path.basename(data[file_idx][compound_idx]['lcmsrun'].hdf5_file),
                data[file_idx][compound_idx])
    return ('None', None)


def within_tolerance(measured, theoretical, tolerance):
    """ Returns True if normalized, absolute difference is with tolerance """
    return abs(measured - theoretical)/theoretical <= tolerance


def layout_radio_button_set(area, anchor='SW'):
    """
    inputs:
        area: [left, bottom, width, height]
        anchor: string for anchor direction
    returns:
        an axes for radiobuttons at area with axis off and equal aspect ratio
    """
    axes = plt.axes(area, anchor=anchor, aspect='equal')
    axes.axis('off')
    return axes


def rt_range_overlaps(rt1, rt2):
    """
    inputs:
        rt1: metatlas.datastructures.metatlas_objects.RtReference
        rt2: metatlas.datastructures.metatlas_objects.RtReference
    returns:
        True if there is overlap in the RT min-max regions of rt1 and and rt2
    """
    return ((rt2.rt_min <= rt1.rt_min <= rt2.rt_max) or (rt2.rt_min <= rt1.rt_max <= rt2.rt_max) or
            (rt1.rt_min <= rt2.rt_min <= rt1.rt_max) or (rt1.rt_min <= rt2.rt_max <= rt1.rt_max))


def disable_interactive_plots():
    """Close interactive figures and turn off interactive plotting"""
    adjust_rt_for_selected_compound.disable()
    plt.ioff()


def tic_pdf(data, polarity, file_name, overwrite=False, sharey=True,
            x_min=1.5, x_max=None, y_min=0, y_max=None, max_plots_per_page=30):

    save_sample_tic_pdf(