Troubleshooting.

This commit is contained in:
lwark
2024-08-01 09:58:52 -05:00
parent ddf7937680
commit 88f3e92069
3 changed files with 75 additions and 39 deletions

View File

@@ -4,6 +4,7 @@ Models for the main submission and sample types.
from __future__ import annotations from __future__ import annotations
import sys import sys
import types
from copy import deepcopy from copy import deepcopy
from getpass import getuser from getpass import getuser
import logging, uuid, tempfile, re, yaml, base64 import logging, uuid, tempfile, re, yaml, base64
@@ -78,7 +79,8 @@ class BasicSubmission(BaseClass):
) #: Relation to SubmissionSampleAssociation ) #: Relation to SubmissionSampleAssociation
samples = association_proxy("submission_sample_associations", samples = association_proxy("submission_sample_associations",
"sample", creator=lambda sample: SubmissionSampleAssociation(sample=sample)) #: Association proxy to SubmissionSampleAssociation.samples "sample", creator=lambda sample: SubmissionSampleAssociation(
sample=sample)) #: Association proxy to SubmissionSampleAssociation.samples
submission_reagent_associations = relationship( submission_reagent_associations = relationship(
"SubmissionReagentAssociation", "SubmissionReagentAssociation",
@@ -735,7 +737,8 @@ class BasicSubmission(BaseClass):
return input_dict return input_dict
@classmethod @classmethod
def custom_info_writer(cls, input_excel: Workbook, info: dict | None = None, backup: bool = False, custom_fields:dict={}) -> Workbook: def custom_info_writer(cls, input_excel: Workbook, info: dict | None = None, backup: bool = False,
custom_fields: dict = {}) -> Workbook:
""" """
Adds custom autofill methods for submission Adds custom autofill methods for submission
@@ -1633,7 +1636,8 @@ class WastewaterArtic(BasicSubmission):
input_dict = super().custom_info_parser(input_dict) input_dict = super().custom_info_parser(input_dict)
egel_section = custom_fields['egel_results'] egel_section = custom_fields['egel_results']
ws = xl[egel_section['sheet']] ws = xl[egel_section['sheet']]
data = [ws.cell(row=ii, column=jj) for jj in range(egel_section['start_column'], egel_section['end_column']) for ii in range(egel_section['start_row'], egel_section['end_row'])] data = [ws.cell(row=ii, column=jj) for jj in range(egel_section['start_column'], egel_section['end_column']) for
ii in range(egel_section['start_row'], egel_section['end_row'])]
data = [cell for cell in data if cell.value is not None and "NTC" in cell.value] data = [cell for cell in data if cell.value is not None and "NTC" in cell.value]
input_dict['gel_controls'] = [ input_dict['gel_controls'] = [
dict(sample_id=cell.value, location=f"{row_map[cell.row - 9]}{str(cell.column - 14).zfill(2)}") for cell in dict(sample_id=cell.value, location=f"{row_map[cell.row - 9]}{str(cell.column - 14).zfill(2)}") for cell in
@@ -1641,7 +1645,9 @@ class WastewaterArtic(BasicSubmission):
# NOTE: Get source plate information # NOTE: Get source plate information
source_plates_section = custom_fields['source_plates'] source_plates_section = custom_fields['source_plates']
ws = xl[source_plates_section['sheet']] ws = xl[source_plates_section['sheet']]
data = [dict(plate=ws.cell(row=ii, column=source_plates_section['plate_column']).value, starting_sample=ws.cell(row=ii, column=source_plates_section['starting_sample_column']).value) for ii in data = [dict(plate=ws.cell(row=ii, column=source_plates_section['plate_column']).value,
starting_sample=ws.cell(row=ii, column=source_plates_section['starting_sample_column']).value) for
ii in
range(source_plates_section['start_row'], source_plates_section['end_row'] + 1)] range(source_plates_section['start_row'], source_plates_section['end_row'] + 1)]
for datum in data: for datum in data:
if datum['plate'] in ["None", None, ""]: if datum['plate'] in ["None", None, ""]:
@@ -1832,7 +1838,8 @@ class WastewaterArtic(BasicSubmission):
return input_dict return input_dict
@classmethod @classmethod
def custom_info_writer(cls, input_excel: Workbook, info: dict | None = None, backup: bool = False, custom_fields:dict={}) -> Workbook: def custom_info_writer(cls, input_excel: Workbook, info: dict | None = None, backup: bool = False,
custom_fields: dict = {}) -> Workbook:
""" """
Adds custom autofill methods for submission. Extends Parent Adds custom autofill methods for submission. Extends Parent
@@ -1846,11 +1853,14 @@ class WastewaterArtic(BasicSubmission):
Workbook: Updated workbook Workbook: Updated workbook
""" """
input_excel = super().custom_info_writer(input_excel, info, backup) input_excel = super().custom_info_writer(input_excel, info, backup)
# logger.debug(f"Info:\n{pformat(info)}") if isinstance(info, types.GeneratorType):
# logger.debug(f"Custom fields:\n{pformat(custom_fields)}") logger.debug(f"Unpacking info generator.")
info = {k: v for k, v in info}
logger.debug(f"Info:\n{pformat(info)}")
logger.debug(f"Custom fields:\n{pformat(custom_fields)}")
# NOTE: check for source plate information # NOTE: check for source plate information
source_plates_section = custom_fields['source_plates']
if check_key_or_attr(key='source_plates', interest=info, check_none=True): if check_key_or_attr(key='source_plates', interest=info, check_none=True):
source_plates_section = custom_fields['source_plates']
worksheet = input_excel[source_plates_section['sheet']] worksheet = input_excel[source_plates_section['sheet']]
start_row = source_plates_section['start_row'] start_row = source_plates_section['start_row']
# NOTE: write source plates to First strand list # NOTE: write source plates to First strand list
@@ -1862,12 +1872,15 @@ class WastewaterArtic(BasicSubmission):
except TypeError: except TypeError:
pass pass
try: try:
worksheet.cell(row=row, column=source_plates_section['starting_sample_column'], value=plate['starting_sample']) worksheet.cell(row=row, column=source_plates_section['starting_sample_column'],
value=plate['starting_sample'])
except TypeError: except TypeError:
pass pass
else:
logger.warning(f"No source plate info found.")
# NOTE: check for gel information # NOTE: check for gel information
egel_section = custom_fields['egel_results']
if check_key_or_attr(key='gel_info', interest=info, check_none=True): if check_key_or_attr(key='gel_info', interest=info, check_none=True):
egel_section = custom_fields['egel_results']
# logger.debug(f"Gel info check passed.") # logger.debug(f"Gel info check passed.")
# NOTE: print json field gel results to Egel results # NOTE: print json field gel results to Egel results
worksheet = input_excel[egel_section['sheet']] worksheet = input_excel[egel_section['sheet']]
@@ -1889,6 +1902,8 @@ class WastewaterArtic(BasicSubmission):
worksheet.cell(row=row, column=column, value=kj['value']) worksheet.cell(row=row, column=column, value=kj['value'])
except AttributeError: except AttributeError:
logger.error(f"Failed {kj['name']} with value {kj['value']} to row {row}, column {column}") logger.error(f"Failed {kj['name']} with value {kj['value']} to row {row}, column {column}")
else:
logger.warning("No gel info found.")
if check_key_or_attr(key='gel_image_path', interest=info, check_none=True): if check_key_or_attr(key='gel_image_path', interest=info, check_none=True):
worksheet = input_excel[egel_section['sheet']] worksheet = input_excel[egel_section['sheet']]
# logger.debug(f"We got an image: {info['gel_image']}") # logger.debug(f"We got an image: {info['gel_image']}")
@@ -1899,6 +1914,8 @@ class WastewaterArtic(BasicSubmission):
img.width = 600 img.width = 600
img.anchor = egel_section['img_anchor'] img.anchor = egel_section['img_anchor']
worksheet.add_image(img) worksheet.add_image(img)
else:
logger.warning("No gel image found.")
return input_excel return input_excel
@classmethod @classmethod

View File

@@ -173,11 +173,13 @@ class InfoWriter(object):
Returns: Returns:
Workbook: workbook with info written. Workbook: workbook with info written.
""" """
final_info = {}
for k, v in self.info: for k, v in self.info:
# NOTE: merge all comments to fit in single cell. # NOTE: merge all comments to fit in single cell.
if k == "comment" and isinstance(v['value'], list): if k == "comment" and isinstance(v['value'], list):
json_join = [item['text'] for item in v['value'] if 'text' in item.keys()] json_join = [item['text'] for item in v['value'] if 'text' in item.keys()]
v['value'] = "\n".join(json_join) v['value'] = "\n".join(json_join)
final_info[k] = v
try: try:
locations = v['locations'] locations = v['locations']
except KeyError: except KeyError:
@@ -186,8 +188,11 @@ class InfoWriter(object):
for loc in locations: for loc in locations:
logger.debug(f"Writing {k} to {loc['sheet']}, row: {loc['row']}, column: {loc['column']}") logger.debug(f"Writing {k} to {loc['sheet']}, row: {loc['row']}, column: {loc['column']}")
sheet = self.xl[loc['sheet']] sheet = self.xl[loc['sheet']]
try:
sheet.cell(row=loc['row'], column=loc['column'], value=v['value']) sheet.cell(row=loc['row'], column=loc['column'], value=v['value'])
return self.sub_object.custom_info_writer(self.xl, info=self.info, custom_fields=self.info_map['custom']) except AttributeError as e:
logger.error(f"Can't write {k} to that cell due to {e}")
return self.sub_object.custom_info_writer(self.xl, info=final_info, custom_fields=self.info_map['custom'])
class ReagentWriter(object): class ReagentWriter(object):

View File

@@ -102,7 +102,21 @@ def check_key_or_attr(key: str, interest: dict | object, check_none: bool = Fals
if check_none: if check_none:
match interest[key]: match interest[key]:
case dict(): case dict():
if interest[key]['value'] is None: if 'value' in interest[key].keys():
try:
check = interest[key]['value'] is None
except KeyError:
check = True
if check:
return False
else:
return True
else:
try:
check = interest[key] is None
except KeyError:
check = True
if check:
return False return False
else: else:
return True return True