landon/Submissions-App
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Debugged upgrades.

Browse Source
This commit is contained in:
lwark
2024-05-13 07:44:06 -05:00
parent f30f6403d6
commit 84fac23890
15 changed files with 447 additions and 487 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
CHANGELOG.md
View File

@@ -1,3 +1,7 @@
## 202405.01
- New Excel writers
## 202404.05
- Addition of default query method using Kwargs.

2
TODO.md
View File

@@ -1,4 +1,4 @@
- [ ] Convert Parsers to using openpyxl.
- [x] Convert Parsers to using openpyxl.
- The hardest part of this is going to be the sample parsing. I'm onto using the cell formulas in the plate map to suss out the location in the lookup table, but it could get a little recursive up in here.
- [ ] Create a default info return function.
- [x] Parse comment from excel sheet.

4
alembic.ini
View File

@@ -55,8 +55,8 @@ version_path_separator = os # Use os.pathsep. Default configuration used for ne
# are written from script.py.mako
# output_encoding = utf-8
;sqlalchemy.url = sqlite:///L:\Robotics Laboratory Support\Submissions\submissions.db
sqlalchemy.url = sqlite:///C:\Users\lwark\Documents\Archives\Submissions_app_backups\DB_backups\submissions-demo.db
sqlalchemy.url = sqlite:///L:\Robotics Laboratory Support\Submissions\submissions.db
;sqlalchemy.url = sqlite:///C:\Users\lwark\Documents\Archives\Submissions_app_backups\DB_backups\submissions-demo.db
;sqlalchemy.url = sqlite:///C:\Users\lwark\Documents\Archives\Submissions_app_backups\DB_backups\submissions-prototypes.db
;sqlalchemy.url = sqlite:///C:\Users\lwark\Documents\python\submissions\mytests\test_assets\submissions-test.db

2
src/submissions/backend/db/models/__init__.py
View File

@@ -116,7 +116,7 @@ class BaseClass(Base):
query: Query = cls.__database_session__.query(model)
# logger.debug(f"Grabbing singles using {model.get_default_info}")
singles = model.get_default_info('singles')
logger.debug(f"Querying: {model}, singles: {singles}")
logger.debug(f"Querying: {model}, with kwargs: {kwargs}")
for k, v in kwargs.items():
logger.debug(f"Using key: {k} with value: {v}")
# logger.debug(f"That key found attribute: {attr} with type: {attr}")

4
src/submissions/backend/db/models/kits.py
View File

@@ -629,6 +629,7 @@ class SubmissionType(BaseClass):
output = {k:v[mode] for k,v in info.items() if v[mode]}
case "write":
output = {k:v[mode] + v['read'] for k,v in info.items() if v[mode] or v['read']}
output = {k:v for k, v in output.items() if all([isinstance(item, dict) for item in v])}
return output
def construct_sample_map(self):
@@ -935,6 +936,9 @@ class SubmissionReagentAssociation(BaseClass):
return f"<{self.submission.rsl_plate_num}&{self.reagent.lot}>"
def __init__(self, reagent=None, submission=None):
if isinstance(reagent, list):
logger.warning(f"Got list for reagent. Likely no lot was provided. Using {reagent[0]}")
reagent = reagent[0]
self.reagent = reagent
self.submission = submission
self.comments = ""

1
src/submissions/backend/db/models/organizations.py
View File

@@ -67,6 +67,7 @@ class Organization(BaseClass):
case _:
pass
return cls.execute_query(query=query, limit=limit)
# return query.first()
@check_authorization
def save(self):

232
src/submissions/backend/db/models/submissions.py
View File

