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

Addition of autofilling excel forms. Improved pydantic validation.

Browse Source
This commit is contained in:
Landon Wark
2023-07-19 14:33:15 -05:00
parent 1c804bfc6a
commit ba35696055
21 changed files with 774 additions and 961 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/submissions/__init__.py
View File

@@ -4,7 +4,7 @@ from pathlib import Path
# Version of the realpython-reader package
__project__ = "submissions"
__version__ = "202307.1b"
__version__ = "202307.2b"
__author__ = {"name":"Landon Wark", "email":"Landon.Wark@phac-aspc.gc.ca"}
__copyright__ = "2022-2023, Government of Canada"

58
src/submissions/backend/db/functions.py
View File

@@ -5,14 +5,12 @@ Convenience functions for interacting with the database.
from . import models
from .models.kits import reagenttypes_kittypes
from .models.submissions import reagents_submissions
# from .models.samples import WWSample
import pandas as pd
import sqlalchemy.exc
import sqlite3
import logging
from datetime import date, datetime, timedelta
from sqlalchemy import and_
import uuid
from sqlalchemy import JSON, event
from sqlalchemy.engine import Engine
import json
@@ -22,6 +20,7 @@ import yaml
from pathlib import Path
logger = logging.getLogger(f"submissions.{__name__}")
# The below _should_ allow automatic creation of foreign keys in the database
@@ -111,12 +110,12 @@ def construct_submission_info(ctx:dict, info_dict:dict) -> models.BasicSubmissio
# convert submission type into model name
query = info_dict['submission_type'].replace(" ", "")
# Ensure an rsl plate number exists for the plate
# if info_dict["rsl_plate_num"] == 'nan' or info_dict["rsl_plate_num"] == None or not check_not_nan(info_dict["rsl_plate_num"]):
if not check_regex_match("^RSL", info_dict["rsl_plate_num"]):
instance = None
msg = "A proper RSL plate number is required."
return instance, {'code': 2, 'message': "A proper RSL plate number is required."}
else:
# enforce conventions on the rsl plate number from the form
info_dict['rsl_plate_num'] = RSLNamer(info_dict["rsl_plate_num"]).parsed_name
# check database for existing object
instance = ctx['database_session'].query(models.BasicSubmission).filter(models.BasicSubmission.rsl_plate_num==info_dict['rsl_plate_num']).first()
@@ -160,10 +159,11 @@ def construct_submission_info(ctx:dict, info_dict:dict) -> models.BasicSubmissio
case "submitter_plate_num":
# Because of unique constraint, there will be problems with
# multiple submissions named 'None', so...
# Should be depreciated with use of pydantic validator
logger.debug(f"Submitter plate id: {info_dict[item]}")
if info_dict[item] == None or info_dict[item] == "None" or info_dict[item] == "":
logger.debug(f"Got None as a submitter plate number, inserting random string to preserve database unique constraint.")
info_dict[item] = uuid.uuid4().hex.upper()
# if info_dict[item] == None or info_dict[item] == "None" or info_dict[item] == "":
# logger.debug(f"Got None as a submitter plate number, inserting random string to preserve database unique constraint.")
# info_dict[item] = uuid.uuid4().hex.upper()
field_value = info_dict[item]
case _:
field_value = info_dict[item]
@@ -233,20 +233,6 @@ def construct_reagent(ctx:dict, info_dict:dict) -> models.Reagent:
# pass
return reagent
# def lookup_reagent(ctx:dict, reagent_lot:str) -> models.Reagent:
# """
# Query db for reagent based on lot number
# Args:
# ctx (dict): settings passed down from gui
# reagent_lot (str): lot number to query
# Returns:
# models.Reagent: looked up reagent
# """
# lookedup = ctx['database_session'].query(models.Reagent).filter(models.Reagent.lot==reagent_lot).first()
# return lookedup
def get_all_reagenttype_names(ctx:dict) -> list[str]:
"""
Lookup all reagent types and get names
@@ -276,7 +262,7 @@ def lookup_reagenttype_by_name(ctx:dict, rt_name:str) -> models.ReagentType:
logger.debug(f"Found ReagentType: {lookedup}")
return lookedup
def lookup_kittype_by_use(ctx:dict, used_by:str) -> list[models.KitType]:
def lookup_kittype_by_use(ctx:dict, used_by:str|None=None) -> list[models.KitType]:
"""
Lookup kits by a sample type its used for
@@ -287,7 +273,10 @@ def lookup_kittype_by_use(ctx:dict, used_by:str) -> list[models.KitType]:
Returns:
list[models.KitType]: list of kittypes that have that sample type in their uses
"""
return ctx['database_session'].query(models.KitType).filter(models.KitType.used_for.contains(used_by)).all()
if used_by != None:
return ctx['database_session'].query(models.KitType).filter(models.KitType.used_for.contains(used_by)).all()
else:
return ctx['database_session'].query(models.KitType).all()
def lookup_kittype_by_name(ctx:dict, name:str) -> models.KitType:
"""
@@ -872,19 +861,34 @@ def platemap_plate(submission:models.BasicSubmission) -> list:
# image = make_plate_map(plate_dicto)
return plate_dicto
def lookup_reagent(ctx:dict, reagent_lot:str|None=None, type_name:str|None=None) -> models.Reagent:
def lookup_reagent(ctx:dict, reagent_lot:str, type_name:str|None=None) -> models.Reagent:
"""
Query db for reagent based on lot number
Query db for reagent based on lot number, with optional reagent type to enforce
Args:
ctx (dict): settings passed down from gui
reagent_lot (str): lot number to query
type_name (str | None, optional): name of reagent type. Defaults to None.
Returns:
models.Reagent: looked up reagent
"""
if reagent_lot != None and type_name != None:
return ctx['database_session'].query(models.Reagent).join(models.Reagent.type, aliased=True).filter(models.ReagentType.name==type_name).filter(models.Reagent.lot==reagent_lot).all()
return ctx['database_session'].query(models.Reagent).join(models.Reagent.type, aliased=True).filter(models.ReagentType.name==type_name).filter(models.Reagent.lot==reagent_lot).first()
elif type_name == None:
return ctx['database_session'].query(models.Reagent).filter(models.Reagent.lot==reagent_lot).first()
return ctx['database_session'].query(models.Reagent).filter(models.Reagent.lot==reagent_lot).first()
def lookup_last_used_reagenttype_lot(ctx:dict, type_name:str) -> models.Reagent:
"""
Look up the last used reagent of the reagent type
Args:
ctx (dict): Settings passed down from gui
type_name (str): Name of reagent type
Returns:
models.Reagent: Reagent object with last used lot.
"""
rt = ctx['database_session'].query(models.ReagentType).filter(models.ReagentType.name==type_name).first()
logger.debug(f"Reagent type looked up for {type_name}: {rt.__str__()}")
return lookup_reagent(ctx=ctx, reagent_lot=rt.last_used, type_name=type_name)

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

@@ -55,11 +55,8 @@ class ReagentType(Base):
instances = relationship("Reagent", back_populates="type") #: concrete instances of this reagent type
eol_ext = Column(Interval()) #: extension of life interval
required = Column(INTEGER, server_default="1") #: sqlite boolean to determine if reagent type is essential for the kit
# __table_args__ = (
# CheckConstraint(required >= 0, name='check_required_positive'),
# CheckConstraint(required < 2, name='check_required_less_2'),
# {})
last_used = Column(String(32)) #: last used lot number of this type of reagent
@validates('required')
def validate_age(self, key, value):
if not 0 <= value < 2:
@@ -125,6 +122,13 @@ class Reagent(Base):
"expiry": place_holder.strftime("%Y-%m-%d")
}
def to_reagent_dict(self) -> dict:
return {
"type": self.type.name,
"lot": self.lot,
"expiry": self.expiry.strftime("%Y-%m-%d")
}
class Discount(Base):
"""

13
src/submissions/backend/db/models/samples.py
View File

@@ -6,6 +6,7 @@ from sqlalchemy import Column, String, TIMESTAMP, INTEGER, ForeignKey, FLOAT, BO
from sqlalchemy.orm import relationship
import logging
logger = logging.getLogger(f"submissions.{__name__}")
@@ -22,7 +23,7 @@ class WWSample(Base):
rsl_plate = relationship("Wastewater", back_populates="samples") #: relationship to parent plate
rsl_plate_id = Column(INTEGER, ForeignKey("_submissions.id", ondelete="SET NULL", name="fk_WWS_submission_id"))
collection_date = Column(TIMESTAMP) #: Date submission received
well_number = Column(String(8)) #: location on 24 well plate
well_number = Column(String(8)) #: location on 96 well plate
# The following are fields from the sample tracking excel sheet Ruth put together.
# I have no idea when they will be implemented or how.
testing_type = Column(String(64))
@@ -36,7 +37,7 @@ class WWSample(Base):
ww_seq_run_id = Column(String(64))
sample_type = Column(String(8))
pcr_results = Column(JSON)
elution_well = Column(String(8)) #: location on 96 well plate
well_24 = Column(String(8)) #: location on 24 well plate
artic_rsl_plate = relationship("WastewaterArtic", back_populates="samples")
artic_well_number = Column(String(8))
@@ -57,10 +58,6 @@ class WWSample(Base):
Returns:
dict: well location and id NOTE: keys must sync with BCSample to_sub_dict below
"""
# well_col = self.well_number[1:]
# well_row = self.well_number[0]
# if well_col > 4:
# well
if self.ct_n1 != None and self.ct_n2 != None:
# logger.debug(f"Using well info in name.")
name = f"{self.ww_sample_full_id}\n\t- ct N1: {'{:.2f}'.format(self.ct_n1)} ({self.n1_status})\n\t- ct N2: {'{:.2f}'.format(self.ct_n2)} ({self.n2_status})"
@@ -87,8 +84,8 @@ class WWSample(Base):
except TypeError as e:
logger.error(f"Couldn't check positives for {self.rsl_number}. Looks like there isn't PCR data.")
return None
well_row = row_dict[self.elution_well[0]]
well_col = self.elution_well[1:]
well_row = row_dict[self.well_number[0]]
well_col = self.well_number[1:]
# if positive:
# try:
# # The first character of the elution well is the row

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

@@ -5,7 +5,6 @@ import math
from . import Base
from sqlalchemy import Column, String, TIMESTAMP, INTEGER, ForeignKey, Table, JSON, FLOAT
from sqlalchemy.orm import relationship
from datetime import datetime as dt
import logging
import json
from json.decoder import JSONDecodeError
@@ -164,7 +163,8 @@ class BasicSubmission(Base):
def calculate_base_cost(self):
try:
cols_count_96 = ceil(int(self.sample_count) / 8)
# cols_count_96 = ceil(int(self.sample_count) / 8)
cols_count_96 = self.calculate_column_count()
except Exception as e:
logger.error(f"Column count error: {e}")
# cols_count_24 = ceil(int(self.sample_count) / 3)
@@ -173,6 +173,11 @@ class BasicSubmission(Base):
except Exception as e:
logger.error(f"Calculation error: {e}")
def calculate_column_count(self):
columns = [int(sample.well_number[-2:]) for sample in self.samples]
logger.debug(f"Here are the columns for {self.rsl_plate_num}: {columns}")
return max(columns)
# Below are the custom submission types
class BacterialCulture(BasicSubmission):

44
src/submissions/backend/excel/__init__.py
View File

@@ -4,47 +4,3 @@ Contains pandas convenience functions for interacting with excel workbooks
from .reports import *
from .parser import *
# from pandas import DataFrame
# import re
# def get_unique_values_in_df_column(df: DataFrame, column_name: str) -> list:
# """
# get all unique values in a dataframe column by name
# Args:
# df (DataFrame): input dataframe
# column_name (str): name of column of interest
# Returns:
# list: sorted list of unique values
# """
# return sorted(df[column_name].unique())
# def drop_reruns_from_df(ctx:dict, df: DataFrame) -> DataFrame:
# """
# Removes semi-duplicates from dataframe after finding sequencing repeats.
# Args:
# settings (dict): settings passed from gui
# df (DataFrame): initial dataframe
# Returns:
# DataFrame: dataframe with originals removed in favour of repeats.
# """
# sample_names = get_unique_values_in_df_column(df, column_name="name")
# if 'rerun_regex' in ctx:
# # logger.debug(f"Compiling regex from: {settings['rerun_regex']}")
# rerun_regex = re.compile(fr"{ctx['rerun_regex']}")
# for sample in sample_names:
# # logger.debug(f'Running search on {sample}')
# if rerun_regex.search(sample):
# # logger.debug(f'Match on {sample}')
# first_run = re.sub(rerun_regex, "", sample)
# # logger.debug(f"First run: {first_run}")
# df = df.drop(df[df.name == first_run].index)
# return df
# else:
# return None

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

