path: root/wqflask/base/trait.py

import requests
import simplejson as json
from wqflask import 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.tools import GN2_BASE_URL, GN_PROXY_URL
from utility.redis_tools import get_redis_conn, get_resource_id

from flask import g, request, url_for

from wqflask.database import database_connection


Redis = get_redis_conn()


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(
                GN2_BASE_URL + "gn3/gene/aliases/" + self.symbol.upper())
            mouse_response = requests.get(
                GN2_BASE_URL + "gn3/gene/aliases/" + self.symbol.capitalize())
            other_response = requests.get(
                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


@app.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