Code cleanup for managers complete.

This commit is contained in:
lwark
2025-09-04 15:01:02 -05:00
parent 1a90543639
commit c8b4762747
6 changed files with 24 additions and 63 deletions

View File

@@ -1,8 +1,8 @@
"""
Module for manager defaults.
"""
import logging import logging
from pathlib import Path from pathlib import Path
from typing import Literal
from backend.db.models import ProcedureType
from frontend.widgets.functions import select_open_file from frontend.widgets.functions import select_open_file
from tools import get_application_from_parent from tools import get_application_from_parent
from backend.validators.pydant import PydBaseClass from backend.validators.pydant import PydBaseClass
@@ -13,7 +13,6 @@ logger = logging.getLogger(f"submissions.{__name__}")
class DefaultManager(object): class DefaultManager(object):
def __init__(self, parent, input_object: Path | str | None = None): def __init__(self, parent, input_object: Path | str | None = None):
logger.debug(f"FName before correction: {type(input_object)}")
self.parent = parent self.parent = parent
match input_object: match input_object:
case str(): case str():
@@ -23,16 +22,12 @@ class DefaultManager(object):
self.input_object = input_object self.input_object = input_object
self.pyd = self.to_pydantic() self.pyd = self.to_pydantic()
case x if issubclass(input_object.__class__, PydBaseClass): case x if issubclass(input_object.__class__, PydBaseClass):
# logger.debug("Subclass of PydBaseClass")
self.pyd = input_object self.pyd = input_object
case x if issubclass(input_object.__class__, BaseClass): case x if issubclass(input_object.__class__, BaseClass):
# logger.debug("Subclass of BaseClass")
self.pyd = input_object.to_pydantic() self.pyd = input_object.to_pydantic()
case _: case _:
self.input_object = select_open_file(file_extension="xlsx", obj=get_application_from_parent(parent)) self.input_object = select_open_file(file_extension="xlsx", obj=get_application_from_parent(parent))
self.pyd = self.to_pydantic() self.pyd = self.to_pydantic()
# logger.debug(f"FName after correction: {input_object}")
from .clientsubmissions import DefaultClientSubmissionManager from .clientsubmissions import DefaultClientSubmissionManager

View File

@@ -1,9 +1,10 @@
"""
Module for manager of ClientSubmission object
"""
from __future__ import annotations from __future__ import annotations
import logging import logging, sys
import sys
from typing import TYPE_CHECKING from typing import TYPE_CHECKING
from pathlib import Path from pathlib import Path
from openpyxl.reader.excel import load_workbook
from openpyxl.workbook import Workbook from openpyxl.workbook import Workbook
from backend.validators import RSLNamer from backend.validators import RSLNamer
from backend.managers import DefaultManager from backend.managers import DefaultManager
@@ -42,23 +43,15 @@ class DefaultClientSubmissionManager(DefaultManager):
self.sample_parser = ClientSubmissionSampleParser(filepath=self.input_object, self.sample_parser = ClientSubmissionSampleParser(filepath=self.input_object,
submissiontype=self.submissiontype, submissiontype=self.submissiontype,
start_row=self.info_parser.end_row) start_row=self.info_parser.end_row)
logger.debug(self.sample_parser.__dict__)
self.clientsubmission = self.info_parser.to_pydantic() self.clientsubmission = self.info_parser.to_pydantic()
self.clientsubmission.full_batch_size = self.sample_parser.end_row - self.sample_parser.start_row self.clientsubmission.full_batch_size = self.sample_parser.end_row - self.sample_parser.start_row
self.clientsubmission.sample = self.sample_parser.to_pydantic() self.clientsubmission.sample = self.sample_parser.to_pydantic()
return self.clientsubmission return self.clientsubmission
# def to_pydantic(self):
# self.clientsubmission = self.info_parser.to_pydantic()
# self.clientsubmission.sample = self.sample_parser.to_pydantic()
def write(self, workbook: Workbook) -> Workbook: def write(self, workbook: Workbook) -> Workbook:
# workbook: Workbook = load_workbook(BytesIO(self.submissiontype.template_file))
self.info_writer = ClientSubmissionInfoWriter(pydant_obj=self.pyd) self.info_writer = ClientSubmissionInfoWriter(pydant_obj=self.pyd)
assert isinstance(self.info_writer, ClientSubmissionInfoWriter) assert isinstance(self.info_writer, ClientSubmissionInfoWriter)
logger.debug("Attempting write.")
workbook = self.info_writer.write_to_workbook(workbook) workbook = self.info_writer.write_to_workbook(workbook)
self.sample_writer = ClientSubmissionSampleWriter(pydant_obj=self.pyd) self.sample_writer = ClientSubmissionSampleWriter(pydant_obj=self.pyd)
workbook = self.sample_writer.write_to_workbook(workbook, start_row=self.info_writer.worksheet.max_row + 1) workbook = self.sample_writer.write_to_workbook(workbook, start_row=self.info_writer.worksheet.max_row + 1)