@@ -8,14 +8,14 @@ import pandas as pd
from pathlib import Path
from backend.db.models import WWSample, BCSample
from backend.db import lookup_ww_sample_by_ww_sample_num
from backend.pydant import PydSubmission
from backend.pydant import PydSubmission, PydReagent
import logging
from collections import OrderedDict
import re
import numpy as np
from datetime import date, datetime
import uuid
from tools import check_not_nan, RSLNamer, massage_common_reagents
from tools import check_not_nan, RSLNamer, massage_common_reagents, convert_nans_to_nones
logger = logging.getLogger(f"submissions.{__name__}")
@@ -26,31 +26,29 @@ class SheetParser(object):
def __init__(self, ctx:dict, filepath:Path|None = None):
"""
Args:
ctx (dict): Settings passed down from gui
filepath (Path | None, optional): file path to excel sheet. Defaults to None.
"""
"""
self.ctx = ctx
logger.debug(f"Parsing {filepath.__str__()}")
# set attributes based on kwargs from gui ctx
# for kwarg in kwargs:
# setattr(self, f"_{kwarg}", kwargs[kwarg])
# self.__dict__.update(kwargs)
if filepath == None:
logger.error(f"No filepath given.")
self.xl = None
else:
self.filepath = filepath
# Open excel file
try:
self.xl = pd.ExcelFile(filepath.__str__())
except ValueError as e:
logger.error(f"Incorrect value: {e}")
self.xl = None
# TODO: replace OrderedDict with pydantic BaseModel
self.sub = OrderedDict()
# make decision about type of sample we have
self.sub['submission_type'] = self.type_decider()
# select proper parser based on sample type
parse_sub = getattr(self, f"parse_{self.sub['submission_type'].lower()}")
parse_sub()
# self.calculate_column_count()
def type_decider(self) -> str:
"""
@@ -65,7 +63,7 @@ class SheetParser(object):
return categories[0].replace(" ", "_")
else:
# This code is going to be depreciated once there is full adoption of the client sheets
# with updated metadata
# with updated metadata... but how will it work for Artic?
try:
for type in self.ctx['submission_types']:
# This gets the *first* submission type that matches the sheet names in the workbook
@@ -76,7 +74,6 @@ class SheetParser(object):
logger.warning(f"We were unable to parse the submission type due to: {e}")
return "Unknown"
def parse_unknown(self) -> None:
"""
Dummy function to handle unknown excel structures
@@ -84,7 +81,6 @@ class SheetParser(object):
logger.error(f"Unknown excel workbook structure. Cannot parse.")
self.sub = None
def parse_generic(self, sheet_name:str) -> pd.DataFrame:
"""
Pulls information common to all wasterwater/bacterial culture types and passes on dataframe
@@ -98,14 +94,17 @@ class SheetParser(object):
# self.xl is a pd.ExcelFile so we need to parse it into a df
submission_info = self.xl.parse(sheet_name=sheet_name, dtype=object)
self.sub['submitter_plate_num'] = submission_info.iloc[0][1]
self.sub['rsl_plate_num'] = RSLNamer(submission_info.iloc[10][1]).parsed_name
if check_not_nan(submission_info.iloc[10][1]):
self.sub['rsl_plate_num'] = RSLNamer(submission_info.iloc[10][1]).parsed_name
else:
# self.sub['rsl_plate_num'] = RSLNamer(self.filepath).parsed_name
self.sub['rsl_plate_num'] = None
self.sub['submitted_date'] = submission_info.iloc[1][1]
self.sub['submitting_lab'] = submission_info.iloc[0][3]
self.sub['sample_count'] = submission_info.iloc[2][3]
self.sub['extraction_kit'] = submission_info.iloc[3][3]
return submission_info
def parse_bacterial_culture(self) -> None:
"""
pulls info specific to bacterial culture sample type
@@ -121,22 +120,27 @@ class SheetParser(object):
for ii, row in df.iterrows():
# skip positive control
logger.debug(f"Running reagent parse for {row[1]} with type {type(row[1])} and value: {row[2]} with type {type(row[2])}")
if not isinstance(row[2], float) and check_not_nan(row[1]):
# if the lot number isn't a float and the reagent type isn't blank
# if not isinstance(row[2], float) and check_not_nan(row[1]):
if check_not_nan(row[1]):
# must be prefixed with 'lot_' to be recognized by gui
# This is no longer true since reagents are loaded into their own key in dictionary
try:
reagent_type = row[1].replace(' ', '_').lower().strip()
except AttributeError:
pass
# If there is a double slash in the type field, such as ethanol/iso
# Use the cell to the left for reagent type.
if reagent_type == "//":
if check_not_nan(row[2]):
reagent_type = row[0].replace(' ', '_').lower().strip()
else:
continue
try:
output_var = row[2].upper()
output_var = convert_nans_to_nones(str(row[2]).upper())
except AttributeError:
logger.debug(f"Couldn't upperize {row[2]}, must be a number")
output_var = row[2]
output_var = convert_nans_to_nones(str(row[2]))
logger.debug(f"Output variable is {output_var}")
logger.debug(f"Expiry date for imported reagent: {row[3]}")
if check_not_nan(row[3]):
@@ -149,22 +153,17 @@ class SheetParser(object):
expiry = datetime.fromordinal(datetime(1900, 1, 1).toordinal() + row[3] - 2)
else:
logger.debug(f"Date: {row[3]}")
expiry = date.today()
# expiry = date.today()
expiry = date(year=1970, month=1, day=1)
# self.sub[f"lot_{reagent_type}"] = {'lot':output_var, 'exp':expiry}
self.sub['reagents'].append(dict(type=reagent_type, lot=output_var, exp=expiry))
# self.sub['reagents'].append(dict(type=reagent_type, lot=output_var, exp=expiry))
self.sub['reagents'].append(PydReagent(type=reagent_type, lot=output_var, exp=expiry))
submission_info = self.parse_generic("Sample List")
# iloc is [row][column] and the first row is set as header row so -2
tech = str(submission_info.iloc[11][1])
# moved to pydantic model
# if tech == "nan":
# tech = "Unknown"
# elif len(tech.split(",")) > 1:
# tech_reg = re.compile(r"[A-Z]{2}")
# tech = ", ".join(tech_reg.findall(tech))
self.sub['technician'] = tech
self.sub['technician'] = str(submission_info.iloc[11][1])
# reagents
# must be prefixed with 'lot_' to be recognized by gui
# TODO: find a more adaptable way to read reagents.
# This is no longer true wince the creation of self.sub['reagents']
self.sub['reagents'] = []
reagent_range = submission_info.iloc[1:14, 4:8]
logger.debug(reagent_range)
@@ -175,7 +174,6 @@ class SheetParser(object):
logger.debug(f"Parser result: {self.sub}")
self.sample_result, self.sub['samples'] = sample_parse()
def parse_wastewater(self) -> None:
"""
pulls info specific to wastewater sample type
@@ -196,17 +194,18 @@ class SheetParser(object):
"""
# iterate through sub-df rows
for ii, row in df.iterrows():
if not isinstance(row[5], float) and check_not_nan(row[5]):
logger.debug(f"Parsing this row for reagents: {row}")
if check_not_nan(row[5]):
# must be prefixed with 'lot_' to be recognized by gui
# regex below will remove 80% from 80% ethanol in the Wastewater kit.
output_key = re.sub(r"^\d{1,3}%\s?", "", row[0].lower().strip().replace(' ', '_'))
output_key = output_key.strip("_")
# output_var is the lot number
try:
output_var = row[5].upper()
output_var = convert_nans_to_nones(str(row[5].upper()))
except AttributeError:
logger.debug(f"Couldn't upperize {row[5]}, must be a number")
output_var = row[5]
output_var = convert_nans_to_nones(str(row[5]))
if check_not_nan(row[7]):
try:
expiry = row[7].date()
@@ -214,8 +213,12 @@ class SheetParser(object):
expiry = date.today()
else:
expiry = date.today()
logger.debug(f"Expiry date for {output_key}: {expiry} of type {type(expiry)}")
# self.sub[f"lot_{output_key}"] = {'lot':output_var, 'exp':expiry}
self.sub['reagents'].append(dict(type=output_key, lot=output_var, exp=expiry))
# self.sub['reagents'].append(dict(type=output_key, lot=output_var, exp=expiry))
reagent = PydReagent(type=output_key, lot=output_var, exp=expiry)
logger.debug(f"Here is the created reagent: {reagent}")
self.sub['reagents'].append(reagent)
# parse submission sheet
submission_info = self.parse_generic("WW Submissions (ENTER HERE)")
# parse enrichment sheet
@@ -230,7 +233,7 @@ class SheetParser(object):
qprc_info = self.xl.parse("qPCR Worksheet", dtype=object)
# set qpcr reagent range
pcr_reagent_range = qprc_info.iloc[0:5, 9:20]
# compile technician info
# compile technician info from all sheets
self.sub['technician'] = f"Enr: {enrichment_info.columns[2]}, Ext: {extraction_info.columns[2]}, PCR: {qprc_info.columns[2]}"
self.sub['reagents'] = []
parse_reagents(enr_reagent_range)
@@ -242,7 +245,6 @@ class SheetParser(object):
self.sample_result, self.sub['samples'] = sample_parse()
self.sub['csv'] = self.xl.parse("Copy to import file", dtype=object)
def parse_wastewater_artic(self) -> None:
"""
pulls info specific to wastewater_arctic submission type
@@ -258,10 +260,10 @@ class SheetParser(object):
output_key = output_key.strip("_")
output_key = massage_common_reagents(output_key)
try:
output_var = row[1].upper()
output_var = convert_nans_to_nones(str(row[1].upper()))
except AttributeError:
logger.debug(f"Couldn't upperize {row[1]}, must be a number")
output_var = row[1]
output_var = convert_nans_to_nones(str(row[1]))
logger.debug(f"Output variable is {output_var}")
logger.debug(f"Expiry date for imported reagent: {row[2]}")
if check_not_nan(row[2]):
@@ -277,7 +279,8 @@ class SheetParser(object):
else:
logger.debug(f"Date: {row[2]}")
expiry = date.today()
self.sub['reagents'].append(dict(type=output_key, lot=output_var, exp=expiry))
# self.sub['reagents'].append(dict(type=output_key, lot=output_var, exp=expiry))
self.sub['reagents'].append(PydReagent(type=output_key, lot=output_var, exp=expiry))
else:
continue
def massage_samples(df:pd.DataFrame) -> pd.DataFrame:
@@ -317,20 +320,19 @@ class SheetParser(object):
sample_parse = getattr(sample_parser, f"parse_{self.sub['submission_type'].lower()}_samples")
self.sample_result, self.sub['samples'] = sample_parse()
def to_pydantic(self) -> PydSubmission:
"""
Generates a pydantic model of scraped data for validation
Returns:
PydSubmission: output pydantic model
"""
psm = PydSubmission(filepath=self.filepath, **self.sub)
"""
logger.debug(f"Submission dictionary coming into 'to_pydantic':\n{pprint.pformat(self.sub)}")
psm = PydSubmission(ctx=self.ctx, filepath=self.filepath, **self.sub)
delattr(psm, "filepath")
return psm
class SampleParser(object):
"""
object to pull data for samples in excel sheet and construct individual sample objects
@@ -385,7 +387,7 @@ class SampleParser(object):
list[WWSample]: list of sample objects
"""
def search_df_for_sample(sample_rsl:str):
# logger.debug(f"Attempting to find sample {sample_rsl} in \n {self.elution_map}")
logger.debug(f"Attempting to find sample {sample_rsl} in \n {self.elution_map}")
well = self.elution_map.where(self.elution_map==sample_rsl)
# logger.debug(f"Well: {well}")
well = well.dropna(how='all').dropna(axis=1, how="all")
@@ -394,9 +396,9 @@ class SampleParser(object):
logger.debug(f"well {sample_rsl} post processing: {well.size}: {type(well)}, {well.index[0]}, {well.columns[0]}")
self.elution_map.at[well.index[0], well.columns[0]] = np.nan
try:
col = str(int(well.columns[0]))
col = str(int(well.columns[0])).zfill(2)
except ValueError:
col = str(well.columns[0])
col = str(well.columns[0]).zfill(2)
except TypeError as e:
logger.error(f"Problem parsing out column number for {well}:\n {e}")
return f"{well.index[0]}{col}"
@@ -424,10 +426,12 @@ class SampleParser(object):
# new.testing_type = sample['Unnamed: 6']
# new.site_status = sample['Unnamed: 7']
new.notes = str(sample['Unnamed: 6']) # previously Unnamed: 8
new.well_number = sample['Unnamed: 1']
new.well_24 = sample['Unnamed: 1']
elu_well = search_df_for_sample(new.rsl_number)
if elu_well != None:
new.elution_well = elu_well
row = elu_well[0]
col = elu_well[1:].zfill(2)
new.well_number = f"{row}{col}"
else:
# try:
return_val += f"{new.rsl_number}\n"
@@ -455,12 +459,14 @@ class SampleParser(object):
missed_samples.append(sample['sample_name'])
continue
logger.debug(f"Got instance: {instance.ww_sample_full_id}")
if sample['well'] != None:
row = sample['well'][0]
col = sample['well'][1:].zfill(2)
sample['well'] = f"{row}{col}"
instance.artic_well_number = sample['well']
new_list.append(instance)
missed_str = "\n\t".join(missed_samples)
return f"Could not find matches for the following samples:\n\t {missed_str}", new_list
class PCRParser(object):
@@ -590,5 +596,5 @@ class PCRParser(object):
self.samples.append(sample_obj)

53
src/submissions/backend/excel/reports.py
View File

@@ -3,24 +3,14 @@ Contains functions for generating summary reports
'''
from pandas import DataFrame
import logging
from jinja2 import Environment, FileSystemLoader
from datetime import date, timedelta
import sys
from pathlib import Path
import re
from tools import check_if_app
from typing import Tuple
from configure import jinja_template_loading
logger = logging.getLogger(f"submissions.{__name__}")
# set path of templates depending on pyinstaller/raw python
# if getattr(sys, 'frozen', False):
if check_if_app():
loader_path = Path(sys._MEIPASS).joinpath("files", "templates")
else:
loader_path = Path(__file__).parents[2].joinpath('templates').absolute().__str__()
loader = FileSystemLoader(loader_path)
env = Environment(loader=loader)
env = jinja_template_loading()
logger = logging.getLogger(f"submissions.{__name__}")
@@ -115,7 +105,6 @@ def convert_data_list_to_df(ctx:dict, input:list[dict], subtype:str|None=None) -
# logger.debug(df)
# move date of sample submitted on same date as previous ahead one.
df = displace_date(df)
# df.sort_values('submitted_date').to_excel("controls.xlsx", engine="openpyxl")
# ad hoc method to make data labels more accurate.
df = df_column_renamer(df=df)
return df
@@ -156,46 +145,33 @@ def displace_date(df:DataFrame) -> DataFrame:
dict_list = [dict(name=item, date=df[df.name == item].iloc[0]['submitted_date']) for item in sorted(df['name'].unique())]
previous_dates = []
for _, item in enumerate(dict_list):
# try:
# # check = item['date'] == dict_list[ii-1]['date']
# check = item['date'] in previous_dates
# except IndexError:
# check = False
# if check:
# # occurences = previous_dates.count(item['date'])
# logger.debug(f"We found one! Increment date!\n\t{item['date']} to {item['date'] + timedelta(days=1)}")
# # get df locations where name == item name
# mask = df['name'] == item['name']
# # increment date in dataframe
# df.loc[mask, 'submitted_date'] = df.loc[mask, 'submitted_date'].apply(lambda x: x + timedelta(days=1))
# outdate = item['date'] + timedelta(days=1)
# # previous_dates.append(item['date'] + timedelta(days=1))
# else:
# outdate = item['date']
# previous_dates.append(outdate)
# logger.debug(f"\n\tCurrent date: {outdate}\n\tPrevious dates:{previous_dates}")
# logger.debug(type(item))
df, previous_dates = check_date(df=df, item=item, previous_dates=previous_dates)
return df
def check_date(df:DataFrame, item:dict, previous_dates:list) -> Tuple[DataFrame, list]:
"""
Checks if an items date is already present in df and adjusts df accordingly
Args:
df (DataFrame): input dataframe
item (dict): control for checking
previous_dates (list): list of dates found in previous controls
Returns:
Tuple[DataFrame, list]: Output dataframe and appended list of previous dates
"""
try:
# check = item['date'] == dict_list[ii-1]['date']
check = item['date'] in previous_dates
except IndexError:
check = False
previous_dates.append(item['date'])
if check:
# occurences = previous_dates.count(item['date'])
logger.debug(f"We found one! Increment date!\n\t{item['date']} to {item['date'] + timedelta(days=1)}")
# get df locations where name == item name
mask = df['name'] == item['name']
# increment date in dataframe
df.loc[mask, 'submitted_date'] = df.loc[mask, 'submitted_date'].apply(lambda x: x + timedelta(days=1))
item['date'] += timedelta(days=1)
# previous_dates.append(item['date'] + timedelta(days=1))
passed = False
else:
passed = True
@@ -249,8 +225,7 @@ def drop_reruns_from_df(ctx:dict, df: DataFrame) -> DataFrame:
# logger.debug(f"First run: {first_run}")
df = df.drop(df[df.name == first_run].index)
return df
# else:
# return df
def make_hitpicks(input:list) -> DataFrame:

144
src/submissions/backend/pydant/__init__.py
View File

@@ -1,65 +1,153 @@
import uuid
from pydantic import BaseModel, validator
from datetime import date
from pydantic import BaseModel, field_validator, model_validator, Extra
from datetime import date, datetime
from typing import List, Any
from tools import RSLNamer
from pathlib import Path
import re
import logging
from tools import check_not_nan, convert_nans_to_nones
import numpy as np
logger = logging.getLogger(f"submissions.{__name__}")
class PydSubmission(BaseModel):
class PydReagent(BaseModel):
type: str|None
lot: str|None
exp: date|None
@field_validator("type", mode='before')
@classmethod
def remove_undesired_types(cls, value):
match value:
case "atcc":
return None
case _:
return value
@field_validator("lot", mode='before')
@classmethod
def enforce_lot_string(cls, value):
if value != None:
return convert_nans_to_nones(str(value))
return value
@field_validator("exp", mode="before")
@classmethod
def enforce_date(cls, value):
if isinstance(value, float) or value == np.nan:
raise ValueError(f"Date cannot be a float: {value}")
else:
return value
class PydSubmission(BaseModel, extra=Extra.allow):
ctx: dict
filepath: Path
submission_type: str
submitter_plate_num: str|None
rsl_plate_num: str
rsl_plate_num: str|dict|None
submitted_date: date
submitting_lab: str
submitting_lab: str|None
sample_count: int
extraction_kit: str
technician: str
reagents: List[dict]
extraction_kit: str|dict|None
technician: str|None
reagents: List[PydReagent] = []
samples: List[Any]
@validator("submitted_date", pre=True)
# missing_fields: List[str] = []
@field_validator("submitted_date", mode="before")
@classmethod
def strip_datetime_string(cls, value):
if isinstance(value, datetime):
return value
if isinstance(value, date):
return value
return re.sub(r"_\d$", "", value)
@validator("submitter_plate_num")
@field_validator("submitter_plate_num")
@classmethod
def enforce_with_uuid(cls, value):
if value == None or value == "" or value == "None":
return uuid.uuid4().hex.upper()
@validator("rsl_plate_num", pre=True)
@classmethod
def rsl_from_file(cls, value, values):
if value == None:
logger.debug(f"Pydant values:\n{values}")
return RSLNamer(values['filepath'].__str__()).parsed_name
else:
return value
@validator("technician")
@field_validator("submitting_lab", mode="before")
@classmethod
def transform_nan(cls, value):
return convert_nans_to_nones(value)
@field_validator("rsl_plate_num", mode='before')
@classmethod
def rsl_from_file(cls, value, values):
logger.debug(f"RSL-plate initial value: {value}")
if check_not_nan(value):
if isinstance(value, str):
return dict(value=value, parsed=True)
else:
return value
else:
logger.debug(f"Pydant values:{type(values)}\n{values}")
return dict(value=RSLNamer(values.data['filepath'].__str__()).parsed_name, parsed=False)
@field_validator("technician")
@classmethod
def enforce_tech(cls, value):
if value == "nan" or value == "None":
value = "Unknown"
# elif len(value.split(",")) > 1:
# tech_reg = re.compile(r"\b[A-Z]{2}\b")
# value = ", ".join(tech_reg.findall(value))
return value
@validator("reagents")
@field_validator("reagents")
@classmethod
def remove_atcc(cls, value):
return_val = []
for reagent in value:
match reagent['type']:
case 'atcc':
continue
case _:
return_val.append(reagent)
logger.debug(f"Pydantic reagent: {reagent}")
# match reagent.type.lower():
# case 'atcc':
# continue
# case _:
# return_val.append(reagent)
if reagent.type == None:
continue
else:
return_val.append(reagent)
return return_val
@field_validator("sample_count", mode='before')
@classmethod
def enforce_sample_count(cls, value):
if check_not_nan(value):
return int(value)
else:
# raise ValueError(f"{value} could not be used to create an integer.")
return convert_nans_to_nones(value)
@field_validator("extraction_kit", mode='before')
@classmethod
def get_kit_if_none(cls, value, values):
from frontend.custom_widgets.pop_ups import KitSelector
if check_not_nan(value):
return dict(value=value, parsed=True)
else:
# logger.debug(values.data)
dlg = KitSelector(ctx=values.data['ctx'], title="Kit Needed", message="At minimum a kit is needed. Please select one.")
if dlg.exec():
return dict(value=dlg.getValues(), parsed=False)
else:
raise ValueError("Extraction kit needed.")
# @model_validator(mode="after")
# def ensure_kit(cls, values):
# logger.debug(f"Model values: {values}")
# missing_fields = [k for k,v in values if v == None]
# if len(missing_fields) > 0:
# logger.debug(f"Missing fields: {missing_fields}")
# values['missing_fields'] = missing_fields
# return values

20
src/submissions/configure/__init__.py
View File

@@ -1,6 +1,7 @@
'''
Contains functions for setting up program from config.yml and database.
'''
from jinja2 import Environment, FileSystemLoader
import yaml
import sys, os, stat, platform, getpass
import logging
@@ -238,3 +239,22 @@ def copy_settings(settings_path:Path, settings:dict) -> dict:
with open(settings_path, 'w') as f:
yaml.dump(settings, f)
return settings
def jinja_template_loading():
"""
Returns jinja2 template environment.
Returns:
_type_: _description_
"""
# determine if pyinstaller launcher is being used
if getattr(sys, 'frozen', False):
loader_path = Path(sys._MEIPASS).joinpath("files", "templates")
else:
loader_path = Path(__file__).parents[1].joinpath('templates').absolute().__str__()
# jinja template loading
loader = FileSystemLoader(loader_path)
env = Environment(loader=loader)
return env

442
src/submissions/frontend/__init__.py
View File

@@ -10,14 +10,12 @@ from PyQt6.QtWidgets import (
)
from PyQt6.QtGui import QAction
from PyQt6.QtWebEngineWidgets import QWebEngineView
# import pandas as pd
from pathlib import Path
from backend.db import (
construct_reagent, get_all_Control_Types_names, get_all_available_modes, store_reagent
)
from .main_window_functions import *
from tools import check_if_app
# from backend.excel.reports import
from frontend.custom_widgets import SubmissionsSheet, AlertPop, AddReagentForm, KitAdder, ControlsDatePicker
import logging
from datetime import date
@@ -31,7 +29,7 @@ class App(QMainWindow):
def __init__(self, ctx: dict = {}):
super().__init__()
self.ctx = ctx
# indicate version and database connected in title bar
# indicate version and connected database in title bar
try:
self.title = f"Submissions App (v{ctx['package'].__version__}) - {ctx['database']}"
except (AttributeError, KeyError):
@@ -123,11 +121,17 @@ class App(QMainWindow):
self.docsAction.triggered.connect(self.openDocs)
def showAbout(self):
"""
Show the 'about' message
"""
output = f"Version: {self.ctx['package'].__version__}\n\nAuthor: {self.ctx['package'].__author__['name']} - {self.ctx['package'].__author__['email']}\n\nCopyright: {self.ctx['package'].__copyright__}"
about = AlertPop(message=output, status="information")
about.exec()
def openDocs(self):
"""
Open the documentation html pages
"""
if check_if_app():
url = Path(sys._MEIPASS).joinpath("files", "docs", "index.html")
else:
@@ -135,8 +139,13 @@ class App(QMainWindow):
logger.debug(f"Attempting to open {url}")
webbrowser.get('windows-default').open(f"file://{url.__str__()}")
# All main window functions return a result which is reported here, unless it is None
def result_reporter(self, result:dict|None=None):
"""
Report any anomolous results - if any - to the user
Args:
result (dict | None, optional): The result from a function. Defaults to None.
"""
if result != None:
msg = AlertPop(message=result['message'], status=result['status'])
msg.exec()
@@ -149,14 +158,12 @@ class App(QMainWindow):
logger.debug(f"Import result: {result}")
self.result_reporter(result)
def kit_reload(self):
"""
Removes all reagents from form before running kit integrity completion.
"""
self, result = kit_reload_function(self)
self.result_reporter(result)
def kit_integrity_completion(self):
"""
@@ -166,7 +173,6 @@ class App(QMainWindow):
"""
self, result = kit_integrity_completion_function(self)
self.result_reporter(result)
def submit_new_sample(self):
"""
@@ -174,82 +180,6 @@ class App(QMainWindow):
"""
self, result = submit_new_sample_function(self)
self.result_reporter(result)
# get info from form
# info = extract_form_info(self.table_widget.tab1)
# reagents = {k:v for k,v in info.items() if k.startswith("lot_")}
# info = {k:v for k,v in info.items() if not k.startswith("lot_")}
# logger.debug(f"Info: {info}")
# logger.debug(f"Reagents: {reagents}")
# parsed_reagents = []
# # compare reagents in form to reagent database
# for reagent in reagents:
# wanted_reagent = lookup_reagent(ctx=self.ctx, reagent_lot=reagents[reagent])
# logger.debug(f"Looked up reagent: {wanted_reagent}")
# # if reagent not found offer to add to database
# if wanted_reagent == None:
# r_lot = reagents[reagent]
# dlg = QuestionAsker(title=f"Add {r_lot}?", message=f"Couldn't find reagent type {reagent.replace('_', ' ').title().strip('Lot')}: {r_lot} in the database.\n\nWould you like to add it?")
# if dlg.exec():
# logger.debug(f"checking reagent: {reagent} in self.reagents. Result: {self.reagents[reagent]}")
# expiry_date = self.reagents[reagent]['exp']
# wanted_reagent = self.add_reagent(reagent_lot=r_lot, reagent_type=reagent.replace("lot_", ""), expiry=expiry_date)
# else:
# # In this case we will have an empty reagent and the submission will fail kit integrity check
# logger.debug("Will not add reagent.")
# if wanted_reagent != None:
# parsed_reagents.append(wanted_reagent)
# # move samples into preliminary submission dict
# info['samples'] = self.samples
# info['uploaded_by'] = getuser()
# # construct submission object
# logger.debug(f"Here is the info_dict: {pprint.pformat(info)}")
# base_submission, result = construct_submission_info(ctx=self.ctx, info_dict=info)
# # check output message for issues
# match result['code']:
# # code 1: ask for overwrite
# case 1:
# dlg = QuestionAsker(title=f"Review {base_submission.rsl_plate_num}?", message=result['message'])
# if dlg.exec():
# base_submission.reagents = []
# else:
# return
# # code 2: No RSL plate number given
# case 2:
# dlg = AlertPop(message=result['message'], status='critical')
# dlg.exec()
# return
# case _:
# pass
# # add reagents to submission object
# for reagent in parsed_reagents:
# base_submission.reagents.append(reagent)
# logger.debug("Checking kit integrity...")
# kit_integrity = check_kit_integrity(base_submission)
# if kit_integrity != None:
# msg = AlertPop(message=kit_integrity['message'], status="critical")
# msg.exec()
# return
# logger.debug(f"Sending submission: {base_submission.rsl_plate_num} to database.")
# result = store_submission(ctx=self.ctx, base_submission=base_submission)
# # check result of storing for issues
# if result != None:
# msg = AlertPop(result['message'])
# msg.exec()
# # update summary sheet
# self.table_widget.sub_wid.setData()
# # reset form
# for item in self.table_widget.formlayout.parentWidget().findChildren(QWidget):
# item.setParent(None)
# if hasattr(self, 'csv'):
# dlg = QuestionAsker("Export CSV?", "Would you like to export the csv file?")
# if dlg.exec():
# home_dir = Path(self.ctx["directory_path"]).joinpath(f"{base_submission.rsl_plate_num}.csv").resolve().__str__()
# fname = Path(QFileDialog.getSaveFileName(self, "Save File", home_dir, filter=".csv")[0])
# try:
# self.csv.to_csv(fname.__str__(), index=False)
# except PermissionError:
# logger.debug(f"Could not get permissions to {fname}. Possibly the request was cancelled.")
def add_reagent(self, reagent_lot:str|None=None, reagent_type:str|None=None, expiry:date|None=None):
"""
@@ -277,7 +207,6 @@ class App(QMainWindow):
# send reagent to db
store_reagent(ctx=self.ctx, reagent=reagent)
return reagent
def generateReport(self):
"""
@@ -285,42 +214,6 @@ class App(QMainWindow):
"""
self, result = generate_report_function(self)
self.result_reporter(result)
# Custom two date picker for start & end dates
# dlg = ReportDatePicker()
# if dlg.exec():
# info = extract_form_info(dlg)
# logger.debug(f"Report info: {info}")
# # find submissions based on date range
# subs = lookup_submissions_by_date_range(ctx=self.ctx, start_date=info['start_date'], end_date=info['end_date'])
# # convert each object to dict
# records = [item.report_dict() for item in subs]
# # make dataframe from record dictionaries
# df = make_report_xlsx(records=records)
# html = make_report_html(df=df, start_date=info['start_date'], end_date=info['end_date'])
# # setup filedialog to handle save location of report
# home_dir = Path(self.ctx["directory_path"]).joinpath(f"Submissions_Report_{info['start_date']}-{info['end_date']}.pdf").resolve().__str__()
# fname = Path(QFileDialog.getSaveFileName(self, "Save File", home_dir, filter=".pdf")[0])
# # logger.debug(f"report output name: {fname}")
# with open(fname, "w+b") as f:
# pisa.CreatePDF(html, dest=f)
# writer = pd.ExcelWriter(fname.with_suffix(".xlsx"), engine='openpyxl')
# df.to_excel(writer, sheet_name="Report")
# worksheet = writer.sheets['Report']
# for idx, col in enumerate(df): # loop through all columns
# series = df[col]
# max_len = max((
# series.astype(str).map(len).max(), # len of largest item
# len(str(series.name)) # len of column name/header
# )) + 20 # adding a little extra space
# try:
# worksheet.column_dimensions[get_column_letter(idx)].width = max_len
# except ValueError:
# pass
# for cell in worksheet['D']:
# if cell.row > 1:
# cell.style = 'Currency'
# writer.close()
def add_kit(self):
"""
@@ -328,29 +221,6 @@ class App(QMainWindow):
"""
self, result = add_kit_function(self)
self.result_reporter(result)
# setup file dialog to find yaml flie
# home_dir = str(Path(self.ctx["directory_path"]))
# fname = Path(QFileDialog.getOpenFileName(self, 'Open file', home_dir, filter = "yml(*.yml)")[0])
# assert fname.exists()
# # read yaml file
# try:
# with open(fname.__str__(), "r") as stream:
# try:
# exp = yaml.load(stream, Loader=yaml.Loader)
# except yaml.YAMLError as exc:
# logger.error(f'Error reading yaml file {fname}: {exc}')
# return {}
# except PermissionError:
# return
# # send to kit creator function
# result = create_kit_from_yaml(ctx=self.ctx, exp=exp)
# match result['code']:
# case 0:
# msg = AlertPop(message=result['message'], status='info')
# case 1:
# msg = AlertPop(message=result['message'], status='critical')
# msg.exec()
def add_org(self):
"""
@@ -358,29 +228,6 @@ class App(QMainWindow):
"""
self, result = add_org_function(self)
self.result_reporter(result)
# # setup file dialog to find yaml flie
# home_dir = str(Path(self.ctx["directory_path"]))
# fname = Path(QFileDialog.getOpenFileName(self, 'Open file', home_dir, filter = "yml(*.yml)")[0])
# assert fname.exists()
# # read yaml file
# try:
# with open(fname.__str__(), "r") as stream:
# try:
# org = yaml.load(stream, Loader=yaml.Loader)
# except yaml.YAMLError as exc:
# logger.error(f'Error reading yaml file {fname}: {exc}')
# return {}
# except PermissionError:
# return
# # send to kit creator function
# result = create_org_from_yaml(ctx=self.ctx, org=org)
# match result['code']:
# case 0:
# msg = AlertPop(message=result['message'], status='information')
# case 1:
# msg = AlertPop(message=result['message'], status='critical')
# msg.exec()
def _controls_getter(self):
"""
@@ -388,39 +235,6 @@ class App(QMainWindow):
"""
self, result = controls_getter_function(self)
self.result_reporter(result)
# # subtype defaults to disabled
# try:
# self.table_widget.sub_typer.disconnect()
# except TypeError:
# pass
# # correct start date being more recent than end date and rerun
# if self.table_widget.datepicker.start_date.date() > self.table_widget.datepicker.end_date.date():
# logger.warning("Start date after end date is not allowed!")
# threemonthsago = self.table_widget.datepicker.end_date.date().addDays(-60)
# # block signal that will rerun controls getter and set start date
# with QSignalBlocker(self.table_widget.datepicker.start_date) as blocker:
# self.table_widget.datepicker.start_date.setDate(threemonthsago)
# self._controls_getter()
# return
# # convert to python useable date object
# self.start_date = self.table_widget.datepicker.start_date.date().toPyDate()
# self.end_date = self.table_widget.datepicker.end_date.date().toPyDate()
# self.con_type = self.table_widget.control_typer.currentText()
# self.mode = self.table_widget.mode_typer.currentText()
# self.table_widget.sub_typer.clear()
# # lookup subtypes
# sub_types = get_control_subtypes(ctx=self.ctx, type=self.con_type, mode=self.mode)
# if sub_types != []:
# # block signal that will rerun controls getter and update sub_typer
# with QSignalBlocker(self.table_widget.sub_typer) as blocker:
# self.table_widget.sub_typer.addItems(sub_types)
# self.table_widget.sub_typer.setEnabled(True)
# self.table_widget.sub_typer.currentTextChanged.connect(self._chart_maker)
# else:
# self.table_widget.sub_typer.clear()
# self.table_widget.sub_typer.setEnabled(False)
# self._chart_maker()
def _chart_maker(self):
"""
@@ -428,42 +242,6 @@ class App(QMainWindow):
"""
self, result = chart_maker_function(self)
self.result_reporter(result)
# logger.debug(f"Control getter context: \n\tControl type: {self.con_type}\n\tMode: {self.mode}\n\tStart Date: {self.start_date}\n\tEnd Date: {self.end_date}")
# if self.table_widget.sub_typer.currentText() == "":
# self.subtype = None
# else:
# self.subtype = self.table_widget.sub_typer.currentText()
# logger.debug(f"Subtype: {self.subtype}")
# # query all controls using the type/start and end dates from the gui
# controls = get_all_controls_by_type(ctx=self.ctx, con_type=self.con_type, start_date=self.start_date, end_date=self.end_date)
# # if no data found from query set fig to none for reporting in webview
# if controls == None:
# fig = None
# else:
# # change each control to list of dicts
# data = [control.convert_by_mode(mode=self.mode) for control in controls]
# # flatten data to one dimensional list
# data = [item for sublist in data for item in sublist]
# # send to dataframe creator
# df = convert_data_list_to_df(ctx=self.ctx, input=data, subtype=self.subtype)
# if self.subtype == None:
# title = self.mode
# else:
# title = f"{self.mode} - {self.subtype}"
# # send dataframe to chart maker
# fig = create_charts(ctx=self.ctx, df=df, ytitle=title)
# logger.debug(f"Updating figure...")
# # construct html for webview
# html = '<html><body>'
# if fig != None:
# html += plotly.offline.plot(fig, output_type='div', include_plotlyjs='cdn')#, image = 'png', auto_open=True, image_filename='plot_image')
# else:
# html += "<h1>No data was retrieved for the given parameters.</h1>"
# html += '</body></html>'
# self.table_widget.webengineview.setHtml(html)
# self.table_widget.webengineview.update()
# logger.debug("Figure updated... I hope.")
def linkControls(self):
"""
@@ -471,52 +249,6 @@ class App(QMainWindow):
"""
self, result = link_controls_function(self)
self.result_reporter(result)
# all_bcs = lookup_all_submissions_by_type(self.ctx, "Bacterial Culture")
# logger.debug(all_bcs)
# all_controls = get_all_controls(self.ctx)
# ac_list = [control.name for control in all_controls]
# count = 0
# for bcs in all_bcs:
# logger.debug(f"Running for {bcs.rsl_plate_num}")
# logger.debug(f"Here is the current control: {[control.name for control in bcs.controls]}")
# samples = [sample.sample_id for sample in bcs.samples]
# logger.debug(bcs.controls)
# for sample in samples:
# # replace below is a stopgap method because some dingus decided to add spaces in some of the ATCC49... so it looks like "ATCC 49"...
# if " " in sample:
# logger.warning(f"There is not supposed to be a space in the sample name!!!")
# sample = sample.replace(" ", "")
# # if sample not in ac_list:
# if not any([ac.startswith(sample) for ac in ac_list]):
# continue
# else:
# for control in all_controls:
# diff = difflib.SequenceMatcher(a=sample, b=control.name).ratio()
# if control.name.startswith(sample):
# logger.debug(f"Checking {sample} against {control.name}... {diff}")
# logger.debug(f"Found match:\n\tSample: {sample}\n\tControl: {control.name}\n\tDifference: {diff}")
# if control in bcs.controls:
# logger.debug(f"{control.name} already in {bcs.rsl_plate_num}, skipping")
# continue
# else:
# logger.debug(f"Adding {control.name} to {bcs.rsl_plate_num} as control")
# bcs.controls.append(control)
# # bcs.control_id.append(control.id)
# control.submission = bcs
# control.submission_id = bcs.id
# self.ctx["database_session"].add(control)
# count += 1
# self.ctx["database_session"].add(bcs)
# logger.debug(f"Here is the new control: {[control.name for control in bcs.controls]}")
# result = f"We added {count} controls to bacterial cultures."
# logger.debug(result)
# self.ctx['database_session'].commit()
# msg = QMessageBox()
# msg.setText("Controls added")
# msg.setInformativeText(result)
# msg.setWindowTitle("Controls added")
# msg.exec()
def linkExtractions(self):
"""
@@ -524,55 +256,6 @@ class App(QMainWindow):
"""
self, result = link_extractions_function(self)
self.result_reporter(result)
# home_dir = str(Path(self.ctx["directory_path"]))
# fname = Path(QFileDialog.getOpenFileName(self, 'Open file', home_dir, filter = "csv(*.csv)")[0])
# with open(fname.__str__(), 'r') as f:
# runs = [col.strip().split(",") for col in f.readlines()]
# count = 0
# for run in runs:
# obj = dict(
# start_time=run[0].strip(),
# rsl_plate_num=run[1].strip(),
# sample_count=run[2].strip(),
# status=run[3].strip(),
# experiment_name=run[4].strip(),
# end_time=run[5].strip()
# )
# for ii in range(6, len(run)):
# obj[f"column{str(ii-5)}_vol"] = run[ii]
# sub = lookup_submission_by_rsl_num(ctx=self.ctx, rsl_num=obj['rsl_plate_num'])
# try:
# logger.debug(f"Found submission: {sub.rsl_plate_num}")
# count += 1
# except AttributeError:
# continue
# if sub.extraction_info != None:
# existing = json.loads(sub.extraction_info)
# else:
# existing = None
# try:
# if json.dumps(obj) in sub.extraction_info:
# logger.debug(f"Looks like we already have that info.")
# continue
# except TypeError:
# pass
# if existing != None:
# try:
# logger.debug(f"Updating {type(existing)}: {existing} with {type(obj)}: {obj}")
# existing.append(obj)
# logger.debug(f"Setting: {existing}")
# sub.extraction_info = json.dumps(existing)
# except TypeError:
# logger.error(f"Error updating!")
# sub.extraction_info = json.dumps([obj])
# logger.debug(f"Final ext info for {sub.rsl_plate_num}: {sub.extraction_info}")
# else:
# sub.extraction_info = json.dumps([obj])
# self.ctx['database_session'].add(sub)
# self.ctx["database_session"].commit()
# dlg = AlertPop(message=f"We added {count} logs to the database.", status='information')
# dlg.exec()
def linkPCR(self):
"""
@@ -580,56 +263,6 @@ class App(QMainWindow):
"""
self, result = link_pcr_function(self)
self.result_reporter(result)
# home_dir = str(Path(self.ctx["directory_path"]))
# fname = Path(QFileDialog.getOpenFileName(self, 'Open file', home_dir, filter = "csv(*.csv)")[0])
# with open(fname.__str__(), 'r') as f:
# runs = [col.strip().split(",") for col in f.readlines()]
# count = 0
# for run in runs:
# obj = dict(
# start_time=run[0].strip(),
# rsl_plate_num=run[1].strip(),
# biomek_status=run[2].strip(),
# quant_status=run[3].strip(),
# experiment_name=run[4].strip(),
# end_time=run[5].strip()
# )
# # for ii in range(6, len(run)):
# # obj[f"column{str(ii-5)}_vol"] = run[ii]
# sub = lookup_submission_by_rsl_num(ctx=self.ctx, rsl_num=obj['rsl_plate_num'])
# try:
# logger.debug(f"Found submission: {sub.rsl_plate_num}")
# except AttributeError:
# continue
# if hasattr(sub, 'pcr_info') and sub.pcr_info != None:
# existing = json.loads(sub.pcr_info)
# else:
# existing = None
# try:
# if json.dumps(obj) in sub.pcr_info:
# logger.debug(f"Looks like we already have that info.")
# continue
# else:
# count += 1
# except TypeError:
# logger.error(f"No json to dump")
# if existing != None:
# try:
# logger.debug(f"Updating {type(existing)}: {existing} with {type(obj)}: {obj}")
# existing.append(obj)
# logger.debug(f"Setting: {existing}")
# sub.pcr_info = json.dumps(existing)
# except TypeError:
# logger.error(f"Error updating!")
# sub.pcr_info = json.dumps([obj])
# logger.debug(f"Final ext info for {sub.rsl_plate_num}: {sub.pcr_info}")
# else:
# sub.pcr_info = json.dumps([obj])
# self.ctx['database_session'].add(sub)
# self.ctx["database_session"].commit()
# dlg = AlertPop(message=f"We added {count} logs to the database.", status='information')
# dlg.exec()
def importPCRResults(self):
"""
@@ -637,54 +270,6 @@ class App(QMainWindow):
"""
self, result = import_pcr_results_function(self)
self.result_reporter(result)
# home_dir = str(Path(self.ctx["directory_path"]))
# fname = Path(QFileDialog.getOpenFileName(self, 'Open file', home_dir, filter = "xlsx(*.xlsx)")[0])
# parser = PCRParser(ctx=self.ctx, filepath=fname)
# logger.debug(f"Attempting lookup for {parser.plate_num}")
# sub = lookup_submission_by_rsl_num(ctx=self.ctx, rsl_num=parser.plate_num)
# try:
# logger.debug(f"Found submission: {sub.rsl_plate_num}")
# except AttributeError:
# logger.error(f"Submission of number {parser.plate_num} not found. Attempting rescue of plate repeat.")
# parser.plate_num = "-".join(parser.plate_num.split("-")[:-1])
# sub = lookup_submission_by_rsl_num(ctx=self.ctx, rsl_num=parser.plate_num)
# try:
# logger.debug(f"Found submission: {sub.rsl_plate_num}")
# except AttributeError:
# logger.error(f"Rescue of {parser.plate_num} failed.")
# return
# # jout = json.dumps(parser.pcr)
# count = 0
# if hasattr(sub, 'pcr_info') and sub.pcr_info != None:
# existing = json.loads(sub.pcr_info)
# else:
# # jout = None
# existing = None
# if existing != None:
# try:
# logger.debug(f"Updating {type(existing)}: {existing} with {type(parser.pcr)}: {parser.pcr}")
# if json.dumps(parser.pcr) not in sub.pcr_info:
# existing.append(parser.pcr)
# logger.debug(f"Setting: {existing}")
# sub.pcr_info = json.dumps(existing)
# except TypeError:
# logger.error(f"Error updating!")
# sub.pcr_info = json.dumps([parser.pcr])
# logger.debug(f"Final pcr info for {sub.rsl_plate_num}: {sub.pcr_info}")
# else:
# sub.pcr_info = json.dumps([parser.pcr])
# self.ctx['database_session'].add(sub)
# logger.debug(f"Existing {type(sub.pcr_info)}: {sub.pcr_info}")
# logger.debug(f"Inserting {type(json.dumps(parser.pcr))}: {json.dumps(parser.pcr)}")
# self.ctx["database_session"].commit()
# logger.debug(f"Got {len(parser.samples)} to update!")
# for sample in parser.samples:
# logger.debug(f"Running update on: {sample['sample']}")
# sample['plate_rsl'] = sub.rsl_plate_num
# update_ww_sample(ctx=self.ctx, sample_obj=sample)
# dlg = AlertPop(message=f"We added PCR info to {sub.rsl_plate_num}.", status='information')
# dlg.exec()
class AddSubForm(QWidget):
@@ -756,4 +341,3 @@ class AddSubForm(QWidget):
self.layout.addWidget(self.tabs)
self.setLayout(self.layout)

