1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
|
"""Script to run partial correlations"""
import json
import traceback
from argparse import ArgumentParser
from gn3.db_utils import database_connector
from gn3.responses.pcorrs_responses import OutputEncoder
from gn3.computations.partial_correlations import (
partial_correlations_with_target_db,
partial_correlations_with_target_traits)
def cleanup_string(the_str):
"Remove tab, newline and carriage return characters."
return the_str.strip('"\t\n\r ')
def process_common_args(args):
"Process the common CLI arguments to a form usable by the functions"
return {
"primary_trait_name": cleanup_string(args.primary_trait),
"control_trait_names": tuple(
cleanup_string(args.control_traits).split(",")),
"method": cleanup_string(args.method)
}
def process_trait_args(args):
"""Process arguments to a form usable by the
`partial_correlations_with_target_traits` function."""
return {
**process_common_args(args),
"target_trait_names": tuple(
cleanup_string(args.target_traits).split(","))
}
def process_db_args(args):
"""Process arguments for the `partial_correlations_with_target_db`
function."""
return {
**process_common_args(args),
"target_db_name": cleanup_string(args.target_database),
"criteria": args.criteria
}
def pcorrs_against_traits(dbconn, args):
"""Run partial correlations agaist selected traits."""
return partial_correlations_with_target_traits(
dbconn, **process_trait_args(args))
def pcorrs_against_db(dbconn, args):
"""Run partial correlations agaist the entire dataset provided."""
return partial_correlations_with_target_db(dbconn, **process_db_args(args))
def run_pcorrs(dbconn, args):
"""Run the selected partial correlations function."""
try:
return args.func(dbconn, args)
except Exception as exc: # pylint: disable=[broad-except,unused-variable]
return {
"status": "exception",
"message": traceback.format_exc()
}
def against_traits_parser(parent_parser):
"""Parser for command to run partial correlations against selected traits"""
parser = parent_parser.add_parser(
"against-traits",
help="Run partial correlations against a select list of traits")
parser.add_argument(
"target_traits",
help=(
"The target traits to run the partial correlations against. "
"This is a comma-separated list of traits' fullnames, in the "
"format <DATASET-NAME>::<TRAIT-NAME> e.g. "
"UCLA_BXDBXH_CARTILAGE_V2::ILM103710672"),
type=str)
parser.set_defaults(func=pcorrs_against_traits)
return parent_parser
def against_db_parser(parent_parser):
"""Parser for command to run partial correlations against entire dataset"""
parser = parent_parser.add_parser(
"against-db",
help="Run partial correlations against an entire dataset")
parser.add_argument(
"target_database",
help="The target database to run the partial correlations against",
type=str)
parser.add_argument(
"--criteria",
help="Number of results to return",
type=int, default=500)
parser.set_defaults(func=pcorrs_against_db)
return parent_parser
def process_cli_arguments():
"""Top level parser"""
parser = ArgumentParser()
parser.add_argument(
"primary_trait",
help="The primary trait's full name",
type=str)
parser.add_argument(
"control_traits",
help="A comma-separated list of traits' full names",
type=str)
parser.add_argument(
"method",
help="The correlation method to use",
type=str)
against_db_parser(against_traits_parser(
parser.add_subparsers(
title="subcommands",
description="valid subcommands",
required=True)))
return parser.parse_args()
def main():
"""Entry point for the script"""
args = process_cli_arguments()
with database_connector() as conn:
print(json.dumps(run_pcorrs(conn, args), cls=OutputEncoder))
if __name__ == "__main__":
main()
|