View File

@@ -1,12 +1,10 @@
"""
Module for manager of Procedure object.
"""
from __future__ import annotations from __future__ import annotations
import logging import logging, sys
from io import BytesIO
from pprint import pformat
from openpyxl.reader.excel import load_workbook
from openpyxl.workbook import Workbook from openpyxl.workbook import Workbook
from backend.managers import DefaultManager
from backend.managers import DefaultManager, results
from typing import TYPE_CHECKING from typing import TYPE_CHECKING
from pathlib import Path from pathlib import Path
from backend.excel.parsers import procedure_parsers from backend.excel.parsers import procedure_parsers
@@ -57,7 +55,6 @@ class DefaultProcedureManager(DefaultManager):
self.equipment = self.equipment_parser.to_pydantic() self.equipment = self.equipment_parser.to_pydantic()
def write(self, workbook: Workbook) -> Workbook: def write(self, workbook: Workbook) -> Workbook:
# workbook = load_workbook(BytesIO(self.proceduretype.template_file))
try: try:
info_writer = getattr(procedure_writers, f"{self.proceduretype.name.replace(' ', '')}InfoWriter") info_writer = getattr(procedure_writers, f"{self.proceduretype.name.replace(' ', '')}InfoWriter")
except AttributeError: except AttributeError:
@@ -82,16 +79,9 @@ class DefaultProcedureManager(DefaultManager):
sample_writer = procedure_writers.ProcedureSampleWriter sample_writer = procedure_writers.ProcedureSampleWriter
self.sample_writer = sample_writer(pydant_obj=self.pyd) self.sample_writer = sample_writer(pydant_obj=self.pyd)
workbook = self.sample_writer.write_to_workbook(workbook) workbook = self.sample_writer.write_to_workbook(workbook)
# # logger.debug(self.pyd.result)
# # TODO: Find way to group results by result_type. # # TODO: Find way to group results by result_type.
for result in self.pyd.result: for result in self.pyd.result:
logger.debug(f"Writing {result.result_type}")
Writer = getattr(results_writers, f"{result.result_type}InfoWriter") Writer = getattr(results_writers, f"{result.result_type}InfoWriter")
res_info_writer = Writer(pydant_obj=result, proceduretype=self.proceduretype) res_info_writer = Writer(pydant_obj=result, proceduretype=self.proceduretype)
workbook = res_info_writer.write_to_workbook(workbook=workbook) workbook = res_info_writer.write_to_workbook(workbook=workbook)
# # sample_results = [sample.result for sample in self.pyd.sample]
# # logger.debug(pformat(self.pyd.sample_results))
# Writer = getattr(results_writers, "PCRSampleWriter")
# res_sample_writer = Writer(pydant_obj=self.pyd.sample_results, proceduretype=self.proceduretype)
# workbook = res_sample_writer.write_to_workbook(workbook=workbook)
return workbook return workbook

View File

@@ -1,4 +1,6 @@
"""
Module for default results manager
"""
from __future__ import annotations from __future__ import annotations
import logging import logging
from .. import DefaultManager from .. import DefaultManager
@@ -7,7 +9,6 @@ from pathlib import Path
from frontend.widgets.functions import select_open_file from frontend.widgets.functions import select_open_file
from tools import get_application_from_parent from tools import get_application_from_parent
from typing import TYPE_CHECKING, List from typing import TYPE_CHECKING, List
if TYPE_CHECKING: if TYPE_CHECKING:
from backend.validators.pydant import PydResults from backend.validators.pydant import PydResults
@@ -17,13 +18,11 @@ logger = logging.getLogger(f"submission.{__name__}")
class DefaultResultsManager(DefaultManager): class DefaultResultsManager(DefaultManager):
def __init__(self, procedure: Procedure, parent, fname: Path | str | None = None): def __init__(self, procedure: Procedure, parent, fname: Path | str | None = None):
logger.debug(f"FName before correction: {fname}")
self.procedure = procedure self.procedure = procedure
if not fname: if not fname:
self.fname = select_open_file(file_extension="xlsx", obj=get_application_from_parent(parent)) self.fname = select_open_file(file_extension="xlsx", obj=get_application_from_parent(parent))
elif isinstance(fname, str): elif isinstance(fname, str):
self.fname = Path(fname) self.fname = Path(fname)
logger.debug(f"FName after correction: {fname}")
def procedure_to_pydantic(self) -> PydResults: def procedure_to_pydantic(self) -> PydResults:
info = self.info_parser.to_pydantic() info = self.info_parser.to_pydantic()