35
src/submissions/frontend/all_window_functions.py
View File

@@ -1,21 +1,40 @@
from pathlib import Path
import logging
from PyQt6.QtWidgets import (
QMainWindow, QLabel, QToolBar,
QTabWidget, QWidget, QVBoxLayout,
QPushButton, QFileDialog,
QLineEdit, QMessageBox, QComboBox, QDateEdit, QHBoxLayout,
QSpinBox, QDoubleSpinBox, QScrollArea
QMainWindow, QWidget, QFileDialog,
QLineEdit, QComboBox, QDateEdit, QSpinBox,
QDoubleSpinBox
)
logger = logging.getLogger(f"submissions.{__name__}")
def select_open_file(obj:QMainWindow, extension:str) -> Path:
def select_open_file(obj:QMainWindow, file_extension:str) -> Path:
"""
File dialog to select a file to read from
Args:
obj (QMainWindow): Original main app window to be parent
file_extension (str): file extension
Returns:
Path: Path of file to be opened
"""
home_dir = str(Path(obj.ctx["directory_path"]))
fname = Path(QFileDialog.getOpenFileName(obj, 'Open file', home_dir, filter = f"{extension}(*.{extension})")[0])
fname = Path(QFileDialog.getOpenFileName(obj, 'Open file', home_dir, filter = f"{file_extension}(*.{file_extension})")[0])
return fname
def select_save_file(obj:QMainWindow, default_name:str, extension:str) -> Path:
"""
File dialog to select a file to write to
Args:
obj (QMainWindow): Original main app window to be parent
default_name (str): default base file name
extension (str): file extension
Returns:
Path: Path of file to be opened
"""
home_dir = Path(obj.ctx["directory_path"]).joinpath(default_name).resolve().__str__()
fname = Path(QFileDialog.getSaveFileName(obj, "Save File", home_dir, filter = f"{extension}(*.{extension})")[0])
return fname
@@ -41,7 +60,7 @@ def extract_form_info(object) -> dict:
except AttributeError:
all_children = object.layout().parentWidget().findChildren(QWidget)
for item in all_children:
logger.debug(f"Looking at: {item.objectName()}")
logger.debug(f"Looking at: {item.objectName()}: {type(item)}")
match item:
case QLineEdit():
dicto[item.objectName()] = item.text()

