import json
import math
import requests
from functools import reduce
from typing import Union, Tuple
from urllib.parse import urljoin
from flask import (
flash,
request,
url_for,
redirect,
current_app,
render_template)
from gn2.wqflask import app
from gn2.utility.tools import get_setting, GN_SERVER_URL
from gn2.wqflask.database import database_connection
from gn3.db.partial_correlations import traits_info
def publish_target_databases(conn, groups, threshold):
query = (
"SELECT PublishFreeze.FullName,PublishFreeze.Name "
"FROM PublishFreeze, InbredSet "
"WHERE PublishFreeze.InbredSetId = InbredSet.Id "
f"AND InbredSet.Name IN ({', '.join(['%s'] * len(groups))}) "
"AND PublishFreeze.public > %s")
with conn.cursor() as cursor:
cursor.execute(query, tuple(groups) + (threshold,))
res = cursor.fetchall()
if res:
return tuple(
dict(zip(("description", "value"), row)) for row in res)
return tuple()
def geno_target_databases(conn, groups, threshold):
query = (
"SELECT GenoFreeze.FullName,GenoFreeze.Name "
"FROM GenoFreeze, InbredSet "
"WHERE GenoFreeze.InbredSetId = InbredSet.Id "
f"AND InbredSet.Name IN ({', '.join(['%s'] * len(groups))}) "
"AND GenoFreeze.public > %s")
with conn.cursor() as cursor:
cursor.execute(query, tuple(groups) + (threshold,))
res = cursor.fetchall()
if res:
return tuple(
dict(zip(("description", "value"), row)) for row in res)
return tuple()
def probeset_target_databases(conn, groups, threshold):
query1 = "SELECT Id, Name FROM Tissue order by Name"
with conn.cursor() as cursor:
cursor.execute(query1)
tissue_res = cursor.fetchall()
if tissue_res:
tissue_ids = tuple(row[0] for row in tissue_res)
groups_clauses = ["InbredSet.Name like %s"] * len(groups)
query2 = (
"SELECT ProbeFreeze.TissueId, ProbeSetFreeze.FullName, "
"ProbeSetFreeze.Name "
"FROM ProbeSetFreeze, ProbeFreeze, InbredSet "
"WHERE ProbeSetFreeze.ProbeFreezeId = ProbeFreeze.Id "
"AND ProbeFreeze.TissueId IN "
f"({', '.join(['%s'] * len(tissue_ids))}) "
"AND ProbeSetFreeze.public > %s "
"AND ProbeFreeze.InbredSetId = InbredSet.Id "
f"AND ({' OR '.join(groups_clauses)}) "
"ORDER BY ProbeSetFreeze.CreateTime desc, ProbeSetFreeze.AvgId")
cursor.execute(query2, tissue_ids + (threshold,) + tuple(groups))
db_res = cursor.fetchall()
if db_res:
databases = tuple(
dict(zip(("tissue_id", "description", "value"), row))
for row in db_res)
return tuple(
{tissue_name: tuple(
{
"value": item["value"],
"description": item["description"]
} for item in databases
if item["tissue_id"] == tissue_id)}
for tissue_id, tissue_name in tissue_res)
return tuple()
def target_databases(conn, traits, threshold):
"""
Retrieves the names of possible target databases from the database.
"""
trait_info = traits_info(
conn, threshold,
tuple(f"{trait['dataset']}::{trait['trait_name']}" for trait in traits))
groups = tuple(set(row["db"]["group"] for row in trait_info))
return (
publish_target_databases(conn, groups, threshold) +
geno_target_databases(conn, groups, threshold) +
probeset_target_databases(conn, groups, threshold))
def primary_error(args):
if len(args["primary_trait"]) == 0 or len(args["primary_trait"]) > 1:
return {
**args,
"errors": (args.get("errors", tuple()) +
("You must provide one, and only one primary trait",))}
return args
def controls_error(args):
if len(args["control_traits"]) == 0 or len(args["control_traits"]) > 3:
return {
**args,
"errors": (
args.get("errors", tuple()) +
(("You must provide at least one control trait, and a maximum "
"of three control traits"),))}
return args
def target_traits_error(args, with_target_traits):
target_traits_present = (
(args.get("target_traits") is not None) and
(len(args["target_traits"]) > 0))
if with_target_traits and not target_traits_present:
return {
**args,
"errors": (
args.get("errors", tuple()) +
(("You must provide at least one target trait"),))}
return args
def target_db_error(args, with_target_db: bool):
if with_target_db and not args["target_db"]:
return {
**args,
"errors": (
args.get("errors", tuple()) +
("The target database must be provided",))}
return args
def method_error(args):
methods = (
"pearson's r", "spearman's rho",
"genetic correlation, pearson's r",
"genetic correlation, spearman's rho",
"sgo literature correlation",
"tissue correlation, pearson's r",
"tissue correlation, spearman's rho")
if not args["method"] or args["method"].lower() not in methods:
return {
**args,
"errors": (
args.get("errors", tuple()) +
("Invalid correlation method provided",))}
return args
def criteria_error(args):
try:
int(args.get("criteria", "invalid"))
return args
except ValueError:
return {
**args,
"errors": (
args.get("errors", tuple()) +
("Invalid return number provided",))}
def errors(args, with_target_db: bool):
return {
**criteria_error(
method_error(
target_traits_error(
target_db_error(
controls_error(primary_error(args)),
with_target_db),
not with_target_db))),
"with_target_db": with_target_db
}
def __classify_args(acc, item):
if item[1].startswith("primary_"):
return {
**acc,
"primary_trait": (acc.get("primary_trait", tuple()) + (item,))}
if item[1].startswith("controls_"):
return {**acc, "control_traits": (acc.get("control_traits", tuple()) + (item,))}
if item[1].startswith("targets_"):
return {**acc, "target_traits": (acc.get("target_traits", tuple()) + (item,))}
if item[0] == "target_db":
return {**acc, "target_db": item[1]}
if item[0] == "method":
return {**acc, "method": item[1]}
if item[0] == "criteria":
return {**acc, "criteria": item[1]}
return acc
def __build_args(raw_form, traits):
args = reduce(__classify_args, raw_form.items(), {})
return {
**args,
"primary_trait": [
item for item in traits if item["trait_name"] in
(name[1][8:] for name in args["primary_trait"])],
"control_traits": [
item for item in traits if item["trait_name"] in
(name[1][9:] for name in args["control_traits"])],
"target_traits": [
item for item in traits if item["trait_name"] in
(name[1][8:] for name in args.get("target_traits", tuple()))]
}
def parse_trait(trait_str):
return dict(zip(
("trait_name", "dataset", "description", "symbol", "location", "mean",
"lrs", "lrs_location"),
trait_str.strip().split("|||")))
def response_error_message(response):
error_messages = {
404: ("We could not connect to the API server at this time. "
"Try again later."),
500: ("The API server experienced a problem. We will be working on a "
"fix. Please try again later.")
}
return error_messages.get(
response.status_code,
"General API server error!!")
def render_error(error_message, command_id = None):
return render_template(
"partial_correlations/pcorrs_error.html",
message = error_message,
command_id = command_id)
def __format_number(num):
if num is None or math.isnan(num):
return ""
if abs(num) <= 1.04E-4:
return f"{num:.2e}"
return f"{num:.5f}"
def handle_200_response(response):
if response.get("queued", False):
return redirect(
url_for(
"poll_partial_correlation_results",
command_id=response["results"]),
code=303)
if response["status"] == "success":
return render_template(
"partial_correlations/pcorrs_results_with_target_traits.html",
primary = response["results"]["results"]["primary_trait"],
controls = response["results"]["results"]["control_traits"],
pcorrs = sorted(
response["results"]["results"]["correlations"],
key = lambda item: item["partial_corr_p_value"]),
method = response["results"]["results"]["method"],
enumerate = enumerate,
format_number = __format_number)
return render_error(response["results"])
def handle_response(response):
if response.status_code != 200:
return render_template(
"partial_correlations/pcorrs_error.html",
message = response_error_message(response))
return handle_200_response(response.json())
@app.route("/partial_correlations", methods=["POST"])
def partial_correlations():
form = request.form
traits = tuple(
parse_trait(trait) for trait in
form.get("trait_list").split(";;;"))
submit = form.get("submit")
if submit in ("with_target_pearsons", "with_target_spearmans"):
method = "pearsons" if "pearsons" in submit else "spearmans"
args = {
**errors(__build_args(form, traits), with_target_db=False),
"method": method
}
if len(args.get("errors", [])) == 0:
post_data = {
**args,
"primary_trait": args["primary_trait"][0],
"with_target_db": args["with_target_db"]
}
return handle_response(requests.post(
url=urljoin(GN_SERVER_URL, "correlation/partial"),
json=post_data))
for error in args["errors"]:
flash(error, "alert-danger")
if submit == "Run Partial Correlations":
args = errors(__build_args(form, traits), with_target_db=True)
if len(args.get("errors", [])) == 0:
post_data = {
**args,
"primary_trait": args["primary_trait"][0],
"with_target_db": args["with_target_db"]
}
return handle_response(requests.post(
url=urljoin(GN_SERVER_URL, "correlation/partial"),
json=post_data))
for error in args["errors"]:
flash(error, "alert-danger")
with database_connection(get_setting("SQL_URI")) as conn:
target_dbs = target_databases(conn, traits, threshold=0)
return render_template(
"partial_correlations/pcorrs_select_operations.html",
trait_list_str=form.get("trait_list"),
traits=traits,
target_dbs=target_dbs)
def process_pcorrs_command_output(result):
if result["status"] == "success":
if result["results"]["dataset_type"] == "NOT SET YET":
return render_template(
"partial_correlations/pcorrs_results_with_target_traits.html",
primary = result["results"]["primary_trait"],
controls = result["results"]["control_traits"],
pcorrs = sorted(
result["results"]["correlations"],
key = lambda item: item["partial_corr_p_value"]),
method = result["results"]["method"],
enumerate = enumerate,
format_number = __format_number)
return render_template(
"partial_correlations/pcorrs_results_presentation.html",
primary=result["results"]["primary_trait"],
controls=result["results"]["control_traits"],
correlations=result["results"]["correlations"],
dataset_type=result["results"]["dataset_type"],
method=result["results"]["method"],
enumerate = enumerate,
format_number=__format_number)
if result["status"] == "error":
return render_error(
f"({result['error_type']}: {result['message']})")
@app.route("/partial_correlations/<command_id>", methods=["GET"])
def poll_partial_correlation_results(command_id):
response = requests.get(
url=urljoin(GN_SERVER_URL, f"async_commands/state/{command_id}"))
if response.status_code == 200:
data = response.json()
raw_result = data["result"]
result = {"status": "computing"}
if raw_result:
result = json.loads(raw_result)
if result["status"].lower() in ("error", "exception"):
return render_error(
"We messed up, and the computation failed due to a system "
"error.",
command_id)
if data["status"] == "success":
return process_pcorrs_command_output(json.loads(data["result"]))
return render_template(
"partial_correlations/pcorrs_poll_results.html",
command_id = command_id)
return render_error(
"We messed up, and the computation failed due to a system "
"error.",
command_id)