@@ -21,7 +21,7 @@ from openpyxl.worksheet.worksheet import Worksheet
from openpyxl.drawing.image import Image as OpenpyxlImage
from tools import check_not_nan, row_map, setup_lookup, jinja_template_loading, rreplace
from datetime import datetime, date
from typing import List, Any, Tuple
from typing import List, Any, Tuple, Literal
from dateutil.parser import parse
from dateutil.parser import ParserError
from pathlib import Path
@@ -418,11 +418,11 @@ class BasicSubmission(BaseClass):
case "samples":
for sample in value:
# logger.debug(f"Parsing {sample} to sql.")
sample, _ = sample.toSQL(submission=self)
sample, _ = sample.to_sql(submission=self)
return
case "reagents":
logger.debug(f"Reagents coming into SQL: {value}")
field_value = [reagent['value'].toSQL()[0] if isinstance(reagent, dict) else reagent.toSQL()[0] for
field_value = [reagent['value'].to_sql()[0] if isinstance(reagent, dict) else reagent.to_sql()[0] for
reagent in value]
logger.debug(f"Reagents coming out of SQL: {field_value}")
case "submission_type":
@@ -620,7 +620,7 @@ class BasicSubmission(BaseClass):
return plate_map
@classmethod
def parse_info(cls, input_dict: dict, xl: Workbook | None = None) -> dict:
def custom_info_parser(cls, input_dict: dict, xl: Workbook | None = None) -> dict:
"""
Update submission dictionary with type specific information
@@ -666,7 +666,7 @@ class BasicSubmission(BaseClass):
return input_dict
@classmethod
def custom_autofill(cls, input_excel: Workbook, info: dict | None = None, backup: bool = False) -> Workbook:
def custom_info_writer(cls, input_excel: Workbook, info: dict | None = None, backup: bool = False) -> Workbook:
"""
Adds custom autofill methods for submission
@@ -730,7 +730,9 @@ class BasicSubmission(BaseClass):
repeat = "1"
except AttributeError as e:
repeat = ""
return re.sub(r"(-\dR)\d?", rf"\1 {repeat}", outstr).replace(" ", "")
outstr = re.sub(r"(-\dR)\d?", rf"\1 {repeat}", outstr).replace(" ", "")
abb = cls.get_default_info('abbreviation')
return re.sub(rf"{abb}(\d)", rf"{abb}-\1", outstr)
# return outstr
@classmethod
@@ -1045,7 +1047,7 @@ class BasicSubmission(BaseClass):
logger.debug(widg)
widg.setParent(None)
pyd = self.to_pydantic(backup=True)
form = pyd.toForm(parent=obj)
form = pyd.to_form(parent=obj)
obj.app.table_widget.formwidget.layout().addWidget(form)
def add_comment(self, obj):
@@ -1112,7 +1114,7 @@ class BasicSubmission(BaseClass):
# wb = pyd.autofill_excel()
# wb = pyd.autofill_samples(wb)
# wb = pyd.autofill_equipment(wb)
writer = pyd.toWriter()
writer = pyd.to_writer()
# wb.save(filename=fname.with_suffix(".xlsx"))
writer.xl.save(filename=fname.with_suffix(".xlsx"))
@@ -1166,25 +1168,25 @@ class BacterialCulture(BasicSubmission):
plate_map.iloc[6, 0] = num2
return plate_map
@classmethod
def custom_autofill(cls, input_excel: Workbook, info: dict | None = None, backup: bool = False) -> Workbook:
"""
Stupid stopgap solution to there being an issue with the Bacterial Culture plate map. Extends parent.
Args:
input_excel (Workbook): Input openpyxl workbook
Returns:
Workbook: Updated openpyxl workbook
"""
input_excel = super().custom_autofill(input_excel)
sheet = input_excel['Plate Map']
if sheet.cell(12, 2).value == None:
sheet.cell(row=12, column=2, value="=IF(ISBLANK('Sample List'!$B42),\"\",'Sample List'!$B42)")
if sheet.cell(13, 2).value == None:
sheet.cell(row=13, column=2, value="=IF(ISBLANK('Sample List'!$B43),\"\",'Sample List'!$B43)")
input_excel["Sample List"].cell(row=15, column=2, value=getuser())
return input_excel
# @classmethod
# def custom_writer(cls, input_excel: Workbook, info: dict | None = None, backup: bool = False) -> Workbook:
# """
# Stupid stopgap solution to there being an issue with the Bacterial Culture plate map. Extends parent.
#
# Args:
# input_excel (Workbook): Input openpyxl workbook
#
# Returns:
# Workbook: Updated openpyxl workbook
# """
# input_excel = super().custom_writer(input_excel)
# sheet = input_excel['Plate Map']
# if sheet.cell(12, 2).value == None:
# sheet.cell(row=12, column=2, value="=IF(ISBLANK('Sample List'!$B42),\"\",'Sample List'!$B42)")
# if sheet.cell(13, 2).value == None:
# sheet.cell(row=13, column=2, value="=IF(ISBLANK('Sample List'!$B43),\"\",'Sample List'!$B43)")
# input_excel["Sample List"].cell(row=15, column=2, value=getuser())
# return input_excel
@classmethod
def get_regex(cls) -> str:
@@ -1297,7 +1299,7 @@ class Wastewater(BasicSubmission):
return output
@classmethod
def parse_info(cls, input_dict: dict, xl: Workbook | None = None) -> dict:
def custom_info_parser(cls, input_dict: dict, xl: Workbook | None = None) -> dict:
"""
Update submission dictionary with type specific information. Extends parent
@@ -1307,7 +1309,7 @@ class Wastewater(BasicSubmission):
Returns:
dict: Updated sample dictionary
"""
input_dict = super().parse_info(input_dict)
input_dict = super().custom_info_parser(input_dict)
if xl != None:
input_dict['csv'] = xl["Copy to import file"]
return input_dict
@@ -1458,7 +1460,7 @@ class WastewaterArtic(BasicSubmission):
return output
@classmethod
def parse_info(cls, input_dict: dict, xl: pd.ExcelFile | None = None) -> dict:
def custom_info_parser(cls, input_dict: dict, xl: Workbook | None = None) -> dict:
"""
Update submission dictionary with type specific information
@@ -1469,17 +1471,23 @@ class WastewaterArtic(BasicSubmission):
Returns:
dict: Updated sample dictionary
"""
input_dict = super().parse_info(input_dict)
workbook = load_workbook(xl.io, data_only=True)
ws = workbook['Egel results']
input_dict = super().custom_info_parser(input_dict)
# workbook = load_workbook(xl.io, data_only=True)
ws = xl['Egel results']
data = [ws.cell(row=ii, column=jj) for jj in range(15, 27) for ii in range(10, 18)]
data = [cell for cell in data if cell.value is not None and "NTC" in cell.value]
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
data]
ws = workbook['First Strand List']
ws = xl['First Strand List']
data = [dict(plate=ws.cell(row=ii, column=3).value, starting_sample=ws.cell(row=ii, column=4).value) for ii in
range(8, 11)]
for datum in data:
if datum['plate'] in ["None", None, ""]:
continue
else:
from backend.validators import RSLNamer
datum['plate'] = RSLNamer(filename=datum['plate'], sub_type="Wastewater").parsed_name
input_dict['source_plates'] = data
return input_dict
@@ -1493,10 +1501,13 @@ class WastewaterArtic(BasicSubmission):
instr = re.sub(r"Artic", "", instr, flags=re.IGNORECASE)
except (AttributeError, TypeError) as e:
logger.error(f"Problem using regex: {e}")
# logger.debug(f"Before RSL addition: {instr}")
logger.debug(f"Before RSL addition: {instr}")
try:
instr = instr.replace("-", "")
except AttributeError:
instr = date.today().strftime("%Y%m%d")
instr = re.sub(r"^(\d{6})", f"RSL-AR-\\1", instr)
# logger.debug(f"name coming out of Artic namer: {instr}")
logger.debug(f"name coming out of Artic namer: {instr}")
outstr = super().enforce_name(instr=instr, data=data)
return outstr
@@ -1527,8 +1538,12 @@ class WastewaterArtic(BasicSubmission):
del input_dict['sample_name_(ww)']
except KeyError:
logger.error(f"Unable to set ww_processing_num for sample {input_dict['submitter_id']}")
if "ENC" in input_dict['submitter_id']:
input_dict['submitter_id'] = cls.en_adapter(input_str=input_dict['submitter_id'])
year = str(date.today().year)[-2:]
# if "ENC" in input_dict['submitter_id']:
if re.search(rf"^{year}-(ENC)", input_dict['submitter_id']):
input_dict['rsl_number'] = cls.en_adapter(input_str=input_dict['submitter_id'])
if re.search(rf"^{year}-(RSL)", input_dict['submitter_id']):
input_dict['rsl_number'] = cls.pbs_adapter(input_str=input_dict['submitter_id'])
return input_dict
@classmethod
@@ -1544,7 +1559,9 @@ class WastewaterArtic(BasicSubmission):
"""
logger.debug(f"input string raw: {input_str}")
# Remove letters.
processed = re.sub(r"[A-QS-Z]+\d*", "", input_str)
processed = input_str.replace("RSL", "")
processed = re.sub(r"\(.*\)$", "", processed).strip()
processed = re.sub(r"[A-QS-Z]+\d*", "", processed)
# Remove trailing '-' if any
processed = processed.strip("-")
logger.debug(f"Processed after stripping letters: {processed}")
@@ -1554,7 +1571,7 @@ class WastewaterArtic(BasicSubmission):
except AttributeError:
en_num = "1"
en_num = en_num.strip("-")
logger.debug(f"Processed after en-num: {processed}")
logger.debug(f"Processed after en_num: {processed}")
try:
plate_num = re.search(r"\-\d{1}R?\d?$", processed).group()
processed = rreplace(processed, plate_num, "")
@@ -1571,7 +1588,58 @@ class WastewaterArtic(BasicSubmission):
logger.debug(f"Processed after month: {processed}")
year = re.search(r'^(?:\d{2})?\d{2}', processed).group()
year = f"20{year}"
final_en_name = f"EN{year}{month}{day}-{en_num}"
final_en_name = f"EN{en_num}-{year}{month}{day}"
logger.debug(f"Final EN name: {final_en_name}")
return final_en_name
@classmethod
def pbs_adapter(cls, input_str):
"""
Stopgap solution because WW names their ENs different
Args:
input_str (str): input name
Returns:
str: output name
"""
logger.debug(f"input string raw: {input_str}")
# Remove letters.
processed = input_str.replace("RSL", "")
processed = re.sub(r"\(.*\)$", "", processed).strip()
processed = re.sub(r"[A-QS-Z]+\d*", "", processed)
# Remove trailing '-' if any
processed = processed.strip("-")
logger.debug(f"Processed after stripping letters: {processed}")
# try:
# en_num = re.search(r"\-\d{1}$", processed).group()
# processed = rreplace(processed, en_num, "")
# except AttributeError:
# en_num = "1"
# en_num = en_num.strip("-")
# logger.debug(f"Processed after en_num: {processed}")
try:
plate_num = re.search(r"\-\d{1}R?\d?$", processed).group()
processed = rreplace(processed, plate_num, "")
except AttributeError:
plate_num = "1"
plate_num = plate_num.strip("-")
logger.debug(f"Plate num: {plate_num}")
repeat_num = re.search(r"R(?P<repeat>\d)?$", "PBS20240426-2R").groups()[0]
if repeat_num is None and "R" in plate_num:
repeat_num = "1"
plate_num = re.sub(r"R", rf"R{repeat_num}", plate_num)
logger.debug(f"Processed after plate-num: {processed}")
day = re.search(r"\d{2}$", processed).group()
processed = rreplace(processed, day, "")
logger.debug(f"Processed after day: {processed}")
month = re.search(r"\d{2}$", processed).group()
processed = rreplace(processed, month, "")
processed = processed.replace("--", "")
logger.debug(f"Processed after month: {processed}")
year = re.search(r'^(?:\d{2})?\d{2}', processed).group()
year = f"20{year}"
final_en_name = f"PBS{year}{month}{day}-{plate_num}"
logger.debug(f"Final EN name: {final_en_name}")
return final_en_name
@@ -1600,11 +1668,17 @@ class WastewaterArtic(BasicSubmission):
dict: Updated parser product.
"""
input_dict = super().finalize_parse(input_dict, xl, info_map)
input_dict['csv'] = xl.parse("hitpicks_csv_to_export")
logger.debug(f"Incoming input_dict: {pformat(input_dict)}")
# TODO: Move to validator?
for sample in input_dict['samples']:
logger.debug(f"Sample: {sample}")
if re.search(r"^NTC", sample['submitter_id']):
sample['submitter_id'] = sample['submitter_id'] + "-WWG-" + input_dict['rsl_plate_num']['value']
input_dict['csv'] = xl["hitpicks_csv_to_export"]
return input_dict
@classmethod
def custom_autofill(cls, input_excel: Workbook, info: dict | None = None, backup: bool = False) -> Workbook:
def custom_info_writer(cls, input_excel: Workbook, info: dict | None = None, backup: bool = False) -> Workbook:
"""
Adds custom autofill methods for submission. Extends Parent
@@ -1616,10 +1690,10 @@ class WastewaterArtic(BasicSubmission):
Returns:
Workbook: Updated workbook
"""
input_excel = super().custom_autofill(input_excel, info, backup)
worksheet = input_excel["First Strand List"]
samples = cls.query(rsl_number=info['rsl_plate_num']['value']).submission_sample_associations
samples = sorted(samples, key=attrgetter('column', 'row'))
input_excel = super().custom_info_writer(input_excel, info, backup)
# worksheet = input_excel["First Strand List"]
# samples = cls.query(rsl_number=info['rsl_plate_num']['value']).submission_sample_associations
# samples = sorted(samples, key=attrgetter('column', 'row'))
logger.debug(f"Info:\n{pformat(info)}")
check = 'source_plates' in info.keys() and info['source_plates'] is not None
if check:
@@ -1708,15 +1782,22 @@ class WastewaterArtic(BasicSubmission):
"""
logger.debug(f"Hello from {self.__class__.__name__} dictionary sample adjuster.")
output = []
set_plate = None
for assoc in self.submission_sample_associations:
dicto = assoc.to_sub_dict()
old_sub = assoc.sample.get_previous_ww_submission(current_artic_submission=self)
try:
dicto['plate_name'] = old_sub.rsl_plate_num
except AttributeError:
dicto['plate_name'] = ""
old_assoc = WastewaterAssociation.query(submission=old_sub, sample=assoc.sample, limit=1)
dicto['well'] = f"{row_map[old_assoc.row]}{old_assoc.column}"
# old_sub = assoc.sample.get_previous_ww_submission(current_artic_submission=self)
# try:
# dicto['plate_name'] = old_sub.rsl_plate_num
# except AttributeError:
# dicto['plate_name'] = ""
# old_assoc = WastewaterAssociation.query(submission=old_sub, sample=assoc.sample, limit=1)
# dicto['well'] = f"{row_map[old_assoc.row]}{old_assoc.column}"
for item in self.source_plates:
old_plate = WastewaterAssociation.query(submission=item['plate'], sample=assoc.sample, limit=1)
if old_plate is not None:
set_plate = old_plate.submission.rsl_plate_num
break
dicto['plate_name'] = set_plate
output.append(dicto)
return output
@@ -1892,6 +1973,10 @@ class BasicSample(BaseClass):
logger.info(f"Recruiting model: {model}")
return model
@classmethod
def sql_enforcer(cls, pyd_sample:"PydSample"):
return pyd_sample
@classmethod
def parse_sample(cls, input_dict: dict) -> dict:
f"""
@@ -1996,15 +2081,16 @@ class BasicSample(BaseClass):
disallowed = ["id"]
if kwargs == {}:
raise ValueError("Need to narrow down query or the first available instance will be returned.")
for key in kwargs.keys():
if key in disallowed:
raise ValueError(
f"{key} is not allowed as a query argument as it could lead to creation of duplicate objects.")
# for key in kwargs.keys():
# if key in disallowed:
# raise ValueError(
# f"{key} is not allowed as a query argument as it could lead to creation of duplicate objects.")
sanitized_kwargs = {k:v for k,v in kwargs.items() if k not in disallowed}
instance = cls.query(sample_type=sample_type, limit=1, **kwargs)
logger.debug(f"Retrieved instance: {instance}")
if instance == None:
used_class = cls.find_polymorphic_subclass(attrs=kwargs, polymorphic_identity=sample_type)
instance = used_class(**kwargs)
if instance is None:
used_class = cls.find_polymorphic_subclass(attrs=sanitized_kwargs, polymorphic_identity=sample_type)
instance = used_class(**sanitized_kwargs)
instance.sample_type = sample_type
logger.debug(f"Creating instance: {instance}")
return instance
@@ -2061,7 +2147,7 @@ class WastewaterSample(BasicSample):
dict: well location and name (sample id, organism) NOTE: keys must sync with WWSample to_sub_dict above
"""
sample = super().to_sub_dict(full_data=full_data)
sample['WW Processing Number'] = self.ww_processing_num
sample['WW Processing Num'] = self.ww_processing_num
sample['Sample Location'] = self.sample_location
sample['Received Date'] = self.received_date
sample['Collection Date'] = self.collection_date
@@ -2080,14 +2166,17 @@ class WastewaterSample(BasicSample):
"""
output_dict = super().parse_sample(input_dict)
logger.debug(f"Initial sample dict: {pformat(output_dict)}")
disallowed = ["", None, "None"]
try:
check = output_dict['rsl_number'] in [None, "None"]
except KeyError:
check = True
if check:
output_dict['rsl_number'] = "RSL-WW-" + output_dict['ww_processing_number']
if output_dict['ww_full_sample_id'] is not None:
output_dict['rsl_number'] = "RSL-WW-" + output_dict['ww_processing_num']
if output_dict['ww_full_sample_id'] is not None and output_dict["submitter_id"] in disallowed:
output_dict["submitter_id"] = output_dict['ww_full_sample_id']
# if re.search(r"^NTC", output_dict['submitter_id']):
# output_dict['submitter_id'] = "Artic-" + output_dict['submitter_id']
# Ad hoc repair method for WW (or possibly upstream) not formatting some dates properly.
# NOTE: Should be handled by validator.
# match output_dict['collection_date']:
@@ -2166,7 +2255,7 @@ class SubmissionSampleAssociation(BaseClass):
submission_id = Column(INTEGER, ForeignKey("_basicsubmission.id"), primary_key=True) #: id of associated submission
row = Column(INTEGER, primary_key=True) #: row on the 96 well plate
column = Column(INTEGER, primary_key=True) #: column on the 96 well plate
submission_rank = Column(INTEGER, nullable=False, default=1) #: Location in sample list
submission_rank = Column(INTEGER, nullable=False, default=0) #: Location in sample list
# reference to the Submission object
submission = relationship(BasicSubmission,
@@ -2186,12 +2275,13 @@ class SubmissionSampleAssociation(BaseClass):
}
def __init__(self, submission: BasicSubmission = None, sample: BasicSample = None, row: int = 1, column: int = 1,
id: int | None = None):
id: int | None = None, submission_rank: int = 0):
self.submission = submission
self.sample = sample
self.row = row
self.column = column
if id != None:
self.submission_rank = submission_rank
if id is not None:
self.id = id
else:
self.id = self.__class__.autoincrement_id()
@@ -2257,6 +2347,7 @@ class SubmissionSampleAssociation(BaseClass):
Returns:
int: incremented id
"""
try:
return max([item.id for item in cls.query()]) + 1
except ValueError as e:
@@ -2360,7 +2451,7 @@ class SubmissionSampleAssociation(BaseClass):
association_type: str = "Basic Association",
submission: BasicSubmission | str | None = None,
sample: BasicSample | str | None = None,
# id:int|None=None,
id:int|None=None,
**kwargs) -> SubmissionSampleAssociation:
"""
Queries for an association, if none exists creates a new one.
@@ -2401,10 +2492,11 @@ class SubmissionSampleAssociation(BaseClass):
instance = cls.query(submission=submission, sample=sample, row=row, column=column, limit=1)
except StatementError:
instance = None
if instance == None:
if instance is None:
# sanitized_kwargs = {k:v for k,v in kwargs.items() if k not in ['id']}
used_cls = cls.find_polymorphic_subclass(polymorphic_identity=association_type)
# instance = used_cls(submission=submission, sample=sample, id=id, **kwargs)
instance = used_cls(submission=submission, sample=sample, **kwargs)
instance = used_cls(submission=submission, sample=sample, id=id, **kwargs)
return instance
def delete(self):