212
src/submissions/frontend/custom_widgets/misc.py
View File

@@ -2,40 +2,27 @@
Contains miscellaneous widgets for frontend functions
'''
from datetime import date
import difflib
from typing import Tuple
from PyQt6.QtWidgets import (
QLabel, QVBoxLayout,
QLineEdit, QComboBox, QDialog,
QDialogButtonBox, QDateEdit, QSizePolicy, QWidget,
QGridLayout, QPushButton, QSpinBox, QDoubleSpinBox,
QHBoxLayout, QMainWindow
QHBoxLayout
)
from PyQt6.QtCore import Qt, QDate, QSize
# from submissions.backend.db.functions import lookup_kittype_by_use
# from submissions.backend.db import lookup_regent_by_type_name_and_kit_name
from tools import check_not_nan
from ..all_window_functions import extract_form_info
from backend.db import get_all_reagenttype_names, lookup_all_sample_types, create_kit_from_yaml, \
lookup_regent_by_type_name, lookup_kittype_by_use, lookup_all_orgs
#, lookup_regent_by_type_name_and_kit_name
from backend.excel.parser import SheetParser
from jinja2 import Environment, FileSystemLoader
import sys
from pathlib import Path
lookup_regent_by_type_name, lookup_last_used_reagenttype_lot
from configure import jinja_template_loading
import logging
import numpy as np
from .pop_ups import AlertPop
from backend.pydant import PydReagent
logger = logging.getLogger(f"submissions.{__name__}")
if getattr(sys, 'frozen', False):
loader_path = Path(sys._MEIPASS).joinpath("files", "templates")
else:
loader_path = Path(__file__).parents[2].joinpath('templates').absolute().__str__()
loader = FileSystemLoader(loader_path)
env = Environment(loader=loader)
env = jinja_template_loading()
class AddReagentForm(QDialog):
"""
@@ -210,45 +197,6 @@ class KitAdder(QWidget):
msg = AlertPop(message=result['message'], status=result['status'])
msg.exec()
# def extract_form_info(self, object):
# """
# retrieves arbitrary number of labels, values from form
# Args:
# object (_type_): the object to extract info from
# Returns:
# _type_: _description_
# """
# labels = []
# values = []
# reagents = {}
# for item in object.findChildren(QWidget):
# logger.debug(item.parentWidget())
# # if not isinstance(item.parentWidget(), ReagentTypeForm):
# match item:
# case QLabel():
# labels.append(item.text().replace(" ", "_").strip(":").lower())
# case QLineEdit():
# # ad hoc check to prevent double reporting of qdatedit under lineedit for some reason
# if not isinstance(prev_item, QDateEdit) and not isinstance(prev_item, QComboBox) and not isinstance(prev_item, QSpinBox) and not isinstance(prev_item, QScrollBar):
# logger.debug(f"Previous: {prev_item}")
# logger.debug(f"Item: {item}, {item.text()}")
# values.append(item.text().strip())
# case QComboBox():
# values.append(item.currentText().strip())
# case QDateEdit():
# values.append(item.date().toPyDate())
# case QSpinBox():
# values.append(item.value())
# case ReagentTypeForm():
# re_labels, re_values, _ = self.extract_form_info(item)
# reagent = {item[0]:item[1] for item in zip(re_labels, re_values)}
# logger.debug(reagent)
# # reagent = {reagent['name:']:{'eol':reagent['extension_of_life_(months):']}}
# reagents[reagent["name_(*exactly*_as_it_appears_in_the_excel_submission_form)"].strip()] = {'eol_ext':int(reagent['extension_of_life_(months)'])}
# prev_item = item
# return labels, values, reagents
class ReagentTypeForm(QWidget):
"""
@@ -300,79 +248,24 @@ class ControlsDatePicker(QWidget):
return QSize(80,20)
# class ImportReagent(QComboBox):
# def __init__(self, ctx:dict, item:str, prsr:SheetParser|None=None):
# super().__init__()
# self.setEditable(True)
# # Ensure that all reagenttypes have a name that matches the items in the excel parser
# query_var = item.replace("lot_", "")
# if prsr != None:
# logger.debug(f"Import Reagent is looking at: {prsr.sub[item]} for {item}")
# else:
# logger.debug(f"Import Reagent is going to retrieve all reagents for {item}")
# logger.debug(f"Query for: {query_var}")
# if prsr != None:
# if isinstance(prsr.sub[item], np.float64):
# logger.debug(f"{prsr.sub[item]['lot']} is a numpy float!")
# try:
# prsr.sub[item] = int(prsr.sub[item]['lot'])
# except ValueError:
# pass
# # query for reagents using type name from sheet and kit from sheet
# logger.debug(f"Attempting lookup of reagents by type: {query_var}")
# # below was lookup_reagent_by_type_name_and_kit_name, but I couldn't get it to work.
# relevant_reagents = [item.__str__() for item in lookup_regent_by_type_name(ctx=ctx, type_name=query_var)]#, kit_name=prsr.sub['extraction_kit'])]
# # relevant_reagents = [item.__str__() for item in lookup_regent_by_type_name_and_kit_name(ctx=ctx, type_name=query_var, kit_name=prsr.sub['extraction_kit'])]
# output_reg = []
# for reagent in relevant_reagents:
# # extract strings from any sets.
# if isinstance(reagent, set):
# for thing in reagent:
# output_reg.append(thing)
# elif isinstance(reagent, str):
# output_reg.append(reagent)
# relevant_reagents = output_reg
# # if reagent in sheet is not found insert it into the front of relevant reagents so it shows
# if prsr != None:
# logger.debug(f"Relevant reagents for {prsr.sub[item]}: {relevant_reagents}")
# if str(prsr.sub[item]['lot']) not in relevant_reagents:
# if check_not_nan(prsr.sub[item]['lot']):
# relevant_reagents.insert(0, str(prsr.sub[item]['lot']))
# else:
# if len(relevant_reagents) > 1:
# logger.debug(f"Found {prsr.sub[item]['lot']} in relevant reagents: {relevant_reagents}. Moving to front of list.")
# idx = relevant_reagents.index(str(prsr.sub[item]['lot']))
# logger.debug(f"The index we got for {prsr.sub[item]['lot']} in {relevant_reagents} was {idx}")
# moved_reag = relevant_reagents.pop(idx)
# relevant_reagents.insert(0, moved_reag)
# else:
# logger.debug(f"Found {prsr.sub[item]['lot']} in relevant reagents: {relevant_reagents}. But no need to move due to short list.")
# logger.debug(f"New relevant reagents: {relevant_reagents}")
# self.setObjectName(f"lot_{item}")
# self.addItems(relevant_reagents)
class ImportReagent(QComboBox):
def __init__(self, ctx:dict, reagent:dict):
def __init__(self, ctx:dict, reagent:PydReagent):
super().__init__()
self.setEditable(True)
# Ensure that all reagenttypes have a name that matches the items in the excel parser
query_var = reagent['type']
logger.debug(f"Import Reagent is looking at: {reagent['lot']} for {reagent['type']}")
if isinstance(reagent['lot'], np.float64):
logger.debug(f"{reagent['lot']} is a numpy float!")
query_var = reagent.type
logger.debug(f"Import Reagent is looking at: {reagent.lot} for {reagent.type}")
if isinstance(reagent.lot, np.float64):
logger.debug(f"{reagent.lot} is a numpy float!")
try:
reagent['lot'] = int(reagent['lot'])
reagent.lot = int(reagent.lot)
except ValueError:
pass
# query for reagents using type name from sheet and kit from sheet
logger.debug(f"Attempting lookup of reagents by type: {query_var}")
# below was lookup_reagent_by_type_name_and_kit_name, but I couldn't get it to work.
relevant_reagents = [item.__str__() for item in lookup_regent_by_type_name(ctx=ctx, type_name=query_var)]#, kit_name=prsr.sub['extraction_kit'])]
# relevant_reagents = [item.__str__() for item in lookup_regent_by_type_name_and_kit_name(ctx=ctx, type_name=query_var, kit_name=prsr.sub['extraction_kit'])]
relevant_reagents = [item.__str__() for item in lookup_regent_by_type_name(ctx=ctx, type_name=query_var)]
output_reg = []
for rel_reagent in relevant_reagents:
# extract strings from any sets.
@@ -383,20 +276,79 @@ class ImportReagent(QComboBox):
output_reg.append(rel_reagent)
relevant_reagents = output_reg
# if reagent in sheet is not found insert it into the front of relevant reagents so it shows
# if prsr != None:
logger.debug(f"Relevant reagents for {reagent['lot']}: {relevant_reagents}")
if str(reagent['lot']) not in relevant_reagents:
if check_not_nan(reagent['lot']):
relevant_reagents.insert(0, str(reagent['lot']))
logger.debug(f"Relevant reagents for {reagent.lot}: {relevant_reagents}")
if str(reagent.lot) not in relevant_reagents:
if check_not_nan(reagent.lot):
relevant_reagents.insert(0, str(reagent.lot))
else:
# TODO: look up the last used reagent of this type in the database
looked_up_reg = lookup_last_used_reagenttype_lot(ctx=ctx, type_name=reagent.type)
logger.debug(f"Because there was no reagent listed for {reagent}, we will insert the last lot used: {looked_up_reg}")
if looked_up_reg != None:
relevant_reagents.remove(str(looked_up_reg.lot))
relevant_reagents.insert(0, str(looked_up_reg.lot))
else:
if len(relevant_reagents) > 1:
logger.debug(f"Found {reagent['lot']} in relevant reagents: {relevant_reagents}. Moving to front of list.")
idx = relevant_reagents.index(str(reagent['lot']))
logger.debug(f"The index we got for {reagent['lot']} in {relevant_reagents} was {idx}")
logger.debug(f"Found {reagent.lot} in relevant reagents: {relevant_reagents}. Moving to front of list.")
idx = relevant_reagents.index(str(reagent.lot))
logger.debug(f"The index we got for {reagent.lot} in {relevant_reagents} was {idx}")
moved_reag = relevant_reagents.pop(idx)
relevant_reagents.insert(0, moved_reag)
else:
logger.debug(f"Found {reagent['lot']} in relevant reagents: {relevant_reagents}. But no need to move due to short list.")
logger.debug(f"Found {reagent.lot} in relevant reagents: {relevant_reagents}. But no need to move due to short list.")
logger.debug(f"New relevant reagents: {relevant_reagents}")
self.setObjectName(f"lot_{reagent['type']}")
self.addItems(relevant_reagents)
self.setObjectName(f"lot_{reagent.type}")
self.addItems(relevant_reagents)
# class ImportReagent(QComboBox):
# def __init__(self, ctx:dict, reagent:dict):
# super().__init__()
# self.setEditable(True)
# # Ensure that all reagenttypes have a name that matches the items in the excel parser
# query_var = reagent['type']
# logger.debug(f"Import Reagent is looking at: {reagent['lot']} for {reagent['type']}")
# if isinstance(reagent['lot'], np.float64):
# logger.debug(f"{reagent['lot']} is a numpy float!")
# try:
# reagent['lot'] = int(reagent['lot'])
# except ValueError:
# pass
# # query for reagents using type name from sheet and kit from sheet
# logger.debug(f"Attempting lookup of reagents by type: {query_var}")
# # below was lookup_reagent_by_type_name_and_kit_name, but I couldn't get it to work.
# relevant_reagents = [item.__str__() for item in lookup_regent_by_type_name(ctx=ctx, type_name=query_var)]
# output_reg = []
# for rel_reagent in relevant_reagents:
# # extract strings from any sets.
# if isinstance(rel_reagent, set):
# for thing in rel_reagent:
# output_reg.append(thing)
# elif isinstance(rel_reagent, str):
# output_reg.append(rel_reagent)
# relevant_reagents = output_reg
# # if reagent in sheet is not found insert it into the front of relevant reagents so it shows
# logger.debug(f"Relevant reagents for {reagent['lot']}: {relevant_reagents}")
# if str(reagent['lot']) not in relevant_reagents:
# if check_not_nan(reagent['lot']):
# relevant_reagents.insert(0, str(reagent['lot']))
# else:
# # TODO: look up the last used reagent of this type in the database
# looked_up_reg = lookup_last_used_reagenttype_lot(ctx=ctx, type_name=reagent['type'])
# logger.debug(f"Because there was no reagent listed for {reagent}, we will insert the last lot used: {looked_up_reg}")
# if looked_up_reg != None:
# relevant_reagents.remove(str(looked_up_reg.lot))
# relevant_reagents.insert(0, str(looked_up_reg.lot))
# else:
# if len(relevant_reagents) > 1:
# logger.debug(f"Found {reagent['lot']} in relevant reagents: {relevant_reagents}. Moving to front of list.")
# idx = relevant_reagents.index(str(reagent['lot']))
# logger.debug(f"The index we got for {reagent['lot']} in {relevant_reagents} was {idx}")
# moved_reag = relevant_reagents.pop(idx)
# relevant_reagents.insert(0, moved_reag)
# else:
# logger.debug(f"Found {reagent['lot']} in relevant reagents: {relevant_reagents}. But no need to move due to short list.")
# logger.debug(f"New relevant reagents: {relevant_reagents}")
# self.setObjectName(f"lot_{reagent['type']}")
# self.addItems(relevant_reagents)

49
src/submissions/frontend/custom_widgets/pop_ups.py
View File

@@ -3,22 +3,15 @@ Contains dialogs for notification and prompting.
'''
from PyQt6.QtWidgets import (
QLabel, QVBoxLayout, QDialog,
QDialogButtonBox, QMessageBox
QDialogButtonBox, QMessageBox, QComboBox
)
from jinja2 import Environment, FileSystemLoader
import sys
from pathlib import Path
from configure import jinja_template_loading
import logging
from backend.db.functions import lookup_kittype_by_use
logger = logging.getLogger(f"submissions.{__name__}")
# determine if pyinstaller launcher is being used
if getattr(sys, 'frozen', False):
loader_path = Path(sys._MEIPASS).joinpath("files", "templates")
else:
loader_path = Path(__file__).parents[2].joinpath('templates').absolute().__str__()
loader = FileSystemLoader(loader_path)
env = Environment(loader=loader)
env = jinja_template_loading()
class QuestionAsker(QDialog):
@@ -52,3 +45,37 @@ class AlertPop(QMessageBox):
self.setInformativeText(message)
self.setWindowTitle(status.title())
class KitSelector(QDialog):
"""
dialog to ask yes/no questions
"""
def __init__(self, ctx:dict, title:str, message:str) -> QDialog:
super().__init__()
self.setWindowTitle(title)
self.widget = QComboBox()
kits = [item.__str__() for item in lookup_kittype_by_use(ctx=ctx)]
self.widget.addItems(kits)
self.widget.setEditable(False)
# set yes/no buttons
QBtn = QDialogButtonBox.StandardButton.Ok | QDialogButtonBox.StandardButton.Cancel
self.buttonBox = QDialogButtonBox(QBtn)
self.buttonBox.accepted.connect(self.accept)
self.buttonBox.rejected.connect(self.reject)
self.layout = QVBoxLayout()
# Text for the yes/no question
message = QLabel(message)
self.layout.addWidget(message)
self.layout.addWidget(self.widget)
self.layout.addWidget(self.buttonBox)
self.setLayout(self.layout)
def getValues(self):
return self.widget.currentText()
# @staticmethod
# def launch(parent):
# dlg = KitSelector(parent)
# r = dlg.exec_()
# if r:
# return dlg.getValues()
# return None

