First import
This commit is contained in:
@@ -0,0 +1,79 @@
|
||||
from io import BytesIO
|
||||
from openpyxl import load_workbook
|
||||
|
||||
|
||||
def excel_dict_reader(fhand, sheet_name, mandatory_column_name=None):
|
||||
fhand.seek(0)
|
||||
wb = load_workbook(filename=BytesIO(fhand.read()), data_only=True,
|
||||
read_only=True)
|
||||
return workbook_sheet_reader(wb, sheet_name, mandatory_column_name=mandatory_column_name)
|
||||
|
||||
|
||||
def is_none(value):
|
||||
return value is None
|
||||
|
||||
|
||||
def workbook_sheet_reader(workbook, sheet_name, mandatory_column_name=None,
|
||||
allowed_empty_line_slots=5):
|
||||
try:
|
||||
sheet = workbook[sheet_name]
|
||||
except KeyError as error:
|
||||
raise ValueError(f"The '{sheet_name}' sheet is missing.") from error
|
||||
|
||||
first = True
|
||||
header = []
|
||||
empty_lines = 0
|
||||
for row in sheet.rows:
|
||||
values = []
|
||||
for cell in row:
|
||||
if cell.value is not None and cell.data_type == 's':
|
||||
value = str(cell.value).strip()
|
||||
else:
|
||||
value = cell.value
|
||||
values.append(value)
|
||||
# values = [cell.value.strip() for cell in row]
|
||||
if first:
|
||||
header = values
|
||||
first = False
|
||||
continue
|
||||
if not any(values):
|
||||
empty_lines += 1
|
||||
if empty_lines >= allowed_empty_line_slots:
|
||||
break
|
||||
continue
|
||||
empty_lines = 0
|
||||
|
||||
data = dict(zip(header, values))
|
||||
if mandatory_column_name is not None and not data[mandatory_column_name]:
|
||||
# msg = f"Exiting before end of sheet {sheet_name} ends.\n"
|
||||
# msg += f"Mandatory column ({mandatory_column_name}) empty. \n"
|
||||
# msg += "Check file for empty lines"
|
||||
# print(msg)
|
||||
continue
|
||||
yield data
|
||||
|
||||
|
||||
def get_all_cell_data_from_sheet(workbook, sheet_name, allowed_empty_line_slots=5):
|
||||
try:
|
||||
sheet = workbook[sheet_name]
|
||||
except KeyError as error:
|
||||
raise ValueError(f"The '{sheet_name}' sheet is missing.") from error
|
||||
|
||||
empty_lines = 0
|
||||
all_values = []
|
||||
for row in sheet.rows:
|
||||
values = []
|
||||
for cell in row:
|
||||
if cell.value is not None and cell.data_type == 's':
|
||||
value = str(cell.value).strip()
|
||||
else:
|
||||
value = cell.value
|
||||
values.append(value)
|
||||
if not any(values):
|
||||
empty_lines += 1
|
||||
if empty_lines >= allowed_empty_line_slots:
|
||||
break
|
||||
continue
|
||||
empty_lines = 0
|
||||
all_values.extend(values)
|
||||
return all_values
|
||||
@@ -0,0 +1,276 @@
|
||||
import re
|
||||
from datetime import date
|
||||
from io import BytesIO
|
||||
|
||||
import pycountry
|
||||
from openpyxl import load_workbook
|
||||
|
||||
from mirri import rsetattr, ValidationError
|
||||
from mirri.biolomics.serializers.sequence import GenomicSequenceBiolomics
|
||||
from mirri.biolomics.serializers.strain import StrainMirri
|
||||
from mirri.entities.growth_medium import GrowthMedium
|
||||
from mirri.io.parsers.excel import workbook_sheet_reader
|
||||
from mirri.entities.publication import Publication
|
||||
from mirri.entities.date_range import DateRange
|
||||
from mirri.entities.strain import OrganismType, StrainId, add_taxon_to_strain
|
||||
from mirri.settings import (COMMERCIAL_USE_WITH_AGREEMENT, GENOMIC_INFO,
|
||||
GROWTH_MEDIA, LITERATURE_SHEET, LOCATIONS,
|
||||
MIRRI_FIELDS, NAGOYA_DOCS_AVAILABLE, NAGOYA_NO_RESTRICTIONS,
|
||||
NAGOYA_PROBABLY_SCOPE, NO_RESTRICTION,
|
||||
ONLY_RESEARCH, ONTOBIOTOPE,
|
||||
PUBLICATION_FIELDS, STRAINS, SUBTAXAS)
|
||||
from mirri.utils import get_country_from_name
|
||||
|
||||
RESTRICTION_USE_TRANSLATOR = {
|
||||
1: NO_RESTRICTION,
|
||||
2: ONLY_RESEARCH,
|
||||
3: COMMERCIAL_USE_WITH_AGREEMENT,
|
||||
}
|
||||
NAGOYA_TRANSLATOR = {
|
||||
1: NAGOYA_NO_RESTRICTIONS,
|
||||
2: NAGOYA_DOCS_AVAILABLE,
|
||||
3: NAGOYA_PROBABLY_SCOPE,
|
||||
}
|
||||
TRUEFALSE_TRANSLATOR = {
|
||||
1: False,
|
||||
2: True
|
||||
}
|
||||
|
||||
|
||||
def parse_mirri_excel(fhand, version="20200601"):
|
||||
if version == "20200601":
|
||||
return _parse_mirri_v20200601(fhand)
|
||||
else:
|
||||
raise NotImplementedError("Only version 20200601 is implemented")
|
||||
|
||||
|
||||
def _parse_mirri_v20200601(fhand):
|
||||
fhand.seek(0)
|
||||
file_content = BytesIO(fhand.read())
|
||||
wb = load_workbook(filename=file_content, read_only=True, data_only=True)
|
||||
|
||||
locations = workbook_sheet_reader(wb, LOCATIONS)
|
||||
ontobiotopes = workbook_sheet_reader(wb, ONTOBIOTOPE)
|
||||
|
||||
growth_media = list(parse_growth_media(wb))
|
||||
|
||||
markers = workbook_sheet_reader(wb, GENOMIC_INFO)
|
||||
|
||||
publications = list(parse_publications(wb))
|
||||
|
||||
strains = parse_strains(wb, locations=locations, growth_media=growth_media,
|
||||
markers=markers, publications=publications,
|
||||
ontobiotopes=ontobiotopes)
|
||||
|
||||
return {"strains": strains, "growth_media": growth_media}
|
||||
|
||||
|
||||
def index_list_by(list_, id_):
|
||||
return {str(item[id_]): item for item in list_}
|
||||
|
||||
|
||||
def index_list_by_attr(list_, id_):
|
||||
return {str(getattr(item, id_)): item for item in list_}
|
||||
|
||||
|
||||
def index_markers(markers):
|
||||
indexed_markers = {}
|
||||
for marker in markers:
|
||||
strain_id = marker["Strain AN"]
|
||||
if strain_id not in indexed_markers:
|
||||
indexed_markers[strain_id] = []
|
||||
indexed_markers[strain_id].append(marker)
|
||||
return indexed_markers
|
||||
|
||||
|
||||
def remove_hard_lines(string=None):
|
||||
if string is not None and string != '':
|
||||
return re.sub(r'\r+\n+|\t+', '', string).strip()
|
||||
else:
|
||||
return None
|
||||
|
||||
|
||||
def parse_growth_media(wb):
|
||||
for row in workbook_sheet_reader(wb, GROWTH_MEDIA):
|
||||
gm = GrowthMedium()
|
||||
gm.acronym = str(row['Acronym'])
|
||||
gm.description = row['Description']
|
||||
gm.full_description = remove_hard_lines(row.get('Full description', None))
|
||||
|
||||
yield gm
|
||||
|
||||
|
||||
def parse_publications(wb):
|
||||
ids = []
|
||||
for row in workbook_sheet_reader(wb, LITERATURE_SHEET):
|
||||
pub = Publication()
|
||||
for pub_field in PUBLICATION_FIELDS:
|
||||
label = pub_field["label"]
|
||||
col_val = row.get(label, None)
|
||||
|
||||
if col_val:
|
||||
attribute = pub_field["attribute"]
|
||||
setattr(pub, attribute, col_val)
|
||||
yield pub
|
||||
|
||||
|
||||
def parse_strains(wb, locations, growth_media, markers, publications,
|
||||
ontobiotopes):
|
||||
|
||||
ontobiotopes_by_id = {str(ont["ID"]): ont['Name'] for ont in ontobiotopes}
|
||||
ontobiotopes_by_name = {v: k for k, v in ontobiotopes_by_id.items()}
|
||||
|
||||
locations = index_list_by(locations, 'Locality')
|
||||
growth_media = index_list_by_attr(growth_media, 'acronym')
|
||||
publications = index_list_by_attr(publications, 'id')
|
||||
markers = index_markers(markers)
|
||||
|
||||
for strain_row in workbook_sheet_reader(wb, STRAINS, "Accession number"):
|
||||
strain = StrainMirri()
|
||||
strain_id = None
|
||||
label = None
|
||||
for field in MIRRI_FIELDS:
|
||||
label = field["label"]
|
||||
attribute = field["attribute"]
|
||||
value = strain_row[label]
|
||||
if value is None or value == '':
|
||||
continue
|
||||
|
||||
if attribute == "id":
|
||||
collection, number = value.split(" ", 1)
|
||||
value = StrainId(collection=collection, number=number)
|
||||
rsetattr(strain, attribute, value)
|
||||
|
||||
elif attribute == "restriction_on_use":
|
||||
rsetattr(strain, attribute, RESTRICTION_USE_TRANSLATOR[value])
|
||||
elif attribute == "nagoya_protocol":
|
||||
rsetattr(strain, attribute, NAGOYA_TRANSLATOR[value])
|
||||
elif attribute == "other_numbers":
|
||||
other_numbers = []
|
||||
for on in value.split(";"):
|
||||
on = on.strip()
|
||||
try:
|
||||
collection, number = on.split(" ", 1)
|
||||
except ValueError:
|
||||
collection = None
|
||||
number = on
|
||||
_id = StrainId(collection=collection, number=number)
|
||||
other_numbers.append(_id)
|
||||
rsetattr(strain, attribute, other_numbers)
|
||||
elif attribute == "taxonomy.taxon_name":
|
||||
try:
|
||||
add_taxon_to_strain(strain, value)
|
||||
except ValueError:
|
||||
msg = f"The '{label}' for strain with Accession Number {strain_id} is not according to the specification."
|
||||
raise ValidationError(msg)
|
||||
elif attribute == "taxonomy.organism_type":
|
||||
value = [OrganismType(val.strip())
|
||||
for val in str(value).split(";")]
|
||||
rsetattr(strain, attribute, value)
|
||||
elif attribute in ("deposit.date", "collect.date", "isolation.date",
|
||||
"catalog_inclusion_date"):
|
||||
if isinstance(value, date):
|
||||
value = DateRange(
|
||||
year=value.year, month=value.month, day=value.day
|
||||
)
|
||||
elif isinstance(value, str):
|
||||
value = DateRange().strpdate(value)
|
||||
else:
|
||||
raise NotImplementedError()
|
||||
rsetattr(strain, attribute, value)
|
||||
elif attribute == 'growth.recommended_temp':
|
||||
temps = value.split(';')
|
||||
if len(temps) == 1:
|
||||
_min, _max = float(temps[0]), float(temps[0])
|
||||
else:
|
||||
_min, _max = float(temps[0]), float(temps[1])
|
||||
rsetattr(strain, attribute, {'min': _min, 'max': _max})
|
||||
elif attribute == "growth.recommended_media":
|
||||
sep = "/"
|
||||
if ";" in value:
|
||||
sep = ";"
|
||||
growth_media = [v.strip() for v in value.split(sep)]
|
||||
rsetattr(strain, attribute, growth_media)
|
||||
elif attribute == 'growth.tested_temp_range':
|
||||
if value:
|
||||
min_, max_ = value.split(";")
|
||||
value = {'min': float(min_), 'max': float(max_)}
|
||||
rsetattr(strain, attribute, value)
|
||||
elif attribute == "form_of_supply":
|
||||
rsetattr(strain, attribute, value.split(";"))
|
||||
elif attribute == "collect.location.coords":
|
||||
items = value.split(";")
|
||||
strain.collect.location.latitude = float(items[0])
|
||||
strain.collect.location.longitude = float(items[1])
|
||||
if len(items) > 2:
|
||||
strain.collect.location.coord_uncertainty = items[2]
|
||||
|
||||
elif attribute == "collect.location":
|
||||
location = locations[value]
|
||||
if 'Country' in location and location['Country']:
|
||||
if location['Country'] == 'Unknown':
|
||||
continue
|
||||
country_3 = _get_country_alpha3(location['Country'])
|
||||
strain.collect.location.country = country_3
|
||||
strain.collect.location.state = location["Region"]
|
||||
strain.collect.location.municipality = location["City"]
|
||||
strain.collect.location.site = location["Locality"]
|
||||
elif attribute in ("abs_related_files", "mta_files"):
|
||||
rsetattr(strain, attribute, value.split(";"))
|
||||
elif attribute in ("is_from_registered_collection",
|
||||
"is_subject_to_quarantine", 'taxonomy.interspecific_hybrid',
|
||||
"is_potentially_harmful", "genetics.gmo"):
|
||||
rsetattr(strain, attribute, TRUEFALSE_TRANSLATOR[value])
|
||||
elif attribute == "publications":
|
||||
value = str(value)
|
||||
pubs = []
|
||||
pub_ids = [v.strip() for v in str(value).split(";")]
|
||||
for pub_id in pub_ids:
|
||||
pub = publications.get(pub_id, None)
|
||||
if pub is None:
|
||||
pub = Publication()
|
||||
if '/' in pub_id:
|
||||
pub.doi = pub_id
|
||||
else:
|
||||
pub.pubmed_id = pub_id
|
||||
pubs.append(pub)
|
||||
rsetattr(strain, attribute, pubs)
|
||||
elif attribute == 'ontobiotope':
|
||||
values = []
|
||||
for val in value.split(';'):
|
||||
if val not in ontobiotopes_by_id:
|
||||
val = ontobiotopes_by_name[val]
|
||||
values.append(val)
|
||||
rsetattr(strain, attribute, value)
|
||||
elif attribute == 'other_denominations':
|
||||
value = [v.strip() for v in value.split(';')]
|
||||
rsetattr(strain, attribute, value)
|
||||
elif attribute == 'genetics.plasmids':
|
||||
value = [v.strip() for v in value.split(';')]
|
||||
rsetattr(strain, attribute, value)
|
||||
else:
|
||||
#print(attribute, value, type(value))
|
||||
rsetattr(strain, attribute, value)
|
||||
|
||||
# add markers
|
||||
strain_id = strain.id.strain_id
|
||||
if strain_id in markers:
|
||||
for marker in markers[strain_id]:
|
||||
_marker = GenomicSequenceBiolomics()
|
||||
_marker.marker_id = marker["INSDC AN"]
|
||||
_marker.marker_type = marker["Marker"]
|
||||
_marker.marker_seq = marker["Sequence"]
|
||||
strain.genetics.markers.append(_marker)
|
||||
yield strain
|
||||
|
||||
|
||||
def _get_country_alpha3(loc_country):
|
||||
if loc_country == 'INW':
|
||||
return loc_country
|
||||
country = get_country_from_name(loc_country)
|
||||
if not country:
|
||||
country = pycountry.countries.get(alpha_3=loc_country)
|
||||
if not country:
|
||||
country = pycountry.historic_countries.get(alpha_3=loc_country)
|
||||
country_3 = country.alpha_3
|
||||
return country_3
|
||||
@@ -0,0 +1,305 @@
|
||||
import csv
|
||||
from copy import deepcopy
|
||||
from openpyxl.workbook.workbook import Workbook
|
||||
|
||||
|
||||
from mirri import rgetattr
|
||||
from mirri.settings import GROWTH_MEDIA, MIRRI_FIELDS, DATA_DIR, PUBLICATION_FIELDS
|
||||
from mirri.io.parsers.mirri_excel import NAGOYA_TRANSLATOR, RESTRICTION_USE_TRANSLATOR
|
||||
|
||||
INITIAL_SEXUAL_STATES = [
|
||||
"Mata",
|
||||
"Matalpha",
|
||||
"Mata/Matalpha",
|
||||
"Mata",
|
||||
"Matb",
|
||||
"Mata/Matb",
|
||||
"MTLa",
|
||||
"MTLalpha",
|
||||
"MTLa/MTLalpha",
|
||||
"MAT1-1",
|
||||
"MAT1-2",
|
||||
"MAT1",
|
||||
"MAT2",
|
||||
"MT+",
|
||||
"MT-",
|
||||
"MT+",
|
||||
"MT-",
|
||||
"H+",
|
||||
"H-",
|
||||
]
|
||||
MARKER_FIELDS = [
|
||||
{"attribute": "acronym", "label": "Acronym", "mandatory": True},
|
||||
{"attribute": "marker", "label": "Marker", "mandatory": True},
|
||||
]
|
||||
MARKER_DATA = [
|
||||
{"acronym": "16S rRNA", "marker": "16S rRNA"},
|
||||
{"acronym": "ACT", "marker": "Actin"},
|
||||
{"acronym": "CaM", "marker": "Calmodulin"},
|
||||
{"acronym": "EF-1α", "marker": "elongation factor 1-alpha (EF-1α)"},
|
||||
{"acronym": "ITS", "marker": "nuclear ribosomal Internal Transcribed Spacer (ITS)"},
|
||||
{"acronym": "LSU", "marker": "nuclear ribosomal Large SubUnit (LSU)"},
|
||||
{"acronym": "RPB1", "marker": "Ribosomal RNA-coding genes RPB1"},
|
||||
{"acronym": "RPB2", "marker": "Ribosomal RNA-coding genes RPB2"},
|
||||
{"acronym": "TUBB", "marker": "β-Tubulin"},
|
||||
]
|
||||
|
||||
REV_RESTRICTION_USE_TRANSLATOR = {v: k for k, v in RESTRICTION_USE_TRANSLATOR.items()}
|
||||
REV_NAGOYA_TRANSLATOR = {v: k for k, v in NAGOYA_TRANSLATOR.items()}
|
||||
PUB_HEADERS = [pb["label"] for pb in PUBLICATION_FIELDS]
|
||||
|
||||
|
||||
def write_mirri_excel(path, strains, growth_media, version):
|
||||
if version == "20200601":
|
||||
_write_mirri_excel_20200601(path, strains, growth_media)
|
||||
|
||||
|
||||
def _write_mirri_excel_20200601(path, strains, growth_media):
|
||||
wb = Workbook()
|
||||
|
||||
write_markers_sheet(wb)
|
||||
|
||||
ontobiotope_path = DATA_DIR / "ontobiotopes.csv"
|
||||
write_ontobiotopes(wb, ontobiotope_path)
|
||||
|
||||
write_growth_media(wb, growth_media)
|
||||
growth_media_indexes = [str(gm.acronym) for gm in growth_media]
|
||||
|
||||
locations = {}
|
||||
publications = {}
|
||||
sexual_states = set(deepcopy(INITIAL_SEXUAL_STATES))
|
||||
genomic_markers = {}
|
||||
strains_data = _deserialize_strains(strains, locations, growth_media_indexes,
|
||||
publications, sexual_states, genomic_markers)
|
||||
strains_data = list(strains_data)
|
||||
|
||||
# write strain to generate indexed data
|
||||
strain_sheet = wb.create_sheet("Strains")
|
||||
strain_sheet.append([field["label"] for field in MIRRI_FIELDS])
|
||||
for strain_row in strains_data:
|
||||
strain_sheet.append(strain_row)
|
||||
redimension_cell_width(strain_sheet)
|
||||
|
||||
# write locations
|
||||
loc_sheet = wb.create_sheet("Geographic origin")
|
||||
loc_sheet.append(["ID", "Country", "Region", "City", "Locality"])
|
||||
for index, loc_index in enumerate(locations.keys()):
|
||||
location = locations[loc_index]
|
||||
row = [index, location.country, location.state, location.municipality,
|
||||
loc_index]
|
||||
loc_sheet.append(row)
|
||||
redimension_cell_width(loc_sheet)
|
||||
|
||||
# write publications
|
||||
pub_sheet = wb.create_sheet("Literature")
|
||||
pub_sheet.append(PUB_HEADERS)
|
||||
for publication in publications.values():
|
||||
row = []
|
||||
for pub_field in PUBLICATION_FIELDS:
|
||||
# if pub_field['attribute'] == 'id':
|
||||
# value = index
|
||||
value = getattr(publication, pub_field['attribute'], None)
|
||||
row.append(value)
|
||||
pub_sheet.append(row)
|
||||
redimension_cell_width(pub_sheet)
|
||||
|
||||
# write sexual states
|
||||
sex_sheet = wb.create_sheet("Sexual states")
|
||||
for sex_state in sorted(list(sexual_states)):
|
||||
sex_sheet.append([sex_state])
|
||||
redimension_cell_width(sex_sheet)
|
||||
|
||||
# write genetic markers
|
||||
markers_sheet = wb.create_sheet("Genomic information")
|
||||
markers_sheet.append(['Strain AN', 'Marker', 'INSDC AN', 'Sequence'])
|
||||
for strain_id, markers in genomic_markers.items():
|
||||
for marker in markers:
|
||||
row = [strain_id, marker.marker_type, marker.marker_id, marker.marker_seq]
|
||||
markers_sheet.append(row)
|
||||
redimension_cell_width(markers_sheet)
|
||||
|
||||
del wb["Sheet"]
|
||||
wb.save(str(path))
|
||||
|
||||
|
||||
def _deserialize_strains(strains, locations, growth_media_indexes,
|
||||
publications, sexual_states, genomic_markers):
|
||||
for strain in strains:
|
||||
strain_row = []
|
||||
for field in MIRRI_FIELDS:
|
||||
attribute = field["attribute"]
|
||||
|
||||
if attribute == "id":
|
||||
value = strain.id.strain_id
|
||||
elif attribute == "restriction_on_use":
|
||||
value = rgetattr(strain, attribute)
|
||||
if value is not None:
|
||||
value = REV_RESTRICTION_USE_TRANSLATOR[value]
|
||||
elif attribute == "nagoya_protocol":
|
||||
value = rgetattr(strain, attribute)
|
||||
if value:
|
||||
value = REV_NAGOYA_TRANSLATOR[value]
|
||||
elif attribute == "other_numbers":
|
||||
value = rgetattr(strain, attribute)
|
||||
if value is not None:
|
||||
value = [f"{on.collection} {on.number}" for on in value]
|
||||
value = "; ".join(value)
|
||||
elif attribute == 'other_denominations':
|
||||
od = strain.other_denominations
|
||||
value = "; ".join(od) if od else None
|
||||
elif attribute in (
|
||||
"is_from_registered_collection",
|
||||
"is_subject_to_quarantine",
|
||||
"is_potentially_harmful",
|
||||
"genetics.gmo",
|
||||
"taxonomy.interspecific_hybrid"
|
||||
):
|
||||
value = rgetattr(strain, attribute)
|
||||
if value is True:
|
||||
value = 2
|
||||
elif value is False:
|
||||
value = 1
|
||||
else:
|
||||
value = None
|
||||
elif attribute == "taxonomy.taxon_name":
|
||||
value = strain.taxonomy.long_name
|
||||
elif attribute in ("deposit.date", "collect.date", "isolation.date",
|
||||
'catalog_inclusion_date'):
|
||||
value = rgetattr(strain, attribute)
|
||||
value = value.strfdate if value else None
|
||||
elif attribute == "growth.recommended_media":
|
||||
value = rgetattr(strain, attribute)
|
||||
if value is not None:
|
||||
for gm in value:
|
||||
gm = str(gm)
|
||||
if gm not in growth_media_indexes:
|
||||
print(gm, growth_media_indexes)
|
||||
msg = f"Growth media {gm} not in the provided ones"
|
||||
continue
|
||||
raise ValueError(msg)
|
||||
value = "/".join(value)
|
||||
elif attribute in ('growth.tested_temp_range',
|
||||
"growth.recommended_temp"):
|
||||
value = rgetattr(strain, attribute)
|
||||
if value:
|
||||
value = f'{value["min"]}; {value["max"]}'
|
||||
elif attribute == "form_of_supply":
|
||||
value = rgetattr(strain, attribute)
|
||||
value = ";".join(value)
|
||||
elif attribute == "collect.location.coords":
|
||||
lat = strain.collect.location.latitude
|
||||
long = strain.collect.location.longitude
|
||||
if lat is not None and long is not None:
|
||||
value = f"{lat};{long}"
|
||||
else:
|
||||
value = None
|
||||
|
||||
elif attribute == "collect.location":
|
||||
location = strain.collect.location
|
||||
loc_index = _build_location_index(location)
|
||||
if loc_index is None:
|
||||
continue
|
||||
if loc_index not in locations:
|
||||
locations[loc_index] = location
|
||||
value = loc_index
|
||||
elif attribute in ("abs_related_files", "mta_files"):
|
||||
value = rgetattr(strain, attribute)
|
||||
value = ";".join(value) if value else None
|
||||
elif attribute == "taxonomy.organism_type":
|
||||
value = rgetattr(strain, attribute)
|
||||
if value:
|
||||
value = "; ".join([str(v.code) for v in value])
|
||||
|
||||
elif attribute == "history":
|
||||
value = rgetattr(strain, attribute)
|
||||
if value is not None:
|
||||
value = " < ".join(value)
|
||||
elif attribute == "genetics.sexual_state":
|
||||
value = rgetattr(strain, attribute)
|
||||
if value:
|
||||
sexual_states.add(value)
|
||||
elif attribute == "genetics.ploidy":
|
||||
value = rgetattr(strain, attribute)
|
||||
elif attribute == "taxonomy.organism_type":
|
||||
organism_types = rgetattr(strain, attribute)
|
||||
if organism_types is not None:
|
||||
value = [org_type.code for org_type in organism_types]
|
||||
value = ";".join(value)
|
||||
elif attribute == 'publications':
|
||||
value = []
|
||||
for pub in strain.publications:
|
||||
value.append(pub.id)
|
||||
if pub.id not in publications:
|
||||
publications[pub.id] = pub
|
||||
value = ';'.join(str(v) for v in value) if value else None
|
||||
elif attribute == 'genetics.plasmids':
|
||||
value = rgetattr(strain, attribute)
|
||||
if value is not None:
|
||||
value = ';'.join(value)
|
||||
else:
|
||||
value = rgetattr(strain, attribute)
|
||||
|
||||
strain_row.append(value)
|
||||
genomic_markers[strain.id.strain_id] = strain.genetics.markers
|
||||
yield strain_row
|
||||
|
||||
|
||||
def _build_location_index(location):
|
||||
index = []
|
||||
if location.country:
|
||||
index.append(location.country)
|
||||
if location.site:
|
||||
index.append(location.site)
|
||||
return ';'.join(index) if index else None
|
||||
|
||||
|
||||
def write_markers_sheet(wb):
|
||||
sheet = wb.create_sheet("Markers")
|
||||
_write_work_sheet(
|
||||
sheet,
|
||||
labels=[f["label"] for f in MARKER_FIELDS],
|
||||
attributes=[f["attribute"] for f in MARKER_FIELDS],
|
||||
data=MARKER_DATA,
|
||||
)
|
||||
redimension_cell_width(sheet)
|
||||
|
||||
|
||||
def write_ontobiotopes(workbook, ontobiotype_path):
|
||||
ws = workbook.create_sheet("Ontobiotope")
|
||||
with ontobiotype_path.open() as fhand:
|
||||
for row in csv.reader(fhand, delimiter="\t"):
|
||||
ws.append(row)
|
||||
redimension_cell_width(ws)
|
||||
|
||||
|
||||
def _write_work_sheet(sheet, labels, attributes, data):
|
||||
sheet.append(labels)
|
||||
for row in data:
|
||||
row_data = [row[field] for field in attributes]
|
||||
sheet.append(row_data)
|
||||
|
||||
redimension_cell_width(sheet)
|
||||
|
||||
|
||||
def write_growth_media(wb, growth_media):
|
||||
ws = wb.create_sheet(GROWTH_MEDIA)
|
||||
ws.append(["Acronym", "Description", "Full description"])
|
||||
for growth_medium in growth_media:
|
||||
row = [
|
||||
growth_medium.acronym,
|
||||
growth_medium.description,
|
||||
growth_medium.full_description,
|
||||
]
|
||||
ws.append(row)
|
||||
redimension_cell_width(ws)
|
||||
|
||||
|
||||
def redimension_cell_width(ws):
|
||||
dims = {}
|
||||
for row in ws.rows:
|
||||
for cell in row:
|
||||
if cell.value:
|
||||
max_ = max((dims.get(cell.column_letter, 0), len(str(cell.value))))
|
||||
dims[cell.column_letter] = max_
|
||||
for col, value in dims.items():
|
||||
ws.column_dimensions[col].width = value
|
||||
Reference in New Issue
Block a user