217
src/submissions/backend/excel/parser.py
View File

@@ -45,7 +45,7 @@ class SheetParser(object):
raise ValueError("No filepath given.")
try:
# self.xl = pd.ExcelFile(filepath)
self.xl = load_workbook(filepath, read_only=True, data_only=True)
self.xl = load_workbook(filepath, data_only=True)
except ValueError as e:
logger.error(f"Incorrect value: {e}")
raise FileNotFoundError(f"Couldn't parse file {self.filepath}")
@@ -53,6 +53,8 @@ class SheetParser(object):
# make decision about type of sample we have
self.sub['submission_type'] = dict(value=RSLNamer.retrieve_submission_type(filename=self.filepath),
missing=True)
self.submission_type = SubmissionType.query(name=self.sub['submission_type'])
self.sub_object = BasicSubmission.find_polymorphic_subclass(polymorphic_identity=self.submission_type)
# grab the info map from the submission type in database
self.parse_info()
self.import_kit_validation_check()
@@ -67,7 +69,7 @@ class SheetParser(object):
"""
Pulls basic information from the excel sheet
"""
parser = InfoParser(xl=self.xl, submission_type=self.sub['submission_type']['value'])
parser = InfoParser(xl=self.xl, submission_type=self.submission_type, sub_object=self.sub_object)
info = parser.parse_info()
self.info_map = parser.map
# exclude_from_info = BasicSubmission.find_polymorphic_subclass(polymorphic_identity=self.sub['submission_type']).exclude_from_info_parser()
@@ -87,21 +89,21 @@ class SheetParser(object):
extraction_kit (str | None, optional): Relevant extraction kit for reagent map. Defaults to None.
"""
if extraction_kit == None:
extraction_kit = extraction_kit = self.sub['extraction_kit']
extraction_kit = self.sub['extraction_kit']
# logger.debug(f"Parsing reagents for {extraction_kit}")
self.sub['reagents'] = ReagentParser(xl=self.xl, submission_type=self.sub['submission_type'],
self.sub['reagents'] = ReagentParser(xl=self.xl, submission_type=self.submission_type,
extraction_kit=extraction_kit).parse_reagents()
def parse_samples(self):
"""
Pulls sample info from the excel sheet
"""
parser = SampleParser(xl=self.xl, submission_type=self.sub['submission_type']['value'])
parser = SampleParser(xl=self.xl, submission_type=self.submission_type)
self.sub['samples'] = parser.reconcile_samples()
# self.plate_map = parser.plate_map
def parse_equipment(self):
parser = EquipmentParser(xl=self.xl, submission_type=self.sub['submission_type']['value'])
parser = EquipmentParser(xl=self.xl, submission_type=self.submission_type)
self.sub['equipment'] = parser.parse_equipment()
def import_kit_validation_check(self):
@@ -120,22 +122,13 @@ class SheetParser(object):
if isinstance(self.sub['extraction_kit'], str):
self.sub['extraction_kit'] = dict(value=self.sub['extraction_kit'], missing=True)
def import_reagent_validation_check(self):
"""
Enforce that only allowed reagents get into the Pydantic Model
"""
kit = KitType.query(name=self.sub['extraction_kit']['value'])
allowed_reagents = [item.name for item in kit.get_reagents()]
# logger.debug(f"List of reagents for comparison with allowed_reagents: {pformat(self.sub['reagents'])}")
self.sub['reagents'] = [reagent for reagent in self.sub['reagents'] if reagent.type in allowed_reagents]
def finalize_parse(self):
"""
Run custom final validations of data for submission subclasses.
"""
finisher = BasicSubmission.find_polymorphic_subclass(
polymorphic_identity=self.sub['submission_type']).finalize_parse
self.sub = finisher(input_dict=self.sub, xl=self.xl, info_map=self.info_map)
# finisher = BasicSubmission.find_polymorphic_subclass(
# polymorphic_identity=self.sub['submission_type']).finalize_parse
self.sub = self.sub_object.finalize_parse(input_dict=self.sub, xl=self.xl, info_map=self.info_map)
def to_pydantic(self) -> PydSubmission:
"""
@@ -163,9 +156,14 @@ class SheetParser(object):
class InfoParser(object):
def __init__(self, xl: Workbook, submission_type: str):
def __init__(self, xl: Workbook, submission_type: str|SubmissionType, sub_object: BasicSubmission|None=None):
logger.info(f"\n\Hello from InfoParser!\n\n")
self.submission_type = submission_type
if isinstance(submission_type, str):
submission_type = SubmissionType.query(name=submission_type)
if sub_object is None:
sub_object = BasicSubmission.find_polymorphic_subclass(polymorphic_identity=submission_type.name)
self.submission_type_obj = submission_type
self.sub_object = sub_object
self.map = self.fetch_submission_info_map()
self.xl = xl
logger.debug(f"Info map for InfoParser: {pformat(self.map)}")
@@ -180,16 +178,14 @@ class InfoParser(object):
Returns:
dict: Location map of all info for this submission type
"""
if isinstance(self.submission_type, str):
self.submission_type = dict(value=self.submission_type, missing=True)
self.submission_type = dict(value=self.submission_type_obj.name, missing=True)
logger.debug(f"Looking up submission type: {self.submission_type['value']}")
# submission_type = SubmissionType.query(name=self.submission_type['value'])
# info_map = submission_type.info_map
self.sub_object: BasicSubmission = \
BasicSubmission.find_polymorphic_subclass(polymorphic_identity=self.submission_type['value'])
# self.sub_object: BasicSubmission = \
# BasicSubmission.find_polymorphic_subclass(polymorphic_identity=self.submission_type['value'])
info_map = self.sub_object.construct_info_map("read")
# Get the parse_info method from the submission type specified
return info_map
def parse_info(self) -> dict:
@@ -199,8 +195,8 @@ class InfoParser(object):
Returns:
dict: key:value of basic info
"""
if isinstance(self.submission_type, str):
self.submission_type = dict(value=self.submission_type, missing=True)
# if isinstance(self.submission_type, str):
# self.submission_type = dict(value=self.submission_type, missing=True)
dicto = {}
# exclude_from_generic = BasicSubmission.find_polymorphic_subclass(polymorphic_identity=self.submission_type['value']).get_default_info("parser_ignore")
# This loop parses generic info
@@ -224,7 +220,13 @@ class InfoParser(object):
# if check:
# relevant[k] = v
for location in v:
if location['sheet'] == sheet:
try:
check = location['sheet'] == sheet
except TypeError:
logger.warning(f"Location is likely a string, skipping")
dicto[k] = dict(value=location, missing=False)
check = False
if check:
new = location
new['name'] = k
relevant.append(new)
@@ -257,13 +259,18 @@ class InfoParser(object):
dicto[item['name']] = dict(value=value, missing=missing)
except (KeyError, IndexError):
continue
return self.sub_object.parse_info(input_dict=dicto, xl=self.xl)
return self.sub_object.custom_info_parser(input_dict=dicto, xl=self.xl)
class ReagentParser(object):
def __init__(self, xl: Workbook, submission_type: str, extraction_kit: str):
def __init__(self, xl: Workbook, submission_type: str, extraction_kit: str, sub_object:BasicSubmission|None=None):
logger.debug("\n\nHello from ReagentParser!\n\n")
self.submission_type_obj = submission_type
self.sub_object = sub_object
if isinstance(extraction_kit, dict):
extraction_kit = extraction_kit['value']
self.kit_object = KitType.query(name=extraction_kit)
self.map = self.fetch_kit_info_map(extraction_kit=extraction_kit, submission_type=submission_type)
logger.debug(f"Reagent Parser map: {self.map}")
self.xl = xl
@@ -279,13 +286,14 @@ class ReagentParser(object):
Returns:
dict: locations of reagent info for the kit.
"""
if isinstance(extraction_kit, dict):
extraction_kit = extraction_kit['value']
kit = KitType.query(name=extraction_kit)
if isinstance(submission_type, dict):
submission_type = submission_type['value']
reagent_map = kit.construct_xl_map_for_use(submission_type.title())
reagent_map = self.kit_object.construct_xl_map_for_use(submission_type)
try:
del reagent_map['info']
except KeyError:
pass
return reagent_map
def parse_reagents(self) -> List[PydReagent]:
@@ -348,7 +356,7 @@ class SampleParser(object):
object to pull data for samples in excel sheet and construct individual sample objects
"""
def __init__(self, xl: Workbook, submission_type: str, sample_map: dict | None = None) -> None:
def __init__(self, xl: Workbook, submission_type: SubmissionType, sample_map: dict | None = None, sub_object:BasicSubmission|None=None) -> None:
"""
convert sample sub-dataframe to dictionary of records
@@ -359,7 +367,11 @@ class SampleParser(object):
logger.debug("\n\nHello from SampleParser!\n\n")
self.samples = []
self.xl = xl
self.submission_type = submission_type
self.submission_type = submission_type.name
self.submission_type_obj = submission_type
if sub_object is None:
sub_object = BasicSubmission.find_polymorphic_subclass(polymorphic_identity=self.submission_type_obj.name)
self.sub_object = sub_object
self.sample_info_map = self.fetch_sample_info_map(submission_type=submission_type, sample_map=sample_map)
logger.debug(f"sample_info_map: {self.sample_info_map}")
# self.plate_map = self.construct_plate_map(plate_map_location=sample_info_map['plate_map'])
@@ -385,9 +397,10 @@ class SampleParser(object):
"""
logger.debug(f"Looking up submission type: {submission_type}")
# submission_type = SubmissionType.query(name=submission_type)
self.sub_object = BasicSubmission.find_polymorphic_subclass(polymorphic_identity=submission_type)
# self.sub_object = BasicSubmission.find_polymorphic_subclass(polymorphic_identity=submission_type)
# self.custom_sub_parser = .parse_samples
self.samp_object = BasicSample.find_polymorphic_subclass(polymorphic_identity=f"{submission_type} Sample")
self.sample_type = self.sub_object.get_default_info("sample_type")
self.samp_object = BasicSample.find_polymorphic_subclass(polymorphic_identity=self.sample_type)
logger.debug(f"Got sample class: {self.samp_object.__name__}")
# self.custom_sample_parser = .parse_sample
# logger.debug(f"info_map: {pformat(se)}")
@@ -398,46 +411,46 @@ class SampleParser(object):
sample_info_map = sample_map
return sample_info_map
def construct_plate_map(self, plate_map_location: dict) -> pd.DataFrame:
"""
Gets location of samples from plate map grid in excel sheet.
Args:
plate_map_location (dict): sheet name, start/end row/column
Returns:
pd.DataFrame: Plate map grid
"""
logger.debug(f"Plate map location: {plate_map_location}")
df = self.xl.parse(plate_map_location['sheet'], header=None, dtype=object)
df = df.iloc[plate_map_location['start_row'] - 1:plate_map_location['end_row'],
plate_map_location['start_column'] - 1:plate_map_location['end_column']]
df = pd.DataFrame(df.values[1:], columns=df.iloc[0])
df = df.set_index(df.columns[0])
logger.debug(f"Vanilla platemap: {df}")
# custom_mapper = BasicSubmission.find_polymorphic_subclass(polymorphic_identity=self.submission_type)
df = self.sub_object.custom_platemap(self.xl, df)
# logger.debug(f"Custom platemap:\n{df}")
return df
def construct_lookup_table(self, lookup_table_location: dict) -> pd.DataFrame:
"""
Gets table of misc information from excel book
Args:
lookup_table_location (dict): sheet name, start/end row
Returns:
pd.DataFrame: _description_
"""
try:
df = self.xl.parse(lookup_table_location['sheet'], header=None, dtype=object)
except KeyError:
return None
df = df.iloc[lookup_table_location['start_row'] - 1:lookup_table_location['end_row']]
df = pd.DataFrame(df.values[1:], columns=df.iloc[0])
df = df.reset_index(drop=True)
return df
# def construct_plate_map(self, plate_map_location: dict) -> pd.DataFrame:
# """
# Gets location of samples from plate map grid in excel sheet.
#
# Args:
# plate_map_location (dict): sheet name, start/end row/column
#
# Returns:
# pd.DataFrame: Plate map grid
# """
# logger.debug(f"Plate map location: {plate_map_location}")
# df = self.xl.parse(plate_map_location['sheet'], header=None, dtype=object)
# df = df.iloc[plate_map_location['start_row'] - 1:plate_map_location['end_row'],
# plate_map_location['start_column'] - 1:plate_map_location['end_column']]
# df = pd.DataFrame(df.values[1:], columns=df.iloc[0])
# df = df.set_index(df.columns[0])
# logger.debug(f"Vanilla platemap: {df}")
# # custom_mapper = BasicSubmission.find_polymorphic_subclass(polymorphic_identity=self.submission_type)
# df = self.sub_object.custom_platemap(self.xl, df)
# # logger.debug(f"Custom platemap:\n{df}")
# return df
#
# def construct_lookup_table(self, lookup_table_location: dict) -> pd.DataFrame:
# """
# Gets table of misc information from excel book
#
# Args:
# lookup_table_location (dict): sheet name, start/end row
#
# Returns:
# pd.DataFrame: _description_
# """
# try:
# df = self.xl.parse(lookup_table_location['sheet'], header=None, dtype=object)
# except KeyError:
# return None
# df = df.iloc[lookup_table_location['start_row'] - 1:lookup_table_location['end_row']]
# df = pd.DataFrame(df.values[1:], columns=df.iloc[0])
# df = df.reset_index(drop=True)
# return df
def parse_plate_map(self):
"""
@@ -471,7 +484,7 @@ class SampleParser(object):
if check_not_nan(id):
if id not in invalids:
sample_dict = dict(id=id, row=ii, column=jj)
sample_dict['sample_type'] = f"{self.submission_type} Sample"
sample_dict['sample_type'] = self.sample_type
plate_map_samples.append(sample_dict)
else:
# logger.error(f"Sample cell ({row}, {column}) has invalid value: {id}.")
@@ -524,7 +537,7 @@ class SampleParser(object):
row_dict[lmap['merge_on_id']] = str(row_dict[lmap['merge_on_id']])
except KeyError:
pass
row_dict['sample_type'] = f"{self.submission_type} Sample"
row_dict['sample_type'] = self.sample_type
row_dict['submission_rank'] = ii
try:
check = check_not_nan(row_dict[lmap['merge_on_id']])
@@ -567,22 +580,22 @@ class SampleParser(object):
new_samples.append(PydSample(**translated_dict))
return result, new_samples
def grab_plates(self) -> List[str]:
"""
Parse plate names from
Returns:
List[str]: list of plate names.
"""
plates = []
for plate in self.plates:
df = self.xl.parse(plate['sheet'], header=None)
if isinstance(df.iat[plate['row'] - 1, plate['column'] - 1], str):
output = RSLNamer.retrieve_rsl_number(filename=df.iat[plate['row'] - 1, plate['column'] - 1])
else:
continue
plates.append(output)
return plates
# def grab_plates(self) -> List[str]:
# """
# Parse plate names from
#
# Returns:
# List[str]: list of plate names.
# """
# plates = []
# for plate in self.plates:
# df = self.xl.parse(plate['sheet'], header=None)
# if isinstance(df.iat[plate['row'] - 1, plate['column'] - 1], str):
# output = RSLNamer.retrieve_rsl_number(filename=df.iat[plate['row'] - 1, plate['column'] - 1])
# else:
# continue
# plates.append(output)
# return plates
def reconcile_samples(self):
# TODO: Move to pydantic validator?
@@ -630,20 +643,24 @@ class SampleParser(object):
else:
new = psample
# samples.append(psample)
new['sample_type'] = f"{self.submission_type} Sample"
# new['sample_type'] = f"{self.submission_type} Sample"
try:
check = new['submitter_id'] is None
except KeyError:
check = True
if check:
new['submitter_id'] = psample['id']
new = self.sub_object.parse_samples(new)
samples.append(new)
samples = remove_key_from_list_of_dicts(samples, "id")
return sorted(samples, key=lambda k: (k['row'], k['column']))
class EquipmentParser(object):
def __init__(self, xl: Workbook, submission_type: str) -> None:
def __init__(self, xl: Workbook, submission_type: str|SubmissionType) -> None:
if isinstance(submission_type, str):
submission_type = SubmissionType.query(name=submission_type)
self.submission_type = submission_type
self.xl = xl
self.map = self.fetch_equipment_map()
@@ -655,8 +672,8 @@ class EquipmentParser(object):
Returns:
List[dict]: List of locations
"""
submission_type = SubmissionType.query(name=self.submission_type)
return submission_type.construct_equipment_map()
# submission_type = SubmissionType.query(name=self.submission_type)
return self.submission_type.construct_equipment_map()
def get_asset_number(self, input: str) -> str:
"""