View File

@@ -1,14 +1,12 @@
""" """
Module for pcr results from Design and Analysis Studio
""" """
from __future__ import annotations from __future__ import annotations
import logging import logging
from io import BytesIO from io import BytesIO
from pathlib import Path from pathlib import Path
from typing import Tuple, List, TYPE_CHECKING from typing import TYPE_CHECKING
from openpyxl.reader.excel import load_workbook from openpyxl.reader.excel import load_workbook
from backend.db.models import Procedure from backend.db.models import Procedure
from backend.excel.parsers.results_parsers.pcr_results_parser import PCRSampleParser, PCRInfoParser from backend.excel.parsers.results_parsers.pcr_results_parser import PCRSampleParser, PCRInfoParser
from backend.excel.writers.results_writers.pcr_results_writer import PCRInfoWriter, PCRSampleWriter from backend.excel.writers.results_writers.pcr_results_writer import PCRInfoWriter, PCRSampleWriter
@@ -33,9 +31,3 @@ class PCRManager(DefaultResultsManager):
self.info_writer = PCRInfoWriter(pydant_obj=self.procedure.to_pydantic(), proceduretype=self.procedure.proceduretype) self.info_writer = PCRInfoWriter(pydant_obj=self.procedure.to_pydantic(), proceduretype=self.procedure.proceduretype)
workbook = self.info_writer.write_to_workbook(workbook) workbook = self.info_writer.write_to_workbook(workbook)
self.sample_writer = PCRSampleWriter(pydant_obj=self.procedure.to_pydantic(), proceduretype=self.procedure.proceduretype) self.sample_writer = PCRSampleWriter(pydant_obj=self.procedure.to_pydantic(), proceduretype=self.procedure.proceduretype)

View File

@@ -1,13 +1,10 @@
"""
Module for managing Runs object
"""
from __future__ import annotations from __future__ import annotations
import logging import logging, sys
from pathlib import Path
from pprint import pformat from pprint import pformat
from openpyxl import load_workbook
from openpyxl.workbook.workbook import Workbook from openpyxl.workbook.workbook import Workbook
from tools import copy_xl_sheet
from backend.managers import DefaultManager from backend.managers import DefaultManager
logger = logging.getLogger(f"submissions.{__name__}") logger = logging.getLogger(f"submissions.{__name__}")
@@ -16,16 +13,11 @@ class DefaultRunManager(DefaultManager):
def write(self) -> Workbook: def write(self) -> Workbook:
from backend.managers import DefaultClientSubmissionManager, DefaultProcedureManager from backend.managers import DefaultClientSubmissionManager, DefaultProcedureManager
logger.debug(f"Initializing write") logger.info(f"Initializing write")
clientsubmission = DefaultClientSubmissionManager(parent=self.parent, input_object=self.pyd.clientsubmission, submissiontype=self.pyd.clientsubmission.submissiontype) clientsubmission = DefaultClientSubmissionManager(parent=self.parent, input_object=self.pyd.clientsubmission, submissiontype=self.pyd.clientsubmission.submissiontype)
workbook = Workbook() workbook = Workbook()
workbook = clientsubmission.write(workbook=workbook) workbook = clientsubmission.write(workbook=workbook)
for procedure in self.pyd.procedure: for procedure in self.pyd.procedure:
# # logger.debug(f"Running procedure: {pformat(procedure.__dict__)}")
procedure = DefaultProcedureManager(proceduretype=procedure.proceduretype, parent=self.parent, input_object=procedure) procedure = DefaultProcedureManager(proceduretype=procedure.proceduretype, parent=self.parent, input_object=procedure)
workbook: Workbook = procedure.write(workbook=workbook) workbook: Workbook = procedure.write(workbook=workbook)
# for sheetname in wb.sheetnames:
# source_sheet = wb[sheetname]
# ws = workbook.create_sheet(sheetname)
# copy_xl_sheet(source_sheet, ws)
return workbook return workbook