-
Notifications
You must be signed in to change notification settings - Fork 2
/
exporter-all.py
executable file
·195 lines (162 loc) · 8.51 KB
/
exporter-all.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
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
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
#!/usr/bin/python3
# vim:ts=4:sw=4:tw=0:sts=4:et
import pprint
import re
import argparse
import logging as log
from builtins import str, isinstance, len, set, int
from typing import List
import json
import pandas as pd
from directory import Directory
from orphacodes import OrphaCodes
from icd10codeshelper import ICD10CodesHelper
import pddfutils
cachesList = ['directory', 'emails', 'geocoding', 'URLs']
pp = pprint.PrettyPrinter(indent=4)
class ExtendAction(argparse.Action):
def __call__(self, parser, namespace, values, option_string=None):
from builtins import getattr, setattr
items = getattr(namespace, self.dest) or []
items.extend(values)
setattr(namespace, self.dest, items)
parser = argparse.ArgumentParser()
parser.register('action', 'extend', ExtendAction)
parser.add_argument('-v', '--verbose', dest='verbose', action='store_true',
help='verbose information on progress of the data checks')
parser.add_argument('-d', '--debug', dest='debug', action='store_true',
help='debug information on progress of the data checks')
parser.add_argument('-X', '--output-XLSX', dest='outputXLSX', nargs=1,
help='output of results into an XLSX with filename provided as parameter')
parser.add_argument('--output-XLSX-withdrawn', dest='outputXLSXwithdrawn', nargs=1,
help='output withdrawn biobanks and collections into an XLSX with filename provided as parameter')
parser.add_argument('-N', '--output-no-stdout', dest='nostdout', action='store_true',
help='no output of results into stdout (default: enabled)')
parser.add_argument('--purge-all-caches', dest='purgeCaches', action='store_const', const=cachesList,
help='disable all long remote checks (email address testing, geocoding, URLs')
parser.add_argument('--purge-cache', dest='purgeCaches', nargs='+', action='extend', choices=cachesList,
help='disable particular long remote checks')
parser.set_defaults(disableChecksRemote=[], disablePlugins=[], purgeCaches=[])
args = parser.parse_args()
if args.debug:
log.basicConfig(format="%(levelname)s: %(message)s", level=log.DEBUG)
elif args.verbose:
log.basicConfig(format="%(levelname)s: %(message)s", level=log.INFO)
else:
log.basicConfig(format="%(levelname)s: %(message)s")
# Main code
dir = Directory(purgeCaches=args.purgeCaches, debug=args.debug, pp=pp)
log.info('Total biobanks: ' + str(dir.getBiobanksCount()))
log.info('Total collections: ' + str(dir.getCollectionsCount()))
allCollections = []
withdrawnCollections = []
allBiobanks = set()
withdrawnBiobanks = set()
allCollectionSamplesExplicit = 0
allCollectionDonorsExplicit = 0
allCollectionSamplesIncOoM = 0
allCountries = set()
for collection in dir.getCollections():
log.debug("Analyzing collection " + collection['id'])
biobankId = dir.getCollectionBiobankId(collection['id'])
biobank = dir.getBiobankById(biobankId)
collection_withdrawn = False
if 'withdrawn' in collection and collection['withdrawn']:
withdrawnCollections.append(collection)
log.debug("Detected a withdrawn collection: " + collection['id'])
collection_withdrawn = True
if 'withdrawn' in biobank and biobank['withdrawn']:
withdrawnBiobanks.add(biobankId)
if not collection_withdrawn:
log.debug("Detected a withdrawn collection " + collection['id'] + " because a withdrawn biobank: " + biobankId)
collection_withdrawn = True
if collection_withdrawn:
continue
biobank_capabilities = []
if 'capabilities' in biobank:
for c in biobank['capabilities']:
biobank_capabilities.append(c['id'])
biobank_covid = []
if 'covid19biobank' in biobank:
for c in biobank['covid19biobank']:
biobank_covid.append(c['id'])
biobank_networks = []
if 'network' in biobank:
for n in biobank['network']:
biobank_networks.append(n['id'])
OoM = collection['order_of_magnitude']['id']
materials = []
if 'materials' in collection:
for m in collection['materials']:
materials.append(m['id'])
data_categories = []
if 'data_categories' in collection:
for c in collection['data_categories']:
data_categories.append(c['id'])
types = []
if 'type' in collection:
for t in collection['type']:
types.append(t['id'])
log.debug("Types: " + str(types))
if biobank['country']['id'] != 'EU':
allCountries.add(biobank['country']['id'])
allCollections.append(collection)
allBiobanks.add(biobankId)
#if 'size' in collection and isinstance(collection['size'], int) and dir.isTopLevelCollection(collection['id']):
if dir.isCountableCollection(collection['id'], 'size'):
allCollectionSamplesExplicit += collection['size']
allCollectionSamplesIncOoM += collection['size']
else:
# Intentionally, the lower bound of the OoM interval is taken - the size of the collection should be in the range of 10**OoM to 10**(OoM+1) - and hence using 10**OoM is a bound nobody can question unless there is a bug in the underlying data. Historically, we used also 0.3*10**(OoM+1).
# note that OoM is only counted for top-level collections to avoid double counting - because OoM is mandatory parameter, any child collection has a parent which has OoM filled in
if dir.isTopLevelCollection(collection['id']):
allCollectionSamplesIncOoM += int(10 ** OoM)
#if 'number_of_donors' in collection and isinstance(collection['number_of_donors'], int) and dir.isTopLevelCollection(collection['id']):
if dir.isCountableCollection(collection['id'], 'number_of_donors'):
allCollectionDonorsExplicit += collection['number_of_donors']
for biobank in dir.getBiobanks():
biobankId = biobank['id']
log.debug("Analyzing biobank " + biobankId)
if not biobankId in allBiobanks and not biobankId in withdrawnBiobanks:
log.info(" Biobank without any collection identified: " + biobankId)
if 'withdrawn' in biobank and biobank['withdrawn']:
withdrawnBiobanks.add(biobankId)
log.debug("Detected a withdrawn biobank without any collection " + biobankId)
else:
allBiobanks.add(biobankId)
pd_allCollections = pd.DataFrame(allCollections)
pd_withdrawnCollections = pd.DataFrame(withdrawnCollections)
pd_allBiobanks = pd.DataFrame([dir.getBiobankById(biobankId) for biobankId in allBiobanks])
pd_withdrawnBiobanks = pd.DataFrame([dir.getBiobankById(biobankId) for biobankId in withdrawnBiobanks])
def printCollectionStdout(collectionList: List):
for collection in collectionList:
biobankId = dir.getCollectionBiobankId(collection['id'])
biobank = dir.getBiobankById(biobankId)
log.info(" Collection: " + collection['id'] + " - " + collection['name'] + ". Parent biobank: " + biobankId + " - " + biobank['name'])
log.info("\n\n")
if not args.nostdout:
printCollectionStdout(allCollections)
print("Totals:")
print("- total of biobanks: %d" % (len(allBiobanks)))
print("- total of withdrawn biobanks: %d" % (len(withdrawnBiobanks)))
print("- total of collections with existing samples: %d" % (len(allCollections)))
print("- total of countries: %d" % ( len(allCountries)))
print("Estimated totals:")
print("- total of samples/donors advertised explicitly in all-relevant collections: %d / %d" % (
allCollectionSamplesExplicit, allCollectionDonorsExplicit))
print("- total of samples advertised in all-relevant collections including OoM estimates: %d" % (
allCollectionSamplesIncOoM))
for df in (pd_allCollections, pd_withdrawnCollections):
pddfutils.tidyCollectionDf(df)
for df in (pd_allBiobanks, pd_withdrawnBiobanks):
pddfutils.tidyBiobankDf(df)
def outputExcelBiobanksCollections(filename : str, dfBiobanks : pd.DataFrame, biobanksLabel : str, dfCollections : pd.DataFrame, collectionsLabel : str):
log.info("Outputting warnings in Excel file " + filename)
writer = pd.ExcelWriter(filename, engine='xlsxwriter')
dfBiobanks.to_excel(writer, sheet_name=biobanksLabel)
dfCollections.to_excel(writer, sheet_name=collectionsLabel)
writer.save()
if args.outputXLSX is not None:
outputExcelBiobanksCollections(args.outputXLSX[0], pd_allBiobanks, "Biobanks", pd_allCollections, "Collections")
if args.outputXLSXwithdrawn is not None:
outputExcelBiobanksCollections(args.outputXLSXwithdrawn[0], pd_withdrawnBiobanks, "Withdrawn biobanks", pd_withdrawnCollections, "Withdrawn collections")