41
src/submissions/backend/excel/writer.py
View File

@@ -1,11 +1,12 @@
import logging
from copy import copy
from pathlib import Path
from pprint import pformat
from typing import List
from openpyxl import load_workbook, Workbook
from tools import row_keys
from backend.db.models import SubmissionType, KitType
from backend.db.models import SubmissionType, KitType, BasicSample, BasicSubmission
from backend.validators.pydant import PydSubmission
from io import BytesIO
from collections import OrderedDict
@@ -32,6 +33,7 @@ class SheetWriter(object):
# self.__setattr__('submission_type', submission.submission_type['value'])
self.sub[k] = v['value']
self.submission_type = SubmissionType.query(name=v['value'])
self.sub_object = BasicSubmission.find_polymorphic_subclass(polymorphic_identity=self.submission_type)
case _:
if isinstance(v, dict):
self.sub[k] = v['value']
@@ -82,13 +84,17 @@ class SheetWriter(object):
class InfoWriter(object):
def __init__(self, xl: Workbook, submission_type: SubmissionType | str, info_dict: dict):
def __init__(self, xl: Workbook, submission_type: SubmissionType | str, info_dict: dict, sub_object:BasicSubmission|None=None):
if isinstance(submission_type, str):
submission_type = SubmissionType.query(name=submission_type)
if sub_object is None:
sub_object = BasicSubmission.find_polymorphic_subclass(polymorphic_identity=submission_type.name)
self.submission_type = submission_type
self.sub_object = sub_object
self.xl = xl
map = submission_type.construct_info_map(mode='write')
self.info = self.reconcile_map(info_dict, map)
logger.debug(pformat(self.info))
def reconcile_map(self, info_dict: dict, map: dict) -> dict:
output = {}
@@ -99,7 +105,8 @@ class InfoWriter(object):
try:
dicto['locations'] = map[k]
except KeyError:
continue
# continue
pass
dicto['value'] = v
if len(dicto) > 0:
output[k] = dicto
@@ -113,10 +120,11 @@ class InfoWriter(object):
logger.error(f"No locations for {k}, skipping")
continue
for loc in locations:
logger.debug(f"Writing {k} to {loc['sheet']}, row: {loc['row']}, column: {loc['column']}")
sheet = self.xl[loc['sheet']]
sheet.cell(row=loc['row'], column=loc['column'], value=v['value'])
return self.xl
return self.sub_object.custom_info_writer(self.xl, info=self.info)
class ReagentWriter(object):
@@ -143,7 +151,7 @@ class ReagentWriter(object):
try:
dicto = dict(value=v, row=mp_info[k]['row'], column=mp_info[k]['column'])
except KeyError as e:
logger.error(f"Keyerror: {e}")
# logger.error(f"Keyerror: {e}")
dicto = v
placeholder[k] = dicto
placeholder['sheet'] = mp_info['sheet']
@@ -156,8 +164,8 @@ class ReagentWriter(object):
for k, v in reagent.items():
if not isinstance(v, dict):
continue
logger.debug(
f"Writing {reagent['type']} {k} to {reagent['sheet']}, row: {v['row']}, column: {v['column']}")
# logger.debug(
# f"Writing {reagent['type']} {k} to {reagent['sheet']}, row: {v['row']}, column: {v['column']}")
sheet.cell(row=v['row'], column=v['column'], value=v['value'])
return self.xl
@@ -214,18 +222,27 @@ class EquipmentWriter(object):
output = []
for ii, equipment in enumerate(equipment_list, start=1):
mp_info = map[equipment['role']]
# logger.debug(f"{equipment['role']} map: {mp_info}")
placeholder = copy(equipment)
if mp_info == {}:
for jj, (k, v) in enumerate(equipment.items(), start=1):
dicto = dict(value=v, row=ii, column=jj)
placeholder[k] = dicto
# output.append(placeholder)
else:
for jj, (k, v) in enumerate(equipment.items(), start=1):
try:
dicto = dict(value=v, row=mp_info[k]['row'], column=mp_info[k]['column'])
except KeyError as e:
logger.error(f"Keyerror: {e}")
dicto = dict(value=v, row=ii, column=jj)
# logger.error(f"Keyerror: {e}")
continue
placeholder[k] = dicto
try:
placeholder['sheet'] = mp_info['sheet']
except KeyError:
placeholder['sheet'] = "Equipment"
# logger.debug(f"Final output of {equipment['role']} : {placeholder}")
output.append(placeholder)
return output
@@ -241,8 +258,12 @@ class EquipmentWriter(object):
if not isinstance(v, dict):
continue
logger.debug(
f"Writing {equipment['role']} {k} to {equipment['sheet']}, row: {v['row']}, column: {v['column']}")
f"Writing {k}: {v['value']} to {equipment['sheet']}, row: {v['row']}, column: {v['column']}")
if isinstance(v['value'], list):
v['value'] = v['value'][0]
try:
sheet.cell(row=v['row'], column=v['column'], value=v['value'])
except AttributeError as e:
logger.error(f"Couldn't write to {equipment['sheet']}, row: {v['row']}, column: {v['column']}")
logger.error(e)
return self.xl