18
src/submissions/frontend/custom_widgets/sub_details.py
View File

@@ -4,22 +4,19 @@ Contains widgets specific to the submission summary and submission details.
import base64
from datetime import datetime
from io import BytesIO
import math
from PyQt6 import QtPrintSupport
from PyQt6.QtWidgets import (
QVBoxLayout, QDialog, QTableView,
QTextEdit, QPushButton, QScrollArea,
QMessageBox, QFileDialog, QMenu, QLabel,
QDialogButtonBox, QToolBar, QMainWindow
QDialogButtonBox, QToolBar
)
from PyQt6.QtCore import Qt, QAbstractTableModel, QSortFilterProxyModel, QItemSelectionModel
from PyQt6.QtCore import Qt, QAbstractTableModel, QSortFilterProxyModel
from PyQt6.QtGui import QFontMetrics, QAction, QCursor, QPixmap, QPainter
from backend.db import submissions_to_df, lookup_submission_by_id, delete_submission_by_id, lookup_submission_by_rsl_num, hitpick_plate
# from backend.misc import hitpick_plate
from backend.excel import make_hitpicks
from jinja2 import Environment, FileSystemLoader
from configure import jinja_template_loading
from xhtml2pdf import pisa
import sys
from pathlib import Path
import logging
from .pop_ups import QuestionAsker, AlertPop
@@ -28,12 +25,7 @@ from getpass import getuser
logger = logging.getLogger(f"submissions.{__name__}")
if getattr(sys, 'frozen', False):
loader_path = Path(sys._MEIPASS).joinpath("files", "templates")
else:
loader_path = Path(__file__).parents[2].joinpath('templates').absolute().__str__()
loader = FileSystemLoader(loader_path)
env = Environment(loader=loader)
env = jinja_template_loading()
class pandasModel(QAbstractTableModel):
"""
@@ -407,7 +399,7 @@ class BarcodeWindow(QDialog):
class SubmissionComment(QDialog):
"""
a window showing text details of submission
a window for adding comment text to a submission
"""
def __init__(self, ctx:dict, rsl:str) -> None:

414
src/submissions/frontend/main_window_functions.py
View File

@@ -29,24 +29,27 @@ from backend.db.functions import (
)
from backend.excel.parser import SheetParser, PCRParser
from backend.excel.reports import make_report_html, make_report_xlsx, convert_data_list_to_df
from backend.pydant import PydReagent
from tools import check_not_nan, check_kit_integrity
from .custom_widgets.pop_ups import AlertPop, QuestionAsker
from .custom_widgets import ReportDatePicker
from .custom_widgets.misc import ImportReagent
from .visualizations.control_charts import create_charts, construct_html
from typing import List
from openpyxl import load_workbook
logger = logging.getLogger(f"submissions.{__name__}")
def import_submission_function(obj:QMainWindow) -> Tuple[QMainWindow, dict|None]:
"""
_summary_
Import a new submission to the app window
Args:
obj (QMainWindow): _description_
obj (QMainWindow): original app window
Returns:
Tuple[QMainWindow, dict|None]: _description_
Tuple[QMainWindow, dict|None]: Collection of new main app window and result dict
"""
logger.debug(f"\n\nStarting Import...\n\n")
result = None
@@ -54,8 +57,10 @@ def import_submission_function(obj:QMainWindow) -> Tuple[QMainWindow, dict|None]
# initialize samples
obj.samples = []
obj.reagents = {}
obj.missing_reagents = []
# set file dialog
fname = select_open_file(obj, extension="xlsx")
fname = select_open_file(obj, file_extension="xlsx")
logger.debug(f"Attempting to parse file: {fname}")
if not fname.exists():
result = dict(message=f"File {fname.__str__()} not found.", status="critical")
@@ -65,116 +70,37 @@ def import_submission_function(obj:QMainWindow) -> Tuple[QMainWindow, dict|None]
prsr = SheetParser(ctx=obj.ctx, filepath=fname)
except PermissionError:
logger.error(f"Couldn't get permission to access file: {fname}")
return
return obj, result
# obj.column_count = prsr.column_count
try:
logger.debug(f"Submission dictionary: {prsr.sub}")
pyd = prsr.to_pydantic()
logger.debug(f"Pydantic result: \n\n{pyd}\n\n")
# with open("pickled.pkl", "wb") as f:
# pickle.dump(pyd, f)
logger.debug(f"Pydantic result: \n\n{pprint.pformat(pyd)}\n\n")
except Exception as e:
return obj, dict(message= f"Problem creating pydantic model:\n\n{e}", status="critical")
# moved to pydantic model
# if prsr.sub['rsl_plate_num'] == None:
# prsr.sub['rsl_plate_num'] = RSLNamer(fname.__str__()).parsed_name
# logger.debug(f"prsr.sub = {prsr.sub}")
for sample in pyd.samples:
if hasattr(sample, "elution_well"):
logger.debug(f"Sample from import: {sample.elution_well}")
# obj.current_submission_type = prsr.sub['submission_type']
try:
obj.xl = prsr.filepath
except Exception as e:
logger.error(f"Unable to make obj xl.")
# for sample in pyd.samples:
# if hasattr(sample, "elution_well"):
# logger.debug(f"Sample from import: {sample.elution_well}")
# I don't remember why this is here.
missing_info = [k for k,v in pyd if v == None]
obj.current_submission_type = pyd.submission_type
# destroy any widgets from previous imports
for item in obj.table_widget.formlayout.parentWidget().findChildren(QWidget):
item.setParent(None)
# # regex to parser out different variable types for decision making
# variable_parser = re.compile(r"""
# (?P<extraction_kit>^extraction_kit$) |
# (?P<submitted_date>^submitted_date$) |
# (?P<submitting_lab>)^submitting_lab$ |
# (?P<samples>)^samples$ |
# (?P<reagent>^lot_.*$) |
# (?P<csv>^csv$)
# """, re.VERBOSE)
# for item in prsr.sub:
# logger.debug(f"Item: {item}")
# # attempt to match variable name to regex group
# try:
# mo = variable_parser.fullmatch(item).lastgroup
# except AttributeError:
# mo = "other"
# logger.debug(f"Mo: {mo}")
# match mo:
# case 'submitting_lab':
# # create label
# obj.table_widget.formlayout.addWidget(QLabel(item.replace("_", " ").title()))
# logger.debug(f"{item}: {prsr.sub[item]}")
# # create combobox to hold looked up submitting labs
# add_widget = QComboBox()
# labs = [item.__str__() for item in lookup_all_orgs(ctx=obj.ctx)]
# # try to set closest match to top of list
# try:
# labs = difflib.get_close_matches(prsr.sub[item], labs, len(labs), 0)
# except (TypeError, ValueError):
# pass
# # set combobox values to lookedup values
# add_widget.addItems(labs)
# case 'extraction_kit':
# # create label
# obj.table_widget.formlayout.addWidget(QLabel(item.replace("_", " ").title()))
# # if extraction kit not available, all other values fail
# if not check_not_nan(prsr.sub[item]):
# msg = AlertPop(message="Make sure to check your extraction kit in the excel sheet!", status="warning")
# msg.exec()
# # create combobox to hold looked up kits
# add_widget = QComboBox()
# # lookup existing kits by 'submission_type' decided on by sheetparser
# uses = [item.__str__() for item in lookup_kittype_by_use(ctx=obj.ctx, used_by=prsr.sub['submission_type'])]
# if check_not_nan(prsr.sub[item]):
# logger.debug(f"The extraction kit in parser was: {prsr.sub[item]}")
# uses.insert(0, uses.pop(uses.index(prsr.sub[item])))
# obj.ext_kit = prsr.sub[item]
# else:
# logger.error(f"Couldn't find {prsr.sub['extraction_kit']}")
# obj.ext_kit = uses[0]
# add_widget.addItems(uses)
# case 'submitted_date':
# # create label
# obj.table_widget.formlayout.addWidget(QLabel(item.replace("_", " ").title()))
# # uses base calendar
# add_widget = QDateEdit(calendarPopup=True)
# # sets submitted date based on date found in excel sheet
# try:
# add_widget.setDate(prsr.sub[item])
# # if not found, use today
# except:
# add_widget.setDate(date.today())
# case 'reagent':
# # create label
# reg_label = QLabel(item.replace("_", " ").title())
# reg_label.setObjectName(f"lot_{item}_label")
# obj.table_widget.formlayout.addWidget(reg_label)
# # create reagent choice widget
# add_widget = ImportReagent(ctx=obj.ctx, item=item, prsr=prsr)
# obj.reagents[item] = prsr.sub[item]
# case 'samples':
# # hold samples in 'obj' until form submitted
# logger.debug(f"{item}: {prsr.sub[item]}")
# obj.samples = prsr.sub[item]
# add_widget = None
# case 'csv':
# obj.csv = prsr.sub[item]
# case _:
# # anything else gets added in as a line edit
# obj.table_widget.formlayout.addWidget(QLabel(item.replace("_", " ").title()))
# add_widget = QLineEdit()
# logger.debug(f"Setting widget text to {str(prsr.sub[item]).replace('_', ' ')}")
# add_widget.setText(str(prsr.sub[item]).replace("_", " "))
# Get list of fields from pydantic model.
fields = list(pyd.model_fields.keys())
fields.remove('filepath')
logger.debug(f"pydantic fields: {fields}")
for field in fields:
value = getattr(pyd, field)
if not check_not_nan(value):
continue
logger.debug(f"Checking: {field}: {value}")
# No longer necessary with addition of pydantic validations
# if not check_not_nan(value):
# continue
match field:
case 'submitting_lab':
# create label
@@ -194,17 +120,19 @@ def import_submission_function(obj:QMainWindow) -> Tuple[QMainWindow, dict|None]
# create label
label = QLabel(field.replace("_", " ").title())
# if extraction kit not available, all other values fail
if not check_not_nan(value):
if not check_not_nan(value['value']):
msg = AlertPop(message="Make sure to check your extraction kit in the excel sheet!", status="warning")
msg.exec()
if not value['parsed']:
missing_info.append(field)
# create combobox to hold looked up kits
add_widget = QComboBox()
# lookup existing kits by 'submission_type' decided on by sheetparser
uses = [item.__str__() for item in lookup_kittype_by_use(ctx=obj.ctx, used_by=pyd.submission_type)]
if check_not_nan(value):
logger.debug(f"The extraction kit in parser was: {value}")
uses.insert(0, uses.pop(uses.index(value)))
obj.ext_kit = value
logger.debug(f"The extraction kit in parser was: {value['value']}")
uses.insert(0, uses.pop(uses.index(value['value'])))
obj.ext_kit = value['value']
else:
logger.error(f"Couldn't find {prsr.sub['extraction_kit']}")
obj.ext_kit = uses[0]
@@ -225,23 +153,33 @@ def import_submission_function(obj:QMainWindow) -> Tuple[QMainWindow, dict|None]
logger.debug(f"{field}:\n\t{value}")
obj.samples = value
continue
case 'csv':
obj.csv = value
case "ctx":
continue
case 'reagents':
for reagent in value:
# create label
reg_label = QLabel(reagent['type'].replace("_", " ").title())
reg_label.setObjectName(f"lot_{reagent['type']}_label")
# obj.table_widget.formlayout.addWidget(reg_label)
# reg_label = QLabel(reagent['type'].replace("_", " ").title())
try:
reg_label = QLabel(reagent.type.replace("_", " ").title())
except AttributeError:
continue
# reg_label.setObjectName(f"lot_{reagent['type']}_label")
reg_label.setObjectName(f"lot_{reagent.type}_label")
# create reagent choice widget
add_widget = ImportReagent(ctx=obj.ctx, reagent=reagent)
add_widget.setObjectName(f"lot_{reagent['type']}")
add_widget.setObjectName(f"lot_{reagent.type}")
logger.debug(f"Widget name set to: {add_widget.objectName()}")
obj.table_widget.formlayout.addWidget(reg_label)
obj.table_widget.formlayout.addWidget(add_widget)
obj.reagents[reagent['type']] = reagent['lot']
obj.reagents[reagent.type] = reagent
continue
case "rsl_plate_num":
if not value['parsed']:
missing_info.append(field)
label = QLabel(field.replace("_", " ").title())
add_widget = QLineEdit()
logger.debug(f"Setting widget text to {str(value['value']).replace('_', ' ')}")
add_widget.setText(str(value['value']).replace("_", " "))
case _:
# anything else gets added in as a line edit
label = QLabel(field.replace("_", " ").title())
@@ -258,13 +196,27 @@ def import_submission_function(obj:QMainWindow) -> Tuple[QMainWindow, dict|None]
# compare obj.reagents with expected reagents in kit
if hasattr(obj, 'ext_kit'):
obj.kit_integrity_completion()
obj.missing_reagents = obj.missing_reagents + missing_info
logger.debug(f"Imported reagents: {obj.reagents}")
if prsr.sample_result != None:
msg = AlertPop(message=prsr.sample_result, status="WARNING")
msg.exec()
logger.debug(f"Pydantic extra fields: {pyd.model_extra}")
if "csv" in pyd.model_extra:
obj.csv = pyd.model_extra['csv']
logger.debug(f"All attributes of obj:\n{pprint.pformat(obj.__dict__)}")
return obj, result
def kit_reload_function(obj:QMainWindow) -> QMainWindow:
def kit_reload_function(obj:QMainWindow) -> Tuple[QMainWindow, dict]:
"""
Reload the fields in the form
Args:
obj (QMainWindow): original app window
Returns:
Tuple[QMainWindow, dict]: Collection of new main app window and result dict
"""
result = None
for item in obj.table_widget.formlayout.parentWidget().findChildren(QWidget):
if isinstance(item, QLabel):
@@ -277,7 +229,16 @@ def kit_reload_function(obj:QMainWindow) -> QMainWindow:
obj.kit_integrity_completion_function()
return obj, result
def kit_integrity_completion_function(obj:QMainWindow) -> QMainWindow:
def kit_integrity_completion_function(obj:QMainWindow) -> Tuple[QMainWindow, dict]:
"""
Compare kit contents to parsed contents
Args:
obj (QMainWindow): The original app window
Returns:
Tuple[QMainWindow, dict]: Collection of new main app window and result dict
"""
result = None
logger.debug(inspect.currentframe().f_back.f_code.co_name)
# find the widget that contains lit info
@@ -296,9 +257,11 @@ def kit_integrity_completion_function(obj:QMainWindow) -> QMainWindow:
# if kit integrity comes back with an error, make widgets with missing reagents using default info
if kit_integrity != None:
result = dict(message=kit_integrity['message'], status="Warning")
obj.missing_reagents = kit_integrity['missing']
for item in kit_integrity['missing']:
obj.table_widget.formlayout.addWidget(QLabel(f"Lot {item.replace('_', ' ').title()}"))
add_widget = ImportReagent(ctx=obj.ctx, reagent=dict(type=item, lot=None, exp=None))#item=item)
reagent = dict(type=item, lot=None, exp=None)
add_widget = ImportReagent(ctx=obj.ctx, reagent=PydReagent(**reagent))#item=item)
obj.table_widget.formlayout.addWidget(add_widget)
submit_btn = QPushButton("Submit")
submit_btn.setObjectName("lot_submit_btn")
@@ -306,38 +269,50 @@ def kit_integrity_completion_function(obj:QMainWindow) -> QMainWindow:
submit_btn.clicked.connect(obj.submit_new_sample)
return obj, result
def submit_new_sample_function(obj:QMainWindow) -> QMainWindow:
def submit_new_sample_function(obj:QMainWindow) -> Tuple[QMainWindow, dict]:
"""
Parse forms and add sample to the database.
Args:
obj (QMainWindow): original app window
Returns:
Tuple[QMainWindow, dict]: Collection of new main app window and result dict
"""
logger.debug(f"\n\nBeginning Submission\n\n")
result = None
# extract info from the form widgets
info = extract_form_info(obj.table_widget.tab1)
# seperate out reagents
reagents = {k:v for k,v in info.items() if k.startswith("lot_")}
reagents = {k.replace("lot_", ""):v for k,v in info.items() if k.startswith("lot_")}
info = {k:v for k,v in info.items() if not k.startswith("lot_")}
logger.debug(f"Info: {info}")
logger.debug(f"Reagents: {reagents}")
parsed_reagents = []
# compare reagents in form to reagent database
for reagent in reagents:
# TODO: have this lookup by type and lot
# Lookup any existing reagent of this type with this lot number
wanted_reagent = lookup_reagent(ctx=obj.ctx, reagent_lot=reagents[reagent], type_name=reagent)
logger.debug(f"Looked up reagent: {wanted_reagent}")
# logger.debug(f"\n\nLooking for {reagent} in {obj.reagents}\n\n")
# if reagent not found offer to add to database
if wanted_reagent == None:
r_lot = reagents[reagent]
dlg = QuestionAsker(title=f"Add {r_lot}?", message=f"Couldn't find reagent type {reagent.replace('_', ' ').title().strip('Lot')}: {r_lot} in the database.\n\nWould you like to add it?")
if dlg.exec():
logger.debug(f"checking reagent: {reagent} in obj.reagents. Result: {obj.reagents[reagent]}")
expiry_date = obj.reagents[reagent]['exp']
expiry_date = obj.reagents[reagent].exp
wanted_reagent = obj.add_reagent(reagent_lot=r_lot, reagent_type=reagent.replace("lot_", ""), expiry=expiry_date)
else:
# In this case we will have an empty reagent and the submission will fail kit integrity check
logger.debug("Will not add reagent.")
if wanted_reagent != None:
parsed_reagents.append(wanted_reagent)
wanted_reagent.type.last_used = reagents[reagent]
# move samples into preliminary submission dict
info['samples'] = obj.samples
info['uploaded_by'] = getuser()
# info['columns'] = obj.column_count
# construct submission object
logger.debug(f"Here is the info_dict: {pprint.pformat(info)}")
base_submission, result = construct_submission_info(ctx=obj.ctx, info_dict=info)
@@ -359,6 +334,7 @@ def submit_new_sample_function(obj:QMainWindow) -> QMainWindow:
# add reagents to submission object
for reagent in parsed_reagents:
base_submission.reagents.append(reagent)
logger.debug(f"Parsed reagents: {pprint.pformat(parsed_reagents)}")
logger.debug("Checking kit integrity...")
kit_integrity = check_kit_integrity(base_submission)
if kit_integrity != None:
@@ -371,7 +347,15 @@ def submit_new_sample_function(obj:QMainWindow) -> QMainWindow:
# reset form
for item in obj.table_widget.formlayout.parentWidget().findChildren(QWidget):
item.setParent(None)
logger.debug(f"All attributes of obj: {pprint.pprint(obj.__dict__)}")
logger.debug(f"All attributes of obj: {pprint.pformat(obj.__dict__)}")
if len(obj.missing_reagents) > 0:
logger.debug(f"We have blank reagents in the excel sheet.\n\tLet's try to fill them in.")
extraction_kit = lookup_kittype_by_name(obj.ctx, name=obj.ext_kit)
logger.debug(f"We have the extraction kit: {extraction_kit.name}")
logger.debug(f"Extraction kit map:\n\n{extraction_kit.used_for[obj.current_submission_type]}")
excel_map = extraction_kit.used_for[obj.current_submission_type]
input_reagents = [item.to_reagent_dict() for item in parsed_reagents]
autofill_excel(obj=obj, xl_map=excel_map, reagents=input_reagents, missing_reagents=obj.missing_reagents, info=info)
if hasattr(obj, 'csv'):
dlg = QuestionAsker("Export CSV?", "Would you like to export the csv file?")
if dlg.exec():
@@ -384,13 +368,18 @@ def submit_new_sample_function(obj:QMainWindow) -> QMainWindow:
delattr(obj, "csv")
except AttributeError:
pass
# if obj.current_submission_type == "Bacterial_Culture":
# hitpick = hitpick_plate(base_submission)
# image = make_plate_map(hitpick)
# image.show()
return obj, result
def generate_report_function(obj:QMainWindow) -> QMainWindow:
def generate_report_function(obj:QMainWindow) -> Tuple[QMainWindow, dict]:
"""
Generate a summary of activities for a time period
Args:
obj (QMainWindow): original app window
Returns:
Tuple[QMainWindow, dict]: Collection of new main app window and result dict
"""
result = None
# ask for date ranges
dlg = ReportDatePicker()
@@ -430,7 +419,16 @@ def generate_report_function(obj:QMainWindow) -> QMainWindow:
writer.close()
return obj, result
def add_kit_function(obj:QMainWindow) -> QMainWindow:
def add_kit_function(obj:QMainWindow) -> Tuple[QMainWindow, dict]:
"""
Add a new kit to the database.
Args:
obj (QMainWindow): original app window
Returns:
Tuple[QMainWindow, dict]: Collection of new main app window and result dict
"""
result = None
# setup file dialog to find yaml flie
fname = select_open_file(obj, extension="yml")
@@ -449,7 +447,16 @@ def add_kit_function(obj:QMainWindow) -> QMainWindow:
result = create_kit_from_yaml(ctx=obj.ctx, exp=exp)
return obj, result
def add_org_function(obj:QMainWindow) -> QMainWindow:
def add_org_function(obj:QMainWindow) -> Tuple[QMainWindow, dict]:
"""
Add a new organization to the database.
Args:
obj (QMainWindow): original app window
Returns:
Tuple[QMainWindow, dict]: Collection of new main app window and result dict
"""
result = None
# setup file dialog to find yaml flie
fname = select_open_file(obj, extension="yml")
@@ -468,7 +475,16 @@ def add_org_function(obj:QMainWindow) -> QMainWindow:
result = create_org_from_yaml(ctx=obj.ctx, org=org)
return obj, result
def controls_getter_function(obj:QMainWindow) -> QMainWindow:
def controls_getter_function(obj:QMainWindow) -> Tuple[QMainWindow, dict]:
"""
Get controls based on start/end dates
Args:
obj (QMainWindow): original app window
Returns:
Tuple[QMainWindow, dict]: Collection of new main app window and result dict
"""
result = None
# subtype defaults to disabled
try:
@@ -480,11 +496,12 @@ def controls_getter_function(obj:QMainWindow) -> QMainWindow:
logger.warning("Start date after end date is not allowed!")
threemonthsago = obj.table_widget.datepicker.end_date.date().addDays(-60)
# block signal that will rerun controls getter and set start date
# Without triggering this function again
with QSignalBlocker(obj.table_widget.datepicker.start_date) as blocker:
obj.table_widget.datepicker.start_date.setDate(threemonthsago)
obj._controls_getter()
return obj, result
# convert to python useable date object
# convert to python useable date objects
obj.start_date = obj.table_widget.datepicker.start_date.date().toPyDate()
obj.end_date = obj.table_widget.datepicker.end_date.date().toPyDate()
obj.con_type = obj.table_widget.control_typer.currentText()
@@ -504,15 +521,15 @@ def controls_getter_function(obj:QMainWindow) -> QMainWindow:
obj._chart_maker()
return obj, result
def chart_maker_function(obj:QMainWindow) -> QMainWindow:
def chart_maker_function(obj:QMainWindow) -> Tuple[QMainWindow, dict]:
"""
create html chart for controls reporting
Create html chart for controls reporting
Args:
obj (QMainWindow): original MainWindow
obj (QMainWindow): original app window
Returns:
QMainWindow: MainWindow with control display updates
Tuple[QMainWindow, dict]: Collection of new main app window and result dict
"""
result = None
logger.debug(f"Control getter context: \n\tControl type: {obj.con_type}\n\tMode: {obj.mode}\n\tStart Date: {obj.start_date}\n\tEnd Date: {obj.end_date}")
@@ -552,7 +569,16 @@ def chart_maker_function(obj:QMainWindow) -> QMainWindow:
logger.debug("Figure updated... I hope.")
return obj, result
def link_controls_function(obj:QMainWindow) -> QMainWindow:
def link_controls_function(obj:QMainWindow) -> Tuple[QMainWindow, dict]:
"""
Link scraped controls to imported submissions.
Args:
obj (QMainWindow): original app window
Returns:
Tuple[QMainWindow, dict]: Collection of new main app window and result dict
"""
result = None
all_bcs = lookup_all_submissions_by_type(obj.ctx, "Bacterial Culture")
logger.debug(all_bcs)
@@ -601,12 +627,20 @@ def link_controls_function(obj:QMainWindow) -> QMainWindow:
# msg.exec()
return obj, result
def link_extractions_function(obj:QMainWindow) -> QMainWindow:
def link_extractions_function(obj:QMainWindow) -> Tuple[QMainWindow, dict]:
"""
Link extractions from runlogs to imported submissions
Args:
obj (QMainWindow): original app window
Returns:
Tuple[QMainWindow, dict]: Collection of new main app window and result dict
"""
result = None
# home_dir = str(Path(obj.ctx["directory_path"]))
# fname = Path(QFileDialog.getOpenFileName(obj, 'Open file', home_dir, filter = "csv(*.csv)")[0])
fname = select_open_file(obj, extension="csv")
fname = select_open_file(obj, file_extension="csv")
with open(fname.__str__(), 'r') as f:
# split csv on commas
runs = [col.strip().split(",") for col in f.readlines()]
count = 0
for run in runs:
@@ -618,9 +652,12 @@ def link_extractions_function(obj:QMainWindow) -> QMainWindow:
experiment_name=run[4].strip(),
end_time=run[5].strip()
)
# elution columns are item 6 in the comma split list to the end
for ii in range(6, len(run)):
new_run[f"column{str(ii-5)}_vol"] = run[ii]
# Lookup imported submissions
sub = lookup_submission_by_rsl_num(ctx=obj.ctx, rsl_num=new_run['rsl_plate_num'])
# If no such submission exists, move onto the next run
try:
logger.debug(f"Found submission: {sub.rsl_plate_num}")
count += 1
@@ -630,12 +667,14 @@ def link_extractions_function(obj:QMainWindow) -> QMainWindow:
existing = json.loads(sub.extraction_info)
else:
existing = None
# Check if the new info already exists in the imported submission
try:
if json.dumps(new_run) in sub.extraction_info:
logger.debug(f"Looks like we already have that info.")
continue
except TypeError:
pass
# Update or create the extraction info
if existing != None:
try:
logger.debug(f"Updating {type(existing)}: {existing} with {type(new_run)}: {new_run}")
@@ -653,12 +692,20 @@ def link_extractions_function(obj:QMainWindow) -> QMainWindow:
result = dict(message=f"We added {count} logs to the database.", status='information')
return obj, result
def link_pcr_function(obj:QMainWindow) -> QMainWindow:
def link_pcr_function(obj:QMainWindow) -> Tuple[QMainWindow, dict]:
"""
Link PCR data from run logs to an imported submission
Args:
obj (QMainWindow): original app window
Returns:
Tuple[QMainWindow, dict]: Collection of new main app window and result dict
"""
result = None
# home_dir = str(Path(obj.ctx["directory_path"]))
# fname = Path(QFileDialog.getOpenFileName(obj, 'Open file', home_dir, filter = "csv(*.csv)")[0])
fname = select_open_file(obj, extension="csv")
fname = select_open_file(obj, file_extension="csv")
with open(fname.__str__(), 'r') as f:
# split csv rows on comma
runs = [col.strip().split(",") for col in f.readlines()]
count = 0
for run in runs:
@@ -670,17 +717,19 @@ def link_pcr_function(obj:QMainWindow) -> QMainWindow:
experiment_name=run[4].strip(),
end_time=run[5].strip()
)
# for ii in range(6, len(run)):
# obj[f"column{str(ii-5)}_vol"] = run[ii]
# lookup imported submission
sub = lookup_submission_by_rsl_num(ctx=obj.ctx, rsl_num=new_run['rsl_plate_num'])
# if imported submission doesn't exist move on to next run
try:
logger.debug(f"Found submission: {sub.rsl_plate_num}")
except AttributeError:
continue
# check if pcr_info already exists
if hasattr(sub, 'pcr_info') and sub.pcr_info != None:
existing = json.loads(sub.pcr_info)
else:
existing = None
# check if this entry already exists in imported submission
try:
if json.dumps(new_run) in sub.pcr_info:
logger.debug(f"Looks like we already have that info.")
@@ -706,17 +755,25 @@ def link_pcr_function(obj:QMainWindow) -> QMainWindow:
result = dict(message=f"We added {count} logs to the database.", status='information')
return obj, result
def import_pcr_results_function(obj:QMainWindow) -> QMainWindow:
def import_pcr_results_function(obj:QMainWindow) -> Tuple[QMainWindow, dict]:
"""
Import Quant-studio PCR data to an imported submission
Args:
obj (QMainWindow): original app window
Returns:
Tuple[QMainWindow, dict]: Collection of new main app window and result dict
"""
result = None
# home_dir = str(Path(obj.ctx["directory_path"]))
# fname = Path(QFileDialog.getOpenFileName(obj, 'Open file', home_dir, filter = "xlsx(*.xlsx)")[0])
fname = select_open_file(obj, extension="xlsx")
fname = select_open_file(obj, file_extension="xlsx")
parser = PCRParser(ctx=obj.ctx, filepath=fname)
logger.debug(f"Attempting lookup for {parser.plate_num}")
sub = lookup_submission_by_rsl_num(ctx=obj.ctx, rsl_num=parser.plate_num)
try:
logger.debug(f"Found submission: {sub.rsl_plate_num}")
except AttributeError:
# If no plate is found, may be because this is a repeat. Lop off the '-1' or '-2' and repeat
logger.error(f"Submission of number {parser.plate_num} not found. Attempting rescue of plate repeat.")
parser.plate_num = "-".join(parser.plate_num.split("-")[:-1])
sub = lookup_submission_by_rsl_num(ctx=obj.ctx, rsl_num=parser.plate_num)
@@ -725,14 +782,13 @@ def import_pcr_results_function(obj:QMainWindow) -> QMainWindow:
except AttributeError:
logger.error(f"Rescue of {parser.plate_num} failed.")
return obj, dict(message="Couldn't find a submission with that RSL number.", status="warning")
# jout = json.dumps(parser.pcr)
count = 0
# Check if PCR info already exists
if hasattr(sub, 'pcr_info') and sub.pcr_info != None:
existing = json.loads(sub.pcr_info)
else:
# jout = None
existing = None
if existing != None:
# update pcr_info
try:
logger.debug(f"Updating {type(existing)}: {existing} with {type(parser.pcr)}: {parser.pcr}")
if json.dumps(parser.pcr) not in sub.pcr_info:
@@ -757,5 +813,59 @@ def import_pcr_results_function(obj:QMainWindow) -> QMainWindow:
update_ww_sample(ctx=obj.ctx, sample_obj=sample)
result = dict(message=f"We added PCR info to {sub.rsl_plate_num}.", status='information')
return obj, result
# dlg.exec()
def autofill_excel(obj:QMainWindow, xl_map:dict, reagents:List[dict], missing_reagents:List[str], info:dict):
"""
Automatically fills in excel cells with reagent info.
Args:
obj (QMainWindow): Original main app window
xl_map (dict): Map of where each reagent goes in the excel workbook.
reagents (List[dict]): All reagents used in the kit.
missing_reagents (List[str]): Reagents that are required for the kit that were not present.
"""
# logger.debug(reagents)
logger.debug(f"Here is the info dict coming in:\n{pprint.pformat(info)}")
logger.debug(f"Here are the missing reagents:\n{missing_reagents}")
relevant_map = {k:v for k,v in xl_map.items() if k in missing_reagents}
# logger.debug(relevant_map)
relevant_reagents = [item for item in reagents if item['type'] in missing_reagents]
relevant_info = {k:v for k,v in info.items() if k in missing_reagents}
logger.debug(f"Here is the relevant info: {pprint.pformat(relevant_info)}")
# logger.debug(f"Relevant reagents:\n{relevant_reagents}")
# construct new objects to put into excel sheets:
new_reagents = []
for reagent in relevant_reagents:
new_reagent = {}
new_reagent['type'] = reagent['type']
new_reagent['lot'] = relevant_map[new_reagent['type']]['lot']
new_reagent['lot']['value'] = reagent['lot']
new_reagent['expiry'] = relevant_map[new_reagent['type']]['expiry']
new_reagent['expiry']['value'] = reagent['expiry']
new_reagent['sheet'] = relevant_map[new_reagent['type']]['sheet']
new_reagents.append(new_reagent)
new_info = []
for item in relevant_info:
new_item = {}
new_item['type'] = item
new_item['location'] = relevant_map[item]
new_item['value'] = relevant_info[item]
new_info.append(new_item)
logger.debug(f"New reagents: {new_reagents}")
workbook = load_workbook(obj.xl)
sheets = workbook.sheetnames
logger.debug(workbook.sheetnames)
for sheet in sheets:
worksheet=workbook[sheet]
sheet_reagents = [item for item in new_reagents if item['sheet'] == sheet]
for reagent in sheet_reagents:
logger.debug(f"Attempting: {reagent['type']}:")
worksheet.cell(row=reagent['lot']['row'], column=reagent['lot']['column'], value=reagent['lot']['value'])
worksheet.cell(row=reagent['expiry']['row'], column=reagent['expiry']['column'], value=reagent['expiry']['value'])
sheet_info = [item for item in new_info if item['location']['sheet'] == sheet]
for item in sheet_info:
logger.debug(f"Attempting: {item['type']}")
worksheet.cell(row=item['location']['row'], column=item['location']['column'], value=item['value'])
fname = select_save_file(obj=obj, default_name=Path(obj.xl).stem, extension="xlsx")
workbook.save(filename=fname.__str__())

9
src/submissions/frontend/visualizations/control_charts.py
View File

@@ -283,6 +283,15 @@ def divide_chunks(input_list:list, chunk_count:int):
def construct_html(figure:Figure) -> str:
"""
Creates final html code from plotly
Args:
figure (Figure): input figure
Returns:
str: html string
"""
html = '<html><body>'
if figure != None:
html += plotly.offline.plot(figure, output_type='div', include_plotlyjs='cdn')#, image = 'png', auto_open=True, image_filename='plot_image')

49
src/submissions/tools/__init__.py
View File

@@ -1,12 +1,14 @@
'''
Contains miscellaenous functions used by both frontend and backend.
'''
from pathlib import Path
import re
import sys
import numpy as np
import logging
import getpass
from backend.db.models import BasicSubmission, KitType
import pandas as pd
from typing import Tuple
logger = logging.getLogger(f"submissions.{__name__}")
@@ -22,19 +24,43 @@ def check_not_nan(cell_contents) -> bool:
bool: True if cell has value, else, false.
"""
# check for nan as a string first
try:
cell_contents = cell_contents.lower()
except AttributeError:
pass
if cell_contents == 'nan':
cell_contents = np.nan
if cell_contents == None:
cell_contents = np.nan
try:
if pd.isnull(cell_contents):
cell_contents = np.nan
except ValueError:
pass
try:
return not np.isnan(cell_contents)
except TypeError:
return True
except Exception as e:
logger.debug(f"Check encounteded unknown error: {type(e).__name__} - {e}")
logger.debug(f"Check encountered unknown error: {type(e).__name__} - {e}")
return False
def convert_nans_to_nones(input_str) -> str|None:
"""
Get rid of various "nan", "NAN", "NaN", etc/
Args:
input_str (str): input string
Returns:
str: _description_
"""
if not check_not_nan(input_str):
return None
return input_str
def check_is_power_user(ctx:dict) -> bool:
"""
Check to ensure current user is in power users list.
@@ -83,7 +109,7 @@ def check_kit_integrity(sub:BasicSubmission|KitType, reagenttypes:list|None=None
# What type is sub?
match sub:
case BasicSubmission():
# very hacky method to ensure interchangeable plates are not
# Get all required reagent types for this kit.
ext_kit_rtypes = [reagenttype.name for reagenttype in sub.extraction_kit.reagent_types if reagenttype.required == 1]
# Overwrite function parameter reagenttypes
try:
@@ -179,7 +205,7 @@ class RSLNamer(object):
self.parsed_name = self.parsed_name.replace("_", "-")
def retrieve_rsl_number(self, in_str:str) -> Tuple[str, str]:
def retrieve_rsl_number(self, in_str:str|Path) -> Tuple[str, str]:
"""
Uses regex to retrieve the plate number and submission type from an input string
@@ -189,6 +215,8 @@ class RSLNamer(object):
Returns:
Tuple[str, str]: tuple of (output rsl number, submission_type)
"""
if isinstance(in_str, Path):
in_str = in_str.stem
logger.debug(f"Attempting split of {in_str}")
try:
in_str = in_str.split("\\")[-1]
@@ -197,7 +225,7 @@ class RSLNamer(object):
self.submission_type = None
return
logger.debug(f"Attempting match of {in_str}")
print(f"The initial plate name is: {in_str}")
logger.debug(f"The initial plate name is: {in_str}")
regex = re.compile(r"""
# (?P<wastewater>RSL(?:-|_)?WW(?:-|_)?20\d{2}-?\d{2}-?\d{2}(?:(?:_|-)\d?((?!\d)|R)?\d(?!\d))?)|
(?P<wastewater>RSL(?:-|_)?WW(?:-|_)?20\d{2}-?\d{2}-?\d{2}(?:(_|-)\d?(\D|$)R?\d?)?)|
@@ -206,7 +234,7 @@ class RSLNamer(object):
""", flags = re.IGNORECASE | re.VERBOSE)
m = regex.search(in_str)
try:
self.parsed_name = m.group().upper()
self.parsed_name = m.group().upper().strip(".")
logger.debug(f"Got parsed submission name: {self.parsed_name}")
self.submission_type = m.lastgroup
except AttributeError as e:
@@ -221,15 +249,15 @@ class RSLNamer(object):
self.parsed_name = self.parsed_name.replace("RSLWW", "RSL-WW")
self.parsed_name = re.sub(r"WW(\d{4})", r"WW-\1", self.parsed_name, flags=re.IGNORECASE)
self.parsed_name = re.sub(r"(\d{4})-(\d{2})-(\d{2})", r"\1\2\3", self.parsed_name)
print(f"Coming out of the preliminary parsing, the plate name is {self.parsed_name}")
logger.debug(f"Coming out of the preliminary parsing, the plate name is {self.parsed_name}")
try:
plate_number = re.search(r"(?:(-|_)\d)(?!\d)", self.parsed_name).group().strip("_").strip("-")
print(f"Plate number is: {plate_number}")
logger.debug(f"Plate number is: {plate_number}")
except AttributeError as e:
plate_number = "1"
# self.parsed_name = re.sub(r"(\d{8})(-|_\d)?(R\d)?", fr"\1-{plate_number}\3", self.parsed_name)
self.parsed_name = re.sub(r"(\d{8})(-|_)?\d?(R\d?)?", rf"\1-{plate_number}\3", self.parsed_name)
print(f"After addition of plate number the plate name is: {self.parsed_name}")
logger.debug(f"After addition of plate number the plate name is: {self.parsed_name}")
try:
repeat = re.search(r"-\dR(?P<repeat>\d)?", self.parsed_name).groupdict()['repeat']
if repeat == None:
@@ -239,8 +267,6 @@ class RSLNamer(object):
self.parsed_name = re.sub(r"(-\dR)\d?", rf"\1 {repeat}", self.parsed_name).replace(" ", "")
def enforce_bacterial_culture(self):
"""
Uses regex to enforce proper formatting of bacterial culture samples
@@ -249,6 +275,9 @@ class RSLNamer(object):
self.parsed_name = re.sub(r"RSL-(\d{2})(\d{4})", r"RSL-\1-\2", self.parsed_name, flags=re.IGNORECASE)
def enforce_wastewater_artic(self):
"""
Uses regex to enforce proper formatting of wastewater samples
"""
self.parsed_name = re.sub(r"(\d{4})-(\d{2})-(\d{2})", r"RSL-AR-\1\2\3", self.parsed_name, flags=re.IGNORECASE)
try:
plate_number = int(re.search(r"_\d?_", self.parsed_name).group().strip("_"))