aboutsummaryrefslogtreecommitdiff
diff options
context:
space:
mode:
-rw-r--r--workflows/pull-data/genbank/genbank.py138
1 files changed, 84 insertions, 54 deletions
diff --git a/workflows/pull-data/genbank/genbank.py b/workflows/pull-data/genbank/genbank.py
index 7ce3913..7383261 100644
--- a/workflows/pull-data/genbank/genbank.py
+++ b/workflows/pull-data/genbank/genbank.py
@@ -7,6 +7,46 @@ import xml.etree.ElementTree as ET
class GBError(Exception):
pass
+"""
+Example of an output JSON:
+
+{
+ "id": "placeholder",
+ "host": {
+ "host_species": "http://purl.obolibrary.org/obo/NCBITaxon_9606"
+ },
+ "sample": {
+ "sample_id": "MT890462.1",
+ "source_database_accession": [
+ "http://identifiers.org/insdc/MT890462.1#sequence"
+ ],
+ "collection_location": "http://www.wikidata.org/entity/Q649",
+ "collection_date": "2020-04-17",
+ "collecting_institution": "N.A.Kovtun Clinical Hospital 1 of Departament of President Affairs"
+ },
+ "virus": {
+ "virus_strain": "SARS-CoV-2/human/RUS/20200417_10/2020",
+ "virus_species": "http://purl.obolibrary.org/obo/NCBITaxon_2697049"
+ },
+ "technology": {
+ "assembly_method": "http://purl.obolibrary.org/obo/GENEPIO_0001628",
+ "alignment_protocol": "bowtie2 v. 2.3.4",
+ "sample_sequencing_technology": [
+ "http://purl.obolibrary.org/obo/OBI_0000759"
+ ]
+ },
+ "submitter": {
+ "authors": [
+ "Blagodatskikh,K.A."
+ ],
+ "submitter_name": [
+ "R&D"
+ ],
+ "submitter_address": "Pirogov Russian National Research Medical University, Ostrovityanova 1, Moscow 117997, Russia"
+ }
+}
+"""
+
def get_metadata(id, gb):
return True,None
@@ -37,15 +77,7 @@ if None:
for GBSeq in GBSet:
accession_version = GBSeq.find('GBSeq_accession-version').text
- GBSeq_sequence = GBSeq.find('GBSeq_sequence')
- if GBSeq_sequence is None:
- print(accession_version, ' - sequence not found')
- continue
-
try:
- # print(path_metadata_xxx_xml, accession_version)
-
- # A general default-empty yaml could be read from the definitive one
info_for_yaml_dict = {
'id': 'placeholder',
'host': {},
@@ -55,17 +87,15 @@ if None:
'submitter': {}
}
-
- info_for_yaml_dict['sample']['sample_id'] = accession_version
- info_for_yaml_dict['sample']['source_database_accession'] = ["http://identifiers.org/insdc/"+accession_version+"#sequence"] #accession is turned into resolvable URL/URI now
-
+ sample['sample_id'] = accession_version
+ sample['source_database_accession'] = ["http://identifiers.org/insdc/"+accession_version+"#sequence"] #accession is turned into resolvable URL/URI now
# submitter info
GBSeq_references = GBSeq.find('GBSeq_references')
if GBSeq_references is not None:
author_list = ["{}".format(x.text) for x in GBSeq_references.iter('GBAuthor')]
if len(author_list) > 0:
- info_for_yaml_dict['submitter']['authors'] = author_list
+ submitter['authors'] = author_list
GBReference = GBSeq_references.find('GBReference')
if GBReference is not None:
@@ -73,13 +103,13 @@ if None:
if GBReference_journal is not None and GBReference_journal.text != 'Unpublished':
if 'Submitted' in GBReference_journal.text:
- info_for_yaml_dict['submitter']['submitter_name'] = ["{}".format(GBReference_journal.text.split(') ')[1].split(',')[0].strip())]
- info_for_yaml_dict['submitter']['submitter_address'] = ','.join(GBReference_journal.text.split(') ')[1].split(',')[1:]).strip()
+ submitter['submitter_name'] = ["{}".format(GBReference_journal.text.split(') ')[1].split(',')[0].strip())]
+ submitter['submitter_address'] = ','.join(GBReference_journal.text.split(') ')[1].split(',')[1:]).strip()
else:
- info_for_yaml_dict['submitter']['additional_submitter_information'] = GBReference_journal.text
+ submitter['additional_submitter_information'] = GBReference_journal.text
# This script download and prepare data and metadata for assemblies samples
- info_for_yaml_dict['technology']['assembly_method'] = 'http://purl.obolibrary.org/obo/GENEPIO_0001628'
+ technology['assembly_method'] = 'http://purl.obolibrary.org/obo/GENEPIO_0001628'
GBSeq_comment = GBSeq.find('GBSeq_comment')
if GBSeq_comment is not None and 'Assembly-Data' in GBSeq_comment.text:
@@ -99,7 +129,7 @@ if None:
if field_in_yaml == 'sequencing_coverage':
# A regular expression would be better!
try:
- info_for_yaml_dict['technology'][field_in_yaml] = [
+ technology[field_in_yaml] = [
float(tech_info_to_parse.replace('(average)', '').replace("reads/nt", '').
replace('(average for 6 sequences)', '').replace(',', '.').strip(' xX>'))
]
@@ -117,9 +147,9 @@ if None:
missing_value_list.append('\t'.join([accession_version, 'sample_sequencing_technology', seq_tec]))
if len(new_seq_tec_list) > 0:
- info_for_yaml_dict['technology']['sample_sequencing_technology'] = [x for x in new_seq_tec_list]
+ technology['sample_sequencing_technology'] = [x for x in new_seq_tec_list]
else:
- info_for_yaml_dict['technology'][field_in_yaml] = tech_info_to_parse
+ technology[field_in_yaml] = tech_info_to_parse
for GBFeature in GBSeq.iter('GBFeature'):
@@ -138,15 +168,15 @@ if None:
GBQualifier_value_text = GBQualifier_value_text.split(';')[0] # For case like Homo sapiens;sex:female
if GBQualifier_value_text in field_to_term_to_uri_dict['ncbi_host_species']:
# Cases like 'Felis catus; Domestic Shorthair'
- info_for_yaml_dict['host']['host_species'] = field_to_term_to_uri_dict['ncbi_host_species'][GBQualifier_value_text]
+ host['host_species'] = field_to_term_to_uri_dict['ncbi_host_species'][GBQualifier_value_text]
else:
GBQualifier_value_text_list = GBQualifier_value_text.split('; ')
if GBQualifier_value_text_list[0] in field_to_term_to_uri_dict['ncbi_host_species']:
- info_for_yaml_dict['host']['host_species'] = field_to_term_to_uri_dict['ncbi_host_species'][GBQualifier_value_text_list[0]]
+ host['host_species'] = field_to_term_to_uri_dict['ncbi_host_species'][GBQualifier_value_text_list[0]]
elif GBQualifier_value_text_list[0] and ('MT215193' in accession_version or 'MT270814' in accession_version):
# Information checked manually from NCBI Virus
- info_for_yaml_dict['host']['host_species'] = field_to_term_to_uri_dict['ncbi_host_species']['Canis lupus familiaris']
+ host['host_species'] = field_to_term_to_uri_dict['ncbi_host_species']['Canis lupus familiaris']
else:
missing_value_list.append('\t'.join([accession_version, 'host_species', GBQualifier_value_text_list[0]]))
@@ -174,9 +204,9 @@ if None:
host_sex = 'female' if host_sex_one_lecter == 'F' else 'male'
if host_sex in ['male', 'female']:
- info_for_yaml_dict['host']['host_sex'] = "http://purl.obolibrary.org/obo/PATO_0000384" if host_sex == 'male' else "http://purl.obolibrary.org/obo/PATO_0000383"
+ host['host_sex'] = "http://purl.obolibrary.org/obo/PATO_0000384" if host_sex == 'male' else "http://purl.obolibrary.org/obo/PATO_0000383"
elif GBQualifier_value_text_list[1] in field_to_term_to_uri_dict['ncbi_host_health_status']:
- info_for_yaml_dict['host']['host_health_status'] = field_to_term_to_uri_dict['ncbi_host_health_status'][GBQualifier_value_text_list[1]]
+ host['host_health_status'] = field_to_term_to_uri_dict['ncbi_host_health_status'][GBQualifier_value_text_list[1]]
else:
missing_value_list.append('\t'.join([accession_version, 'host_sex or host_health_status', GBQualifier_value_text_list[1]]))
@@ -188,15 +218,15 @@ if None:
host_age = int(GBQualifier_value_text_list[2].split(' ')[-1])
if host_age >= 0 and host_age < 110:
- info_for_yaml_dict['host']['host_age'] = host_age
- info_for_yaml_dict['host']['host_age_unit'] = 'http://purl.obolibrary.org/obo/UO_0000036'
+ host['host_age'] = host_age
+ host['host_age_unit'] = 'http://purl.obolibrary.org/obo/UO_0000036'
elif len(GBQualifier_value_text_list) > 2:
missing_value_list.append('\t'.join([accession_version, 'host_age', GBQualifier_value_text_list[2]]))
elif GBQualifier_name_text == 'collected_by':
if any([x in GBQualifier_value_text.lower() for x in ['institute', 'hospital', 'city', 'center']]):
- info_for_yaml_dict['sample']['collecting_institution'] = GBQualifier_value_text
+ sample['collecting_institution'] = GBQualifier_value_text
else:
- info_for_yaml_dict['sample']['collector_name'] = GBQualifier_value_text
+ sample['collector_name'] = GBQualifier_value_text
elif GBQualifier_name_text == 'isolation_source':
if GBQualifier_value_text.upper() in field_to_term_to_uri_dict['ncbi_speciesman_source']:
GBQualifier_value_text = GBQualifier_value_text.upper() # For example, in case of 'usa: wa'
@@ -205,18 +235,18 @@ if None:
GBQualifier_value_text = GBQualifier_value_text.strip("/'")
if GBQualifier_value_text in field_to_term_to_uri_dict['ncbi_speciesman_source']:
- info_for_yaml_dict['sample']['specimen_source'] = [field_to_term_to_uri_dict['ncbi_speciesman_source'][GBQualifier_value_text]]
+ sample['specimen_source'] = [field_to_term_to_uri_dict['ncbi_speciesman_source'][GBQualifier_value_text]]
else:
if GBQualifier_value_text.lower() in ['np/op', 'np-op', 'np/op swab', 'np/np swab', 'nasopharyngeal and oropharyngeal swab', 'nasopharyngeal/oropharyngeal swab', 'combined nasopharyngeal and oropharyngeal swab', 'naso and/or oropharyngeal swab']:
- info_for_yaml_dict['sample']['specimen_source'] = [field_to_term_to_uri_dict['ncbi_speciesman_source']['nasopharyngeal swab'], field_to_term_to_uri_dict['ncbi_speciesman_source']['oropharyngeal swab']]
+ sample['specimen_source'] = [field_to_term_to_uri_dict['ncbi_speciesman_source']['nasopharyngeal swab'], field_to_term_to_uri_dict['ncbi_speciesman_source']['oropharyngeal swab']]
elif GBQualifier_value_text.lower() in ['nasopharyngeal swab/throat swab', 'nasopharyngeal/throat swab', 'nasopharyngeal swab and throat swab', 'nasal swab and throat swab', 'nasopharyngeal aspirate/throat swab', 'Nasopharyngeal/Throat']:
- info_for_yaml_dict['sample']['specimen_source'] = [field_to_term_to_uri_dict['ncbi_speciesman_source']['nasopharyngeal swab'], field_to_term_to_uri_dict['ncbi_speciesman_source']['throat swab']]
+ sample['specimen_source'] = [field_to_term_to_uri_dict['ncbi_speciesman_source']['nasopharyngeal swab'], field_to_term_to_uri_dict['ncbi_speciesman_source']['throat swab']]
elif GBQualifier_value_text.lower() in ['nasopharyngeal aspirate & throat swab', 'nasopharyngeal aspirate and throat swab']:
- info_for_yaml_dict['sample']['specimen_source'] = [field_to_term_to_uri_dict['ncbi_speciesman_source']['nasopharyngeal aspirate'], field_to_term_to_uri_dict['ncbi_speciesman_source']['throat swab']]
+ sample['specimen_source'] = [field_to_term_to_uri_dict['ncbi_speciesman_source']['nasopharyngeal aspirate'], field_to_term_to_uri_dict['ncbi_speciesman_source']['throat swab']]
elif GBQualifier_value_text.lower() in ['nasal swab and throat swab']:
- info_for_yaml_dict['sample']['specimen_source'] = [field_to_term_to_uri_dict['ncbi_speciesman_source']['nasal swab'], field_to_term_to_uri_dict['ncbi_speciesman_source']['throat swab']]
+ sample['specimen_source'] = [field_to_term_to_uri_dict['ncbi_speciesman_source']['nasal swab'], field_to_term_to_uri_dict['ncbi_speciesman_source']['throat swab']]
elif GBQualifier_value_text.lower() in ['nasal-swab and oro-pharyngeal swab']:
- info_for_yaml_dict['sample']['specimen_source'] = [field_to_term_to_uri_dict['ncbi_speciesman_source']['nasal swab'], field_to_term_to_uri_dict['ncbi_speciesman_source']['oropharyngeal swab']]
+ sample['specimen_source'] = [field_to_term_to_uri_dict['ncbi_speciesman_source']['nasal swab'], field_to_term_to_uri_dict['ncbi_speciesman_source']['oropharyngeal swab']]
else:
missing_value_list.append('\t'.join([accession_version, 'specimen_source', GBQualifier_value_text]))
elif GBQualifier_name_text == 'collection_date':
@@ -229,60 +259,60 @@ if None:
else:
date_to_write = "{}-01-15".format(GBQualifier_value_text)
- if 'additional_collection_information' in info_for_yaml_dict['sample']:
- info_for_yaml_dict['sample']['additional_collection_information'] += "; The 'collection_date' is estimated (the original date was: {})".format(GBQualifier_value_text)
+ if 'additional_collection_information' in sample:
+ sample['additional_collection_information'] += "; The 'collection_date' is estimated (the original date was: {})".format(GBQualifier_value_text)
else:
- info_for_yaml_dict['sample']['additional_collection_information'] = "The 'collection_date' is estimated (the original date was: {})".format(GBQualifier_value_text)
+ sample['additional_collection_information'] = "The 'collection_date' is estimated (the original date was: {})".format(GBQualifier_value_text)
elif len(GBQualifier_value_text.split('-')) == 2:
date_to_write = parse(GBQualifier_value_text).strftime('%Y-%m') + '-15'
- if 'additional_collection_information' in info_for_yaml_dict['sample']:
- info_for_yaml_dict['sample']['additional_collection_information'] += "; The 'collection_date' is estimated (the original date was: {})".format(GBQualifier_value_text)
+ if 'additional_collection_information' in sample:
+ sample['additional_collection_information'] += "; The 'collection_date' is estimated (the original date was: {})".format(GBQualifier_value_text)
else:
- info_for_yaml_dict['sample']['additional_collection_information'] = "The 'collection_date' is estimated (the original date was: {})".format(GBQualifier_value_text)
+ sample['additional_collection_information'] = "The 'collection_date' is estimated (the original date was: {})".format(GBQualifier_value_text)
elif len(GBQualifier_value_text.split('-')) == 3:
GBQualifier_value_text_list = GBQualifier_value_text.split('-')
if GBQualifier_value_text_list[1].isalpha():
date_to_write = parse(GBQualifier_value_text).strftime('%Y-%m-%d')
- info_for_yaml_dict['sample']['collection_date'] = date_to_write
+ sample['collection_date'] = date_to_write
elif GBQualifier_name_text in ['lat_lon', 'country']:
if GBQualifier_name_text == 'country' and ': ' in GBQualifier_value_text:
GBQualifier_value_text = GBQualifier_value_text.replace(': ', ':')
if GBQualifier_value_text in field_to_term_to_uri_dict['ncbi_countries']:
- info_for_yaml_dict['sample']['collection_location'] = field_to_term_to_uri_dict['ncbi_countries'][GBQualifier_value_text]
+ sample['collection_location'] = field_to_term_to_uri_dict['ncbi_countries'][GBQualifier_value_text]
else:
missing_value_list.append('\t'.join([accession_version, GBQualifier_name_text, GBQualifier_value_text]))
elif GBQualifier_name_text == 'note':
- if 'additional_collection_information' in info_for_yaml_dict['sample']:
- info_for_yaml_dict['sample']['additional_collection_information'] += '; ' + GBQualifier_value_text
+ if 'additional_collection_information' in sample:
+ sample['additional_collection_information'] += '; ' + GBQualifier_value_text
else:
- info_for_yaml_dict['sample']['additional_collection_information'] = GBQualifier_value_text
+ sample['additional_collection_information'] = GBQualifier_value_text
elif GBQualifier_name_text == 'isolate':
- info_for_yaml_dict['virus']['virus_strain'] = GBQualifier_value_text
+ virus['virus_strain'] = GBQualifier_value_text
elif GBQualifier_name_text == 'db_xref':
- info_for_yaml_dict['virus']['virus_species'] = "http://purl.obolibrary.org/obo/NCBITaxon_"+GBQualifier_value_text.split('taxon:')[1]
+ virus['virus_species'] = "http://purl.obolibrary.org/obo/NCBITaxon_"+GBQualifier_value_text.split('taxon:')[1]
# Check if mandatory fields are missing
- if 'sample_sequencing_technology' not in info_for_yaml_dict['technology']:
+ if 'sample_sequencing_technology' not in technology:
# print(accession_version, ' - technology not found')
if accession_version not in not_created_accession_dict:
not_created_accession_dict[accession_version] = []
not_created_accession_dict[accession_version].append('sample_sequencing_technology not found')
- if 'collection_location' not in info_for_yaml_dict['sample']:
+ if 'collection_location' not in sample:
if accession_version not in not_created_accession_dict:
not_created_accession_dict[accession_version] = []
not_created_accession_dict[accession_version].append('collection_location not found')
- if 'collection_date' not in info_for_yaml_dict['sample']:
+ if 'collection_date' not in sample:
if accession_version not in not_created_accession_dict:
not_created_accession_dict[accession_version] = []
not_created_accession_dict[accession_version].append('collection_date not found')
else:
- year, month, day = [int(x) for x in info_for_yaml_dict['sample']['collection_date'].split('-')]
+ year, month, day = [int(x) for x in sample['collection_date'].split('-')]
collection_date_in_yaml = datetime(year, month, day)
if collection_date_in_yaml < min_acceptable_collection_date:
@@ -290,12 +320,12 @@ if None:
not_created_accession_dict[accession_version] = []
not_created_accession_dict[accession_version].append('collection_date too early')
- if 'authors' not in info_for_yaml_dict['submitter']:
+ if 'authors' not in submitter:
if accession_version not in not_created_accession_dict:
not_created_accession_dict[accession_version] = []
not_created_accession_dict[accession_version].append('authors not found')
- if 'host_species' not in info_for_yaml_dict['host']:
+ if 'host_species' not in host:
if accession_version not in not_created_accession_dict:
not_created_accession_dict[accession_version] = []
not_created_accession_dict[accession_version].append('host_species not found')