-
Notifications
You must be signed in to change notification settings - Fork 0
/
PlotFeatures.py
112 lines (100 loc) · 4.12 KB
/
PlotFeatures.py
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
from bs.operations import base
from bbcflib.bFlatMajor.numeric import feature_matrix
from bbcflib.bFlatMajor.figure import heatmap, lineplot
from bbcflib.btrack import track
from numpy import vstack, concatenate
prom_up_def = 1000
prom_down_def = 100
plot_types = [(0, 'heatmap'), (1, 'average lineplot'), (2, 'mosaic plot')]
max_pages = 200
import tw2.forms as twf
import tw2.core as twc
class PlotFeaturesForm(base.BaseForm):
signal = base.MultipleFileUpload(label='Signal: ',
help_text='Select signal files (e.g. bedgraph)',
validator=twf.FileValidator(required=True))
features = twf.FileField(label='Features: ',
help_text='Select a feature file (e.g. bed)',
validator=twf.FileValidator())
mode = twf.SingleSelectField(label='Plot type: ',
options=plot_types,
prompt_text=None,
validator=twc.Validator(required=True))
upstream = twf.TextField(label='Upstream flank: ',
validator=twc.IntValidator(),
value=prom_up_def,
help_text='Size of upstream flank in bp')
downstream = twf.TextField(label='Downstream flank: ',
validator=twc.IntValidator(),
value=prom_down_def,
help_text='Size of downstream flank in bp')
submit = twf.SubmitButton(id="submit", value="Plot")
meta = {'version': "1.0.0",
'author': "BBCF",
'contact': "[email protected]"}
in_parameters = [{'id': 'signal', 'type': 'track', 'multiple': True, 'required': True},
{'id': 'features', 'type': 'track'},
{'id': 'mode', 'type': 'list', 'required': True},
{'id': 'upstream', 'type': 'int'},
{'id': 'downstream', 'type': 'int'}
]
out_parameters = [{'id': 'plot_features', 'type': 'pdf'}]
class PlotFeaturesPlugin(base.OperationPlugin):
info = {
'title': 'Plot signal',
'description': 'Plot signal from a selection of regions',
'path': ['Plot', 'Plot features'],
'output': PlotFeaturesForm,
'in': in_parameters,
'out': out_parameters,
'meta': meta,
}
def __call__(self, **kw):
features = track(kw.get('features'))
signal = [track(sig, chrmeta=features.chrmeta) for sig in kw.get('signal', [])]
labels = None
data = None
for chrom in features.chrmeta:
_l, _d = feature_matrix([s.read(chrom) for s in signal],
features.read(chrom), segment=True)
if _d.size == 0:
continue
if data is None:
labels = _l
data = _d
else:
labels = concatenate((labels, _l))
data = vstack((data, _d))
pdf = self.temporary_path(fname='plot_features', ext='.pdf')
if data is None:
raise ValueError("No data")
kw['mode'] = int(kw.get('mode', 0))
if kw['mode'] == 0:
new = True
for n in range(data.shape[-1] - 1):
heatmap(data[:, :, n], output=pdf, new=new, last=False,
rows=labels, orderRows=True, orderCols=False)
new = False
heatmap(data[:, :, -1], output=pdf, new=new, last=True,
rows=labels, orderRows=True, orderCols=False)
elif kw['mode'] == 1:
X = range(data.shape[1])
Y = data.mean(axis=0)
lineplot(X, [Y[:, n] for n in range(data.shape[-1])],
output=pdf, new=True, last=True)
elif kw['mode'] == 2:
X = range(data.shape[1])
new = True
mfrow = [4, 3]
nplot = min(data.shape[0], max_pages * mfrow[0] * mfrow[1])
for reg in range(nplot - 1):
lineplot(X, [data[reg, :, n] for n in range(data.shape[-1])],
output=pdf, new=new, last=False, mfrow=mfrow)
new = False
mfrow = []
lineplot(X, [data[nplot - 1, :, n] for n in range(data.shape[-1])],
output=pdf, new=new, last=True)
else:
raise ValueError("Mode not implemented: %s" % kw['mode'])
self.new_file(pdf, 'plot_features')
return 1