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
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
|
"""
This module deals with partial correlations.
It is an attempt to migrate over the partial correlations feature from
GeneNetwork1.
"""
from functools import reduce
from typing import Any, Tuple, Sequence
from scipy.stats import pearsonr, spearmanr
def control_samples(controls: Sequence[dict], sampleslist: Sequence[str]):
"""
Fetches data for the control traits.
This migrates `web/webqtl/correlation/correlationFunction.controlStrain` in
GN1, with a few modifications to the arguments passed in.
PARAMETERS:
controls: A map of sample names to trait data. Equivalent to the `cvals`
value in the corresponding source function in GN1.
sampleslist: A list of samples. Equivalent to `strainlst` in the
corresponding source function in GN1
"""
def __process_control__(trait_data):
def __process_sample__(acc, sample):
if sample in trait_data["data"].keys():
sample_item = trait_data["data"][sample]
val = sample_item["value"]
if val is not None:
return (
acc[0] + (sample,),
acc[1] + (val,),
acc[2] + (sample_item["variance"],))
return acc
return reduce(
__process_sample__, sampleslist, (tuple(), tuple(), tuple()))
return reduce(
lambda acc, item: (
acc[0] + (item[0],),
acc[1] + (item[1],),
acc[2] + (item[2],),
acc[3] + (len(item[0]),),
),
[__process_control__(trait_data) for trait_data in controls],
(tuple(), tuple(), tuple(), tuple()))
def dictify_by_samples(samples_vals_vars: Sequence[Sequence]) -> Sequence[dict]:
"""
Build a sequence of dictionaries from a sequence of separate sequences of
samples, values and variances.
This is a partial migration of
`web.webqtl.correlation.correlationFunction.fixStrains` function in GN1.
This implementation extracts code that will find common use, and that will
find use in more than one place.
"""
return tuple(
{
sample: {"sample_name": sample, "value": val, "variance": var}
for sample, val, var in zip(*trait_line)
} for trait_line in zip(*(samples_vals_vars[0:3])))
def fix_samples(primary_trait: dict, control_traits: Sequence[dict]) -> Sequence[Sequence[Any]]:
"""
Corrects sample_names, values and variance such that they all contain only
those samples that are common to the reference trait and all control traits.
This is a partial migration of the
`web.webqtl.correlation.correlationFunction.fixStrain` function in GN1.
"""
primary_samples = tuple(
present[0] for present in
((sample, all(sample in control.keys() for control in control_traits))
for sample in primary_trait.keys())
if present[1])
control_vals_vars: tuple = reduce(
lambda acc, x: (acc[0] + (x[0],), acc[1] + (x[1],)),
((item["value"], item["variance"])
for sublist in [tuple(control.values()) for control in control_traits]
for item in sublist),
(tuple(), tuple()))
return (
primary_samples,
tuple(primary_trait[sample]["value"] for sample in primary_samples),
control_vals_vars[0],
tuple(primary_trait[sample]["variance"] for sample in primary_samples),
control_vals_vars[1])
def find_identical_traits(
primary_name: str, primary_value: float, control_names: Tuple[str, ...],
control_values: Tuple[float, ...]) -> Tuple[str, ...]:
"""
Find traits that have the same value when the values are considered to
3 decimal places.
This is a migration of the
`web.webqtl.correlation.correlationFunction.findIdenticalTraits` function in
GN1.
"""
def __merge_identicals__(
acc: Tuple[str, ...],
ident: Tuple[str, Tuple[str, ...]]) -> Tuple[str, ...]:
return acc + ident[1]
def __dictify_controls__(acc, control_item):
ckey = "{:.3f}".format(control_item[0])
return {**acc, ckey: acc.get(ckey, tuple()) + (control_item[1],)}
return (reduce(## for identical control traits
__merge_identicals__,
(item for item in reduce(# type: ignore[var-annotated]
__dictify_controls__, zip(control_values, control_names),
{}).items() if len(item[1]) > 1),
tuple())
or
reduce(## If no identical control traits, try primary and controls
__merge_identicals__,
(item for item in reduce(# type: ignore[var-annotated]
__dictify_controls__,
zip((primary_value,) + control_values,
(primary_name,) + control_names), {}).items()
if len(item[1]) > 1),
tuple()))
def tissue_correlation(
primary_trait_values: Tuple[float, ...],
target_trait_values: Tuple[float, ...],
method: str) -> Tuple[float, float]:
"""
Compute the correlation between the primary trait values, and the values of
a single target value.
This migrates the `cal_tissue_corr` function embedded in the larger
`web.webqtl.correlation.correlationFunction.batchCalTissueCorr` function in
GeneNetwork1.
"""
def spearman_corr(*args):
result = spearmanr(*args)
return (result.correlation, result.pvalue)
method_fns = {"pearson": pearsonr, "spearman": spearman_corr}
assert len(primary_trait_values) == len(target_trait_values), (
"The lengths of the `primary_trait_values` and `target_trait_values` "
"must be equal")
assert method in method_fns.keys(), (
"Method must be one of: {}".format(",".join(method_fns.keys())))
return method_fns[method](primary_trait_values, target_trait_values)
def batch_computed_tissue_correlation(
primary_trait_values: Tuple[float, ...], target_traits_dict: dict,
method: str) -> Tuple[dict, dict]:
"""
This is a migration of the
`web.webqtl.correlation.correlationFunction.batchCalTissueCorr` function in
GeneNetwork1
"""
def __corr__(acc, target):
corr = tissue_correlation(primary_trait_values, target[1], method)
return ({**acc[0], target[0]: corr[0]}, {**acc[0], target[1]: corr[1]})
return reduce(__corr__, target_traits_dict.items(), ({}, {}))
def correlations_of_all_tissue_traits(
primary_trait_symbol_value_dict: dict, symbol_value_dict: dict,
method: str) -> Tuple[dict, dict]:
"""
Computes and returns the correlation of all tissue traits.
This is a migration of the
`web.webqtl.correlation.correlationFunction.calculateCorrOfAllTissueTrait`
function in GeneNetwork1.
"""
primary_trait_values = tuple(primary_trait_symbol_value_dict.values())[0]
return batch_computed_tissue_correlation(
primary_trait_values, symbol_value_dict, method)
|
