import requests import simplejson as json from flask import g, request, url_for, Blueprint, current_app as app import utility.hmac as hmac from base import webqtlConfig from base.webqtlCaseData import webqtlCaseData from base.data_set import create_dataset from utility.authentication_tools import check_resource_availability from utility.configuration import get_setting from utility.redis_tools import get_redis_conn, get_resource_id from wqflask.database import database_connection Redis = get_redis_conn() trait_bp = Blueprint("trait", __name__) def create_trait(**kw): assert bool(kw.get('dataset')) != bool( kw.get('dataset_name')), "Needs dataset ob. or name" assert bool(kw.get('name')), "Needs trait name" if bool(kw.get('dataset')): dataset = kw.get('dataset') else: if kw.get('dataset_name') != "Temp": dataset = create_dataset(kw.get('dataset_name')) else: dataset = create_dataset( dataset_name="Temp", dataset_type="Temp", group_name= kw.get('name').split("_")[2]) if dataset.type == 'Publish': permissions = check_resource_availability( dataset, g.user_session.user_id, kw.get('name')) else: permissions = check_resource_availability( dataset, g.user_session.user_id) if permissions['data'] != "no-access": the_trait = GeneralTrait(**dict(kw,dataset=dataset)) if the_trait.dataset.type != "Temp": the_trait = retrieve_trait_info( the_trait, the_trait.dataset, get_qtl_info=kw.get('get_qtl_info')) return the_trait else: return None class GeneralTrait: """ Trait class defines a trait in webqtl, can be either Microarray, Published phenotype, genotype, or user input trait """ def __init__(self, get_qtl_info=False, get_sample_info=True, **kw): # xor assertion assert kw.get("dataset"), "Dataset obj is needed as a kwarg" # Trait ID, ProbeSet ID, Published ID, etc. self.name = kw.get('name') self.dataset = kw.get("dataset") self.cellid = kw.get('cellid') self.identification = kw.get('identification', 'un-named trait') self.haveinfo = kw.get('haveinfo', False) # Blat sequence, available for ProbeSet self.sequence = kw.get('sequence') self.data = kw.get('data', {}) self.view = True # Sets defaults self.locus = None self.lrs = None self.pvalue = None self.mean = None self.additive = None self.num_overlap = None self.strand_probe = None self.symbol = None self.abbreviation = None self.display_name = self.name self.LRS_score_repr = "N/A" self.LRS_location_repr = "N/A" self.chr = self.mb = self.locus_chr = self.locus_mb = "" if kw.get('fullname'): name2 = value.split("::") if len(name2) == 2: self.dataset, self.name = name2 # self.cellid is set to None above elif len(name2) == 3: self.dataset, self.name, self.cellid = name2 # Todo: These two lines are necessary most of the time, but # perhaps not all of the time So we could add a simple if # statement to short-circuit this if necessary if get_sample_info is not False: self = retrieve_sample_data(self, self.dataset) def export_informative(self, include_variance=0): """ export informative sample mostly used in qtl regression """ samples = [] vals = [] the_vars = [] sample_aliases = [] for sample_name, sample_data in list(self.data.items()): if sample_data.value is not None: if not include_variance or sample_data.variance is not None: samples.append(sample_name) vals.append(sample_data.value) the_vars.append(sample_data.variance) sample_aliases.append(sample_data.name2) return samples, vals, the_vars, sample_aliases @property def description_fmt(self): """Return a text formated description""" if self.dataset.type == 'ProbeSet': if self.description: formatted = self.description if self.probe_target_description: formatted += "; " + self.probe_target_description else: formatted = "Not available" elif self.dataset.type == 'Publish': if self.confidential: formatted = self.pre_publication_description else: formatted = self.post_publication_description else: formatted = "Not available" if isinstance(formatted, bytes): formatted = formatted.decode("utf-8") return formatted @property def alias_fmt(self): """Return a text formatted alias""" alias = 'Not available' if getattr(self, "alias", None): alias = self.alias.replace(";", " ") alias = ", ".join(alias.split()) return alias @property def wikidata_alias_fmt(self): """Return a text formatted alias""" alias = 'Not available' if self.symbol: human_response = requests.get( get_setting(app, "GN2_BASE_URL") + "gn3/gene/aliases/" + self.symbol.upper()) mouse_response = requests.get( get_setting(app, "GN2_BASE_URL") + "gn3/gene/aliases/" + self.symbol.capitalize()) other_response = requests.get( get_setting(app, "GN2_BASE_URL") + "gn3/gene/aliases/" + self.symbol.lower()) if human_response and mouse_response and other_response: alias_list = json.loads(human_response.content) + json.loads( mouse_response.content) + \ json.loads(other_response.content) filtered_aliases = [] seen = set() for item in alias_list: if item in seen: continue else: filtered_aliases.append(item) seen.add(item) alias = "; ".join(filtered_aliases) return alias @property def location_fmt(self): """Return a text formatted location While we're at it we set self.location in case we need it later (do we?) """ if self.chr == "Un": return 'Not available' if self.chr and self.mb: self.location = 'Chr %s @ %s Mb' % (self.chr, self.mb) elif self.chr: self.location = 'Chr %s @ Unknown position' % (self.chr) else: self.location = 'Not available' fmt = self.location # XZ: deal with direction if self.strand_probe == '+': fmt += (' on the plus strand ') elif self.strand_probe == '-': fmt += (' on the minus strand ') return fmt def retrieve_sample_data(trait, dataset, samplelist=None): if samplelist is None: samplelist = [] if dataset.type == "Temp": results = Redis.get(trait.name).split() else: results = dataset.retrieve_sample_data(trait.name) # Todo: is this necessary? If not remove trait.data.clear() if results: if dataset.type == "Temp": all_samples_ordered = dataset.group.all_samples_ordered() for i, item in enumerate(results): try: trait.data[all_samples_ordered[i]] = webqtlCaseData( all_samples_ordered[i], float(item)) except: pass else: for item in results: name, value, variance, num_cases, name2 = item if not samplelist or (samplelist and name in samplelist): # name, value, variance, num_cases) trait.data[name] = webqtlCaseData(*item) return trait @trait_bp.route("/trait/get_sample_data") def get_sample_data(): params = request.args trait = params['trait'] dataset = params['dataset'] trait_ob = create_trait(name=trait, dataset_name=dataset) if trait_ob: trait_dict = {} trait_dict['name'] = trait trait_dict['db'] = dataset trait_dict['type'] = trait_ob.dataset.type trait_dict['group'] = trait_ob.dataset.group.name trait_dict['tissue'] = trait_ob.dataset.tissue trait_dict['species'] = trait_ob.dataset.group.species trait_dict['url'] = url_for( 'show_trait_page', trait_id=trait, dataset=dataset) if trait_ob.dataset.type == "ProbeSet": trait_dict['symbol'] = trait_ob.symbol trait_dict['location'] = trait_ob.location_repr trait_dict['description'] = trait_ob.description_display elif trait_ob.dataset.type == "Publish": trait_dict['description'] = trait_ob.description_display if trait_ob.pubmed_id: trait_dict['pubmed_link'] = trait_ob.pubmed_link trait_dict['pubmed_text'] = trait_ob.pubmed_text else: trait_dict['location'] = trait_ob.location_repr return json.dumps([trait_dict, {key: value.value for key, value in list( trait_ob.data.items())}]) else: return None def jsonable(trait, dataset=None): """Return a dict suitable for using as json Actual turning into json doesn't happen here though""" if not dataset: dataset = create_dataset(dataset_name=trait.dataset.name, dataset_type=trait.dataset.type, group_name=trait.dataset.group.name) trait_symbol = "N/A" trait_mean = "N/A" if trait.symbol: trait_symbol = trait.symbol if trait.mean: trait_mean = trait.mean if dataset.type == "ProbeSet": return dict(name=trait.name, display_name=trait.display_name, hmac=hmac.data_hmac('{}:{}'.format(trait.display_name, dataset.name)), view=str(trait.view), symbol=trait_symbol, dataset=dataset.name, dataset_name=dataset.shortname, description=trait.description_display, mean=trait_mean, location=trait.location_repr, chr=trait.chr, mb=trait.mb, lrs_score=trait.LRS_score_repr, lrs_location=trait.LRS_location_repr, lrs_chr=trait.locus_chr, lrs_mb=trait.locus_mb, additive=trait.additive ) elif dataset.type == "Publish": if trait.pubmed_id: return dict(name=trait.name, display_name=trait.display_name, hmac=hmac.data_hmac('{}:{}'.format(trait.name, dataset.name)), view=str(trait.view), symbol=trait.abbreviation, dataset=dataset.name, dataset_name=dataset.shortname, description=trait.description_display, abbreviation=trait.abbreviation, authors=trait.authors, pubmed_id=trait.pubmed_id, pubmed_text=trait.pubmed_text, pubmed_link=trait.pubmed_link, mean=trait_mean, lrs_score=trait.LRS_score_repr, lrs_location=trait.LRS_location_repr, lrs_chr=trait.locus_chr, lrs_mb=trait.locus_mb, additive=trait.additive ) else: return dict(name=trait.name, display_name=trait.display_name, hmac=hmac.data_hmac('{}:{}'.format(trait.name, dataset.name)), view=str(trait.view), symbol=trait.abbreviation, dataset=dataset.name, dataset_name=dataset.shortname, description=trait.description_display, abbreviation=trait.abbreviation, authors=trait.authors, pubmed_text=trait.pubmed_text, mean=trait_mean, lrs_score=trait.LRS_score_repr, lrs_location=trait.LRS_location_repr, lrs_chr=trait.locus_chr, lrs_mb=trait.locus_mb, additive=trait.additive ) elif dataset.type == "Geno": return dict(name=trait.name, display_name=trait.display_name, hmac=hmac.data_hmac('{}:{}'.format(trait.display_name, dataset.name)), view=str(trait.view), dataset=dataset.name, dataset_name=dataset.shortname, location=trait.location_repr, chr=trait.chr, mb=trait.mb ) elif dataset.name == "Temp": return dict(name=trait.name, display_name=trait.display_name, hmac=hmac.data_hmac('{}:{}'.format(trait.display_name, dataset.name)), view=str(trait.view), dataset="Temp", dataset_name="Temp") else: return dict() def retrieve_trait_info(trait, dataset, get_qtl_info=False): if not dataset: raise ValueError("Dataset doesn't exist") with database_connection() as conn, conn.cursor() as cursor: trait_info = () if dataset.type == 'Publish': cursor.execute( "SELECT PublishXRef.Id, InbredSet.InbredSetCode, " "Publication.PubMed_ID, " "CAST(Phenotype.Pre_publication_description AS BINARY), " "CAST(Phenotype.Post_publication_description AS BINARY), " "CAST(Phenotype.Original_description AS BINARY), " "CAST(Phenotype.Pre_publication_abbreviation AS BINARY), " "CAST(Phenotype.Post_publication_abbreviation AS BINARY), " "PublishXRef.mean, Phenotype.Lab_code, " "Phenotype.Submitter, Phenotype.Owner, " "Phenotype.Authorized_Users, " "CAST(Publication.Authors AS BINARY), " "CAST(Publication.Title AS BINARY), " "CAST(Publication.Abstract AS BINARY), " "CAST(Publication.Journal AS BINARY), " "Publication.Volume, Publication.Pages, " "Publication.Month, Publication.Year, " "PublishXRef.Sequence, Phenotype.Units, " "PublishXRef.comments FROM PublishXRef, Publication, " "Phenotype, PublishFreeze, InbredSet WHERE " "PublishXRef.Id = %s AND " "Phenotype.Id = PublishXRef.PhenotypeId " "AND Publication.Id = PublishXRef.PublicationId " "AND PublishXRef.InbredSetId = PublishFreeze.InbredSetId " "AND PublishXRef.InbredSetId = InbredSet.Id AND " "PublishFreeze.Id = %s", (trait.name, dataset.id,) ) trait_info = cursor.fetchone() # XZ, 05/08/2009: Xiaodong add this block to use ProbeSet.Id to find the probeset instead of just using ProbeSet.Name # XZ, 05/08/2009: to avoid the problem of same probeset name from different platforms. elif dataset.type == 'ProbeSet': display_fields_string = ', ProbeSet.'.join(dataset.display_fields) display_fields_string = f'ProbeSet.{display_fields_string}' cursor.execute( f"SELECT {display_fields_string} FROM ProbeSet, ProbeSetFreeze, " "ProbeSetXRef WHERE " "ProbeSetXRef.ProbeSetFreezeId = ProbeSetFreeze.Id " "AND ProbeSetXRef.ProbeSetId = ProbeSet.Id AND " "ProbeSetFreeze.Name = %s AND " "ProbeSet.Name = %s", (dataset.name, str(trait.name),) ) trait_info = cursor.fetchone() # XZ, 05/08/2009: We also should use Geno.Id to find marker instead of just using Geno.Name # to avoid the problem of same marker name from different species. elif dataset.type == 'Geno': display_fields_string = ',Geno.'.join(dataset.display_fields) display_fields_string = f'Geno.{display_fields_string}' cursor.execute( f"SELECT {display_fields_string} FROM Geno, GenoFreeze, " "GenoXRef WHERE " "GenoXRef.GenoFreezeId = GenoFreeze.Id " "AND GenoXRef.GenoId = Geno.Id " "AND GenoFreeze.Name = %s " "AND Geno.Name = %s", (dataset.name, trait.name) ) trait_info = cursor.fetchone() else: # Temp type cursor.execute( f"SELECT {','.join(dataset.display_fields)} " f"FROM {dataset.type} WHERE Name = %s", (trait.name,) ) trait_info = cursor.fetchone() if trait_info: trait.haveinfo = True for i, field in enumerate(dataset.display_fields): holder = trait_info[i] if isinstance(holder, bytes): holder = holder.decode("utf-8", errors="ignore") setattr(trait, field, holder) if dataset.type == 'Publish': if trait.group_code: trait.display_name = trait.group_code + "_" + str(trait.name) trait.confidential = 0 if trait.pre_publication_description and not trait.pubmed_id: trait.confidential = 1 description = trait.post_publication_description # If the dataset is confidential and the user has access to confidential # phenotype traits, then display the pre-publication description instead # of the post-publication description trait.description_display = "N/A" trait.abbreviation = "N/A" if not trait.pubmed_id: if trait.pre_publication_abbreviation: trait.abbreviation = trait.pre_publication_abbreviation if trait.pre_publication_description: trait.description_display = trait.pre_publication_description else: if trait.post_publication_abbreviation: trait.abbreviation = trait.post_publication_abbreviation if description: trait.description_display = description.strip() if not trait.year.isdigit(): trait.pubmed_text = "N/A" else: trait.pubmed_text = trait.year if trait.pubmed_id: trait.pubmed_link = webqtlConfig.PUBMEDLINK_URL % trait.pubmed_id if dataset.type == 'ProbeSet' and dataset.group: description_string = trait.description target_string = trait.probe_target_description if str(description_string or "") != "" and description_string != 'None': description_display = description_string else: description_display = trait.symbol if (str(description_display or "") != "" and description_display != 'N/A' and str(target_string or "") != "" and target_string != 'None'): description_display = description_display + '; ' + target_string.strip() # Save it for the jinja2 template trait.description_display = description_display trait.location_repr = 'N/A' if trait.chr and trait.mb: trait.location_repr = 'Chr%s: %.6f' % ( trait.chr, float(trait.mb)) elif dataset.type == "Geno": trait.location_repr = 'N/A' if trait.chr and trait.mb: trait.location_repr = 'Chr%s: %.6f' % ( trait.chr, float(trait.mb)) if get_qtl_info: # LRS and its location trait.LRS_score_repr = "N/A" trait.LRS_location_repr = "N/A" trait.locus = trait.locus_chr = trait.locus_mb = trait.lrs = trait.pvalue = trait.additive = "" if dataset.type == 'ProbeSet' and not trait.cellid: trait.mean = "" cursor.execute( "SELECT ProbeSetXRef.Locus, ProbeSetXRef.LRS, " "ProbeSetXRef.pValue, ProbeSetXRef.mean, " "ProbeSetXRef.additive FROM ProbeSetXRef, " "ProbeSet WHERE " "ProbeSetXRef.ProbeSetId = ProbeSet.Id " "AND ProbeSet.Name = %s AND " "ProbeSetXRef.ProbeSetFreezeId = %s", (trait.name, dataset.id,) ) trait_qtl = cursor.fetchone() if any(trait_qtl): trait.locus, trait.lrs, trait.pvalue, trait.mean, trait.additive = trait_qtl if trait.locus: cursor.execute( "SELECT Geno.Chr, Geno.Mb FROM " "Geno, Species WHERE " "Species.Name = %s AND " "Geno.Name = %s AND " "Geno.SpeciesId = Species.Id", (dataset.group.species, trait.locus,) ) if result := cursor.fetchone() : trait.locus_chr = result[0] trait.locus_mb = result[1] else: trait.locus_chr = trait.locus_mb = "" else: trait.locus = trait.locus_chr = trait.locus_mb = trait.additive = "" if dataset.type == 'Publish': cursor.execute( "SELECT PublishXRef.Locus, PublishXRef.LRS, " "PublishXRef.additive FROM " "PublishXRef, PublishFreeze WHERE " "PublishXRef.Id = %s AND " "PublishXRef.InbredSetId = PublishFreeze.InbredSetId " "AND PublishFreeze.Id = %s", (trait.name, dataset.id,) ) if trait_qtl := cursor.fetchone(): trait.locus, trait.lrs, trait.additive = trait_qtl if trait.locus: cursor.execute( "SELECT Geno.Chr, Geno.Mb FROM Geno, " "Species WHERE Species.Name = %s " "AND Geno.Name = %s AND " "Geno.SpeciesId = Species.Id", (dataset.group.species, trait.locus,) ) if result := cursor.fetchone(): trait.locus_chr = result[0] trait.locus_mb = result[1] else: trait.locus = trait.locus_chr = trait.locus_mb = trait.additive = "" else: trait.locus = trait.locus_chr = trait.locus_mb = trait.additive = "" else: trait.locus = trait.lrs = trait.additive = "" if (dataset.type == 'Publish' or dataset.type == "ProbeSet"): if str(trait.locus_chr or "") != "" and str(trait.locus_mb or "") != "": trait.LRS_location_repr = LRS_location_repr = 'Chr%s: %.6f' % ( trait.locus_chr, float(trait.locus_mb)) if str(trait.lrs or "") != "": trait.LRS_score_repr = LRS_score_repr = '%3.1f' % trait.lrs else: raise KeyError( f"{repr(trait.name)} information is not found in the database " f"for dataset '{dataset.name}' with id '{dataset.id}'.") return trait