10
src/submissions/backend/validators/__init__.py
View File

@@ -4,6 +4,8 @@ from openpyxl import load_workbook
from backend.db.models import BasicSubmission, SubmissionType
from tools import jinja_template_loading
from jinja2 import Template
from dateutil.parser import parse
from datetime import datetime
logger = logging.getLogger(f"submissions.{__name__}")
@@ -21,15 +23,15 @@ class RSLNamer(object):
# logger.debug("Creating submission type because none exists")
self.submission_type = self.retrieve_submission_type(filename=filename)
logger.debug(f"got submission type: {self.submission_type}")
if self.submission_type != None:
if self.submission_type is not None:
# logger.debug("Retrieving BasicSubmission subclass")
enforcer = BasicSubmission.find_polymorphic_subclass(polymorphic_identity=self.submission_type)
self.parsed_name = self.retrieve_rsl_number(filename=filename, regex=enforcer.get_regex())
self.sub_object = BasicSubmission.find_polymorphic_subclass(polymorphic_identity=self.submission_type)
self.parsed_name = self.retrieve_rsl_number(filename=filename, regex=self.sub_object.get_regex())
if data is None:
data = dict(submission_type=self.submission_type)
if "submission_type" not in data.keys():
data['submission_type'] = self.submission_type
self.parsed_name = enforcer.enforce_name(instr=self.parsed_name, data=data)
self.parsed_name = self.sub_object.enforce_name(instr=self.parsed_name, data=data)
@classmethod
def retrieve_submission_type(cls, filename: str | Path) -> str:

321
src/submissions/backend/validators/pydant.py
View File

@@ -4,10 +4,9 @@ Contains pydantic models and accompanying validators
from __future__ import annotations
from operator import attrgetter
import uuid, re, logging
from pydantic import BaseModel, field_validator, Field, model_validator
from pydantic import BaseModel, field_validator, Field, model_validator, PrivateAttr
from datetime import date, datetime, timedelta
from dateutil.parser import parse
# from dateutil.parser._parser import ParserError
from dateutil.parser import ParserError
from typing import List, Tuple, Literal
from . import RSLNamer
@@ -22,9 +21,6 @@ from io import BytesIO
logger = logging.getLogger(f"submissions.{__name__}")
# class PydMixin(object):
class PydReagent(BaseModel):
lot: str | None
type: str | None
@@ -125,7 +121,7 @@ class PydReagent(BaseModel):
# output[k] = value
return {k: getattr(self, k) for k in fields}
def toSQL(self, submission: BasicSubmission | str = None) -> Tuple[Reagent, Report]:
def toSQL(self, submission: BasicSubmission | str = None) -> Tuple[Reagent, SubmissionReagentAssociation]:
"""
Converts this instance into a backend.db.models.kit.Reagent instance
@@ -139,7 +135,7 @@ class PydReagent(BaseModel):
logger.debug(f"Reagent SQL constructor is looking up type: {self.type}, lot: {self.lot}")
reagent = Reagent.query(lot_number=self.lot, name=self.name)
logger.debug(f"Result: {reagent}")
if reagent == None:
if reagent is None:
reagent = Reagent()
for key, value in self.__dict__.items():
if isinstance(value, dict):
@@ -164,17 +160,22 @@ class PydReagent(BaseModel):
reagent.__setattr__(key, value)
except AttributeError:
logger.error(f"Couldn't set {key} to {value}")
if submission != None:
if submission is not None and reagent not in submission.reagents:
assoc = SubmissionReagentAssociation(reagent=reagent, submission=submission)
assoc.comments = self.comment
reagent.reagent_submission_associations.append(assoc)
# reagent.reagent_submission_associations.append(assoc)
else:
assoc = None
else:
if submission is not None and reagent not in submission.reagents:
assoc = SubmissionReagentAssociation(reagent=reagent, submission=submission)
assoc.comments = self.comment
# reagent.reagent_submission_associations.append(assoc)
else:
assoc = None
# add end-of-life extension from reagent type to expiry date
# NOTE: this will now be done only in the reporting phase to account for potential changes in end-of-life extensions
return reagent, report
# def improved_dict(self) -> dict:
# fields = list(self.model_fields.keys()) + list(self.model_extra.keys())
# return {k:getattr(self,k) for k in fields}
return reagent, assoc
class PydSample(BaseModel, extra='allow'):
@@ -182,8 +183,8 @@ class PydSample(BaseModel, extra='allow'):
sample_type: str
row: int | List[int] | None
column: int | List[int] | None
assoc_id: int | List[int] | None = Field(default=None)
submission_rank: int | List[int] | None
assoc_id: int | List[int | None] | None = Field(default=None, validate_default=True)
submission_rank: int | List[int] | None = Field(default=0, validate_default=True)
@model_validator(mode='after')
@classmethod
@@ -191,19 +192,22 @@ class PydSample(BaseModel, extra='allow'):
logger.debug(f"Data for pydsample: {data}")
model = BasicSample.find_polymorphic_subclass(polymorphic_identity=data.sample_type)
for k, v in data.model_extra.items():
# print(k, v)
print(k, v)
if k in model.timestamps():
if isinstance(v, str):
v = datetime.strptime(v, "%Y-%m-%d")
data.__setattr__(k, v)
# print(dir(data))
logger.debug(f"Data coming out of validation: {pformat(data)}")
return data
@field_validator("row", "column", "assoc_id", "submission_rank")
@classmethod
def row_int_to_list(cls, value):
if isinstance(value, int):
match value:
case int() | None:
return [value]
case _:
return value
@field_validator("submitter_id", mode="before")
@@ -230,23 +234,23 @@ class PydSample(BaseModel, extra='allow'):
logger.debug(f"Here is the incoming sample dict: \n{self.__dict__}")
instance = BasicSample.query_or_create(sample_type=self.sample_type, submitter_id=self.submitter_id)
for key, value in self.__dict__.items():
# logger.debug(f"Setting sample field {key} to {value}")
match key:
case "row" | "column":
continue
case _:
# instance.set_attribute(name=key, value=value)
# logger.debug(f"Setting sample field {key} to {value}")
instance.__setattr__(key, value)
out_associations = []
if submission != None:
if submission is not None:
assoc_type = self.sample_type.replace("Sample", "").strip()
for row, column, id in zip(self.row, self.column, self.assoc_id):
for row, column, aid, submission_rank in zip(self.row, self.column, self.assoc_id, self.submission_rank):
logger.debug(f"Looking up association with identity: ({submission.submission_type_name} Association)")
logger.debug(f"Looking up association with identity: ({assoc_type} Association)")
association = SubmissionSampleAssociation.query_or_create(association_type=f"{assoc_type} Association",
submission=submission,
sample=instance,
row=row, column=column, id=id)
row=row, column=column, id=aid,
submission_rank=submission_rank)
# logger.debug(f"Using submission_sample_association: {association}")
try:
# instance.sample_submission_associations.append(association)
@@ -336,7 +340,7 @@ class PydSubmission(BaseModel, extra='allow'):
submitter_plate_num: dict | None = Field(default=dict(value=None, missing=True), validate_default=True)
submitted_date: dict | None
rsl_plate_num: dict | None = Field(default=dict(value=None, missing=True), validate_default=True)
submitted_date: dict | None
submitted_date: dict | None = Field(default=dict(value=date.today(), missing=True), validate_default=True)
submitting_lab: dict | None
sample_count: dict | None
extraction_kit: dict | None
@@ -387,14 +391,15 @@ class PydSubmission(BaseModel, extra='allow'):
@field_validator("submitted_date")
@classmethod
def strip_datetime_string(cls, value):
if isinstance(value['value'], datetime):
match value['value']:
case date():
return value
if isinstance(value['value'], date):
return value
if isinstance(value['value'], int):
case datetime():
return value.date()
case int():
return dict(value=datetime.fromordinal(datetime(1900, 1, 1).toordinal() + value['value'] - 2).date(),
missing=True)
case str():
string = re.sub(r"(_|-)\d$", "", value['value'])
try:
output = dict(value=parse(string).date(), missing=True)
@@ -405,6 +410,8 @@ class PydSubmission(BaseModel, extra='allow'):
except Exception as e:
logger.error(f"Problem with parse fallback: {e}")
return output
case _:
raise ValueError(f"Could not get datetime from {value['value']}")
@field_validator("submitting_lab", mode="before")
@classmethod
@@ -417,6 +424,7 @@ class PydSubmission(BaseModel, extra='allow'):
@classmethod
def lookup_submitting_lab(cls, value):
if isinstance(value['value'], str):
logger.debug(f"Looking up organization {value['value']}")
try:
value['value'] = Organization.query(name=value['value']).name
except AttributeError:
@@ -448,6 +456,7 @@ class PydSubmission(BaseModel, extra='allow'):
if check_not_nan(value['value']):
return value
else:
logger.debug("Constructing plate name.")
output = RSLNamer(filename=values.data['filepath'].__str__(), sub_type=sub_type,
data=values.data).parsed_name
return dict(value=output, missing=True)
@@ -549,6 +558,12 @@ class PydSubmission(BaseModel, extra='allow'):
case _:
return value
def __init__(self, **data):
super().__init__(**data)
# this could also be done with default_factory
self.submission_object = BasicSubmission.find_polymorphic_subclass(
polymorphic_identity=self.submission_type['value'])
def set_attribute(self, key, value):
self.__setattr__(name=key, value=value)
@@ -592,7 +607,10 @@ class PydSubmission(BaseModel, extra='allow'):
output = {k: getattr(self, k) for k in fields}
output['reagents'] = [item.improved_dict() for item in self.reagents]
output['samples'] = [item.improved_dict() for item in self.samples]
try:
output['equipment'] = [item.improved_dict() for item in self.equipment]
except TypeError:
pass
else:
# logger.debug("Extracting 'value' from attributes")
output = {k: (getattr(self, k) if not isinstance(getattr(self, k), dict) else getattr(self, k)['value']) for
@@ -612,7 +630,7 @@ class PydSubmission(BaseModel, extra='allow'):
missing_reagents = [reagent for reagent in self.reagents if reagent.missing]
return missing_info, missing_reagents
def toSQL(self) -> Tuple[BasicSubmission, Result]:
def to_sql(self) -> Tuple[BasicSubmission, Result]:
"""
Converts this instance into a backend.db.models.submissions.BasicSubmission instance
@@ -632,11 +650,18 @@ class PydSubmission(BaseModel, extra='allow'):
value = value['value']
logger.debug(f"Setting {key} to {value}")
match key:
case "reagents":
for reagent in self.reagents:
reagent, assoc = reagent.toSQL(submission=instance)
if assoc is not None and assoc not in instance.submission_reagent_associations:
instance.submission_reagent_associations.append(assoc)
# instance.reagents.append(reagent)
case "samples":
for sample in self.samples:
sample, associations, _ = sample.toSQL(submission=instance)
# logger.debug(f"Sample SQL object to be added to submission: {sample.__dict__}")
logger.debug(f"Sample SQL object to be added to submission: {sample.__dict__}")
for assoc in associations:
if assoc is not None and assoc not in instance.submission_sample_associations:
instance.submission_sample_associations.append(assoc)
case "equipment":
logger.debug(f"Equipment: {pformat(self.equipment)}")
@@ -700,7 +725,7 @@ class PydSubmission(BaseModel, extra='allow'):
logger.debug(f"Constructed submissions message: {msg}")
return instance, result
def toForm(self, parent: QWidget):
def to_form(self, parent: QWidget):
"""
Converts this instance into a frontend.widgets.submission_widget.SubmissionFormWidget
@@ -713,211 +738,7 @@ class PydSubmission(BaseModel, extra='allow'):
from frontend.widgets.submission_widget import SubmissionFormWidget
return SubmissionFormWidget(parent=parent, submission=self)
def autofill_excel(self, missing_only: bool = True, backup: bool = False) -> Workbook:
"""
Fills in relevant information/reagent cells in an excel workbook.
Args:
missing_only (bool, optional): Only fill missing items or all. Defaults to True.
backup (bool, optional): Do a full backup of the submission (adds samples). Defaults to False.
Returns:
Workbook: Filled in workbook
"""
# open a new workbook using openpyxl
if self.filepath.stem.startswith("tmp"):
template = SubmissionType.query(name=self.submission_type['value']).template_file
workbook = load_workbook(BytesIO(template))
missing_only = False
else:
try:
workbook = load_workbook(self.filepath)
except Exception as e:
logger.error(f"Couldn't open workbook due to {e}")
template = SubmissionType.query(name=self.submission_type).template_file
workbook = load_workbook(BytesIO(template))
missing_only = False
if missing_only:
info, reagents = self.find_missing()
else:
info = {k: v for k, v in self.improved_dict().items() if isinstance(v, dict)}
reagents = self.reagents
if len(reagents + list(info.keys())) == 0:
# logger.warning("No info to fill in, returning")
return None
logger.debug(f"Info: {pformat(info)}")
logger.debug(f"We have blank info and/or reagents in the excel sheet.\n\tLet's try to fill them in.")
# extraction_kit = lookup_kit_types(ctx=self.ctx, name=self.extraction_kit['value'])
extraction_kit = KitType.query(name=self.extraction_kit['value'])
logger.debug(f"We have the extraction kit: {extraction_kit.name}")
excel_map = extraction_kit.construct_xl_map_for_use(self.submission_type['value'])
# logger.debug(f"Extraction kit map:\n\n{pformat(excel_map)}")
# logger.debug(f"Missing reagents going into autofile: {pformat(reagents)}")
# logger.debug(f"Missing info going into autofile: {pformat(info)}")
new_reagents = []
# logger.debug("Constructing reagent map and values")
for reagent in reagents:
new_reagent = {}
new_reagent['type'] = reagent.type
new_reagent['lot'] = excel_map[new_reagent['type']]['lot']
new_reagent['lot']['value'] = reagent.lot or "NA"
new_reagent['expiry'] = excel_map[new_reagent['type']]['expiry']
new_reagent['expiry']['value'] = reagent.expiry or "NA"
new_reagent['sheet'] = excel_map[new_reagent['type']]['sheet']
# name is only present for Bacterial Culture
try:
new_reagent['name'] = excel_map[new_reagent['type']]['name']
new_reagent['name']['value'] = reagent.name or "Not Applicable"
except Exception as e:
logger.error(f"Couldn't get name due to {e}")
new_reagents.append(new_reagent)
new_info = []
# logger.debug("Constructing info map and values")
for k, v in info.items():
try:
new_item = {}
new_item['type'] = k
new_item['location'] = excel_map['info'][k]
match k:
case "comment":
if v['value'] is not None:
new_item['value'] = "--".join([comment['text'] for comment in v['value']])
else:
new_item['value'] = None
case _:
new_item['value'] = v['value']
new_info.append(new_item)
except KeyError:
logger.error(f"Unable to fill in {k}, not found in relevant info.")
logger.debug(f"New reagents: {new_reagents}")
logger.debug(f"New info: {new_info}")
# get list of sheet names
for sheet in workbook.sheetnames:
# open sheet
worksheet = workbook[sheet]
# Get relevant reagents for that sheet
sheet_reagents = [item for item in new_reagents if sheet in item['sheet']]
for reagent in sheet_reagents:
# logger.debug(f"Attempting to write lot {reagent['lot']['value']} in: row {reagent['lot']['row']}, column {reagent['lot']['column']}")
worksheet.cell(row=reagent['lot']['row'], column=reagent['lot']['column'],
value=reagent['lot']['value'])
# logger.debug(f"Attempting to write expiry {reagent['expiry']['value']} in: row {reagent['expiry']['row']}, column {reagent['expiry']['column']}")
if isinstance(reagent['expiry']['value'], date) and reagent['expiry']['value'].year == 1970:
reagent['expiry']['value'] = "NA"
worksheet.cell(row=reagent['expiry']['row'], column=reagent['expiry']['column'],
value=reagent['expiry']['value'])
try:
# logger.debug(f"Attempting to write name {reagent['name']['value']} in: row {reagent['name']['row']}, column {reagent['name']['column']}")
worksheet.cell(row=reagent['name']['row'], column=reagent['name']['column'],
value=reagent['name']['value'])
except Exception as e:
logger.error(f"Could not write name {reagent['name']['value']} due to {e}")
# Get relevant info for that sheet
new_info = [item for item in new_info if isinstance(item['location'], dict)]
logger.debug(f"New info: {pformat(new_info)}")
sheet_info = [item for item in new_info if item['location']['sheet'] == sheet]
for item in sheet_info:
logger.debug(
f"Attempting: {item['type']} in row {item['location']['row']}, column {item['location']['column']}")
worksheet.cell(row=item['location']['row'], column=item['location']['column'], value=item['value'])
# Hacky way to pop in 'signed by'
custom_parser = BasicSubmission.find_polymorphic_subclass(polymorphic_identity=self.submission_type['value'])
workbook = custom_parser.custom_autofill(workbook, info=self.improved_dict(), backup=backup)
return workbook
def autofill_samples(self, workbook: Workbook) -> Workbook:
"""
Fill in sample rows on the excel sheet
Args:
workbook (Workbook): Input excel workbook
Returns:
Workbook: Updated excel workbook
"""
# sample_info = SubmissionType.query(name=self.submission_type['value']).info_map['samples']
sample_info = SubmissionType.query(name=self.submission_type['value']).construct_sample_map()
logger.debug(f"Sample info: {pformat(sample_info)}")
logger.debug(f"Workbook sheets: {workbook.sheetnames}")
worksheet = workbook[sample_info["lookup_table"]['sheet']]
# logger.debug("Sorting samples by row/column")
samples = sorted(self.samples, key=attrgetter('column', 'row'))
submission_obj = BasicSubmission.find_polymorphic_subclass(polymorphic_identity=self.submission_type)
# custom function to adjust values for writing.
samples = submission_obj.adjust_autofill_samples(samples=samples)
logger.debug(f"Samples: {pformat(samples)}")
# Fail safe against multiple instances of the same sample
for iii, sample in enumerate(samples, start=1):
logger.debug(f"Sample: {sample}")
# custom function to find the row of this sample
row = submission_obj.custom_sample_autofill_row(sample, worksheet=worksheet)
logger.debug(f"Writing to {row}")
if row == None:
row = sample_info['lookup_table']['start_row'] + iii
fields = [field for field in list(sample.model_fields.keys()) +
list(sample.model_extra.keys()) if field in sample_info['lookup_table']['sample_columns'].keys()]
logger.debug(f"Here are the fields we are going to fill:\n\t{fields}")
for field in fields:
column = sample_info['lookup_table']['sample_columns'][field]
value = getattr(sample, field)
match value:
case list():
value = value[0]
case _:
value = value
if field == "row":
value = row_map[value]
worksheet.cell(row=row, column=column, value=value)
return workbook
def autofill_equipment(self, workbook: Workbook) -> Workbook:
"""
Fill in equipment on the excel sheet
Args:
workbook (Workbook): Input excel workbook
Returns:
Workbook: Updated excel workbook
"""
equipment_map = SubmissionType.query(name=self.submission_type['value']).construct_equipment_map()
logger.debug(f"Equipment map: {equipment_map}")
# See if all equipment has a location map
# If not, create a new sheet to store them in.
if not all([len(item.keys()) > 1 for item in equipment_map]):
logger.warning("Creating 'Equipment' sheet to hold unmapped equipment")
workbook.create_sheet("Equipment")
equipment = []
# logger.debug("Contructing equipment info map/values")
for ii, equip in enumerate(self.equipment, start=1):
loc = [item for item in equipment_map if item['role'] == equip.role][0]
try:
loc['name']['value'] = equip.name
loc['process']['value'] = equip.processes[0]
except KeyError:
loc['name'] = dict(row=ii, column=2)
loc['process'] = dict(row=ii, column=3)
loc['name']['value'] = equip.name
loc['process']['value'] = equip.processes[0]
loc['sheet'] = "Equipment"
equipment.append(loc)
logger.debug(f"Using equipment: {equipment}")
for sheet in workbook.sheetnames:
logger.debug(f"Looking at: {sheet}")
worksheet = workbook[sheet]
relevant = [item for item in equipment if item['sheet'] == sheet]
for rel in relevant:
match sheet:
case "Equipment":
worksheet.cell(row=rel['name']['row'], column=1, value=rel['role'])
case _:
pass
worksheet.cell(row=rel['name']['row'], column=rel['name']['column'], value=rel['name']['value'])
worksheet.cell(row=rel['process']['row'], column=rel['process']['column'],
value=rel['process']['value'])
return workbook
def toWriter(self):
def to_writer(self):
from backend.excel.writer import SheetWriter
return SheetWriter(self)
@@ -928,16 +749,14 @@ class PydSubmission(BaseModel, extra='allow'):
Returns:
str: Output filename
"""
template = BasicSubmission.find_polymorphic_subclass(
polymorphic_identity=self.submission_type).filename_template()
template = self.submission_object.filename_template()
# logger.debug(f"Using template string: {template}")
render = RSLNamer.construct_export_name(template=template, **self.improved_dict(dictionaries=False)).replace(
"/", "")
# logger.debug(f"Template rendered as: {render}")
return render
def check_kit_integrity(self, reagenttypes: list = [], extraction_kit: str | dict | None = None) -> Tuple[
List[PydReagent], Report]:
def check_kit_integrity(self, extraction_kit: str | dict | None = None) -> Tuple[List[PydReagent], Report]:
"""
Ensures all reagents expected in kit are listed in Submission
@@ -953,23 +772,12 @@ class PydSubmission(BaseModel, extra='allow'):
extraction_kit = dict(value=extraction_kit)
if extraction_kit is not None and extraction_kit != self.extraction_kit['value']:
self.extraction_kit['value'] = extraction_kit['value']
# reagenttypes = []
# else:
# reagenttypes = [item.type for item in self.reagents]
# else:
# reagenttypes = [item.type for item in self.reagents]
logger.debug(f"Looking up {self.extraction_kit['value']}")
ext_kit = KitType.query(name=self.extraction_kit['value'])
ext_kit_rtypes = [item.to_pydantic() for item in
ext_kit.get_reagents(required=True, submission_type=self.submission_type['value'])]
logger.debug(f"Kit reagents: {ext_kit_rtypes}")
logger.debug(f"Submission reagents: {self.reagents}")
# check if lists are equal
# check = set(ext_kit_rtypes) == set(reagenttypes)
# logger.debug(f"Checking if reagents match kit contents: {check}")
# # what reagent types are in both lists?
# missing = list(set(ext_kit_rtypes).difference(reagenttypes))
# missing = []
# Exclude any reagenttype found in this pyd not expected in kit.
expected_check = [item.type for item in ext_kit_rtypes]
output_reagents = [rt for rt in self.reagents if rt.type in expected_check]
@@ -977,11 +785,6 @@ class PydSubmission(BaseModel, extra='allow'):
missing_check = [item.type for item in output_reagents]
missing_reagents = [rt for rt in ext_kit_rtypes if rt.type not in missing_check]
missing_reagents += [rt for rt in output_reagents if rt.missing]
# for rt in ext_kit_rtypes:
# if rt.type not in [item.type for item in output_reagents]:
# missing.append(rt)
# if rt.type not in [item.type for item in output_reagents]:
# output_reagents.append(rt)
output_reagents += [rt for rt in missing_reagents if rt not in output_reagents]
logger.debug(f"Missing reagents types: {missing_reagents}")
# if lists are equal return no problem
@@ -1026,7 +829,7 @@ class PydOrganization(BaseModel):
for field in self.model_fields:
match field:
case "contacts":
value = [item.toSQL() for item in getattr(self, field)]
value = [item.to_sql() for item in getattr(self, field)]
case _:
value = getattr(self, field)
# instance.set_attribute(name=field, value=value)
@@ -1095,7 +898,7 @@ class PydEquipmentRole(BaseModel):
equipment: List[PydEquipment]
processes: List[str] | None
def toForm(self, parent, used: list) -> "RoleComboBox":
def to_form(self, parent, used: list) -> "RoleComboBox":
"""
Creates a widget for user input into this class.

2
src/submissions/frontend/widgets/equipment_usage.py
View File

@@ -36,7 +36,7 @@ class EquipmentUsage(QDialog):
self.layout.addWidget(label)
# logger.debug("Creating widgets for equipment")
for eq in self.opt_equipment:
widg = eq.toForm(parent=self, used=self.used_equipment)
widg = eq.to_form(parent=self, used=self.used_equipment)
self.layout.addWidget(widg)
widg.update_processes()
self.layout.addWidget(self.buttonBox)

2
src/submissions/frontend/widgets/submission_details.py
View File

@@ -105,7 +105,7 @@ class SubmissionDetails(QDialog):
logger.debug(f"Signing off on {submission} - ({getuser()})")
if isinstance(submission, str):
# submission = BasicSubmission.query(rsl_number=submission)
submission = BasicSubmission.query(rsl_plate_number=submission)
submission = BasicSubmission.query(rsl_plate_num=submission)
submission.signed_by = getuser()
submission.save()
self.submission_details(submission=self.rsl_plate_num)

19
src/submissions/frontend/widgets/submission_widget.py
View File

@@ -1,3 +1,5 @@
import sys
from PyQt6.QtWidgets import (
QWidget, QPushButton, QVBoxLayout,
QComboBox, QDateEdit, QLineEdit, QLabel
@@ -7,7 +9,7 @@ from pathlib import Path
from . import select_open_file, select_save_file
import logging, difflib, inspect
from pathlib import Path
from tools import Report, Result, check_not_nan
from tools import Report, Result, check_not_nan, workbook_2_csv
from backend.excel.parser import SheetParser
from backend.validators import PydSubmission, PydReagent
from backend.db import (
@@ -104,7 +106,7 @@ class SubmissionFormContainer(QWidget):
logger.debug(f"Submission dictionary:\n{pformat(self.prsr.sub)}")
self.pyd = self.prsr.to_pydantic()
logger.debug(f"Pydantic result: \n\n{pformat(self.pyd)}\n\n")
self.form = self.pyd.toForm(parent=self)
self.form = self.pyd.to_form(parent=self)
self.layout().addWidget(self.form)
# if self.prsr.sample_result != None:
# report.add_result(msg=self.prsr.sample_result, status="Warning")
@@ -347,7 +349,8 @@ class SubmissionFormWidget(QWidget):
self.app.result_reporter()
return
logger.debug(f"PYD before transformation into SQL:\n\n{self.pyd}\n\n")
base_submission, result = self.pyd.toSQL()
base_submission, result = self.pyd.to_sql()
logger.debug(f"SQL object: {pformat(base_submission.__dict__)}")
# logger.debug(f"Base submission: {base_submission.to_dict()}")
# check output message for issues
match result.code:
@@ -371,6 +374,7 @@ class SubmissionFormWidget(QWidget):
return
case _:
pass
# assert base_submission.reagents != []
# add reagents to submission object
for reagent in base_submission.reagents:
# logger.debug(f"Updating: {reagent} with {reagent.lot}")
@@ -397,13 +401,18 @@ class SubmissionFormWidget(QWidget):
Args:
fname (Path | None, optional): Input filename. Defaults to None.
"""
self.parse_form()
# self.parse_form()
if isinstance(fname, bool) or fname == None:
fname = select_save_file(obj=self, default_name=self.pyd.construct_filename(), extension="csv")
try:
self.pyd.csv.to_csv(fname.__str__(), index=False)
# logger.debug(f'')
# self.pyd.csv.to_csv(fname.__str__(), index=False)
workbook_2_csv(worksheet=self.pyd.csv, filename=fname)
except PermissionError:
logger.debug(f"Could not get permissions to {fname}. Possibly the request was cancelled.")
except AttributeError:
logger.error(f"No csv file found in the submission at this point.")
def parse_form(self) -> PydSubmission:
"""

9
src/submissions/tools.py
View File

@@ -4,7 +4,7 @@ Contains miscellaenous functions used by both frontend and backend.
from __future__ import annotations
from pathlib import Path
import numpy as np
import logging, re, yaml, sys, os, stat, platform, getpass, inspect
import logging, re, yaml, sys, os, stat, platform, getpass, inspect, csv
import pandas as pd
from jinja2 import Environment, FileSystemLoader
from logging import handlers
@@ -16,6 +16,7 @@ from pydantic_settings import BaseSettings, SettingsConfigDict
from typing import Any, Tuple, Literal, List
from PyQt6.QtGui import QTextDocument, QPageSize
from PyQt6.QtWebEngineWidgets import QWebEngineView
from openpyxl.worksheet.worksheet import Worksheet
# from PyQt6 import QtPrintSupport, QtCore, QtWebEngineWidgets
from PyQt6.QtPrintSupport import QPrinter
@@ -562,6 +563,12 @@ def remove_key_from_list_of_dicts(input:list, key:str):
del item[key]
return input
def workbook_2_csv(worksheet: Worksheet, filename:Path):
with open(filename, 'w', newline="") as f:
c = csv.writer(f)
for r in worksheet.rows:
c.writerow([cell.value for cell in r])
ctx = get_config(None)
def is_power_user() -> bool: