diff --git a/src/submissions/backend/db/models/kits.py b/src/submissions/backend/db/models/kits.py index c09310a..c49beb0 100644 --- a/src/submissions/backend/db/models/kits.py +++ b/src/submissions/backend/db/models/kits.py @@ -6,6 +6,7 @@ import json, zipfile, yaml, logging, re, sys from operator import itemgetter from pprint import pformat +import numpy as np from jinja2 import Template, TemplateNotFound from sqlalchemy import Column, String, TIMESTAMP, JSON, INTEGER, ForeignKey, Interval, Table, FLOAT, BLOB from sqlalchemy.orm import relationship, validates, Query @@ -539,9 +540,19 @@ class ReagentRole(BaseClass): logger.debug(f"Constructing OmniReagentRole with name {self.name}") return OmniReagentRole(instance_object=self, name=self.name, eol_ext=self.eol_ext) - @property - def reagents(self): - return [f"{reagent.name} - {reagent.lot}" for reagent in self.reagent] + def get_reagents(self, kittype: str | KitType | None = None): + if not kittype: + return [f"{reagent.name} - {reagent.lot}" for reagent in self.reagent] + if isinstance(kittype, str): + kittype = KitType.query(name=kittype) + assoc = next((item for item in self.reagentrolekittypeassociation if item.kittype == kittype), None) + reagents = [reagent for reagent in self.reagent] + if assoc: + last_used = Reagent.query(name=assoc.last_used) + if last_used: + reagents.insert(0, reagents.pop(reagents.index(last_used))) + return [f"{reagent.name} - {reagent.lot}" for reagent in reagents] + class Reagent(BaseClass, LogMixin): """ @@ -1151,21 +1162,25 @@ class ProcedureType(BaseClass): def as_dict(self): return dict( name=self.name, - kittype=[item.name for item in self.kittype] + kittype=[item.name for item in self.kittype], + plate_rows=self.plate_rows, + plate_columns=self.plate_columns ) - def construct_dummy_procedure(self): + def construct_dummy_procedure(self, run: Run|None=None): from backend.validators.pydant import PydProcedure + if run: + samples = run.constuct_sample_dicts_for_proceduretype(proceduretype=self) output = dict( proceduretype=self, #name=dict(value=self.name, missing=True), #possible_kits=[kittype.name for kittype in self.kittype], - repeat=False, - plate_map=self.construct_plate_map() + repeat=False + # plate_map=plate_map ) return PydProcedure(**output) - def construct_plate_map(self) -> str: + def construct_plate_map(self, sample_dicts: List[dict]) -> str: """ Constructs an html based plate map for procedure details. @@ -1179,22 +1194,31 @@ class ProcedureType(BaseClass): """ if self.plate_rows == 0 or self.plate_columns == 0: return "
" - plate_rows = range(1, self.plate_rows + 1) - plate_columns = range(1, self.plate_columns + 1) - total_wells = self.plate_columns * self.plate_rows - vw = round((-0.07 * total_wells) + 12.2, 1) - - - wells = [dict(name="", row=row, column=column, background_color="#ffffff") - for row in plate_rows - for column in plate_columns] + # plate_rows = range(1, self.plate_rows + 1) + # plate_columns = range(1, self.plate_columns + 1) + # total_wells = self.plate_columns * self.plate_rows + vw = round((-0.07 * len(sample_dicts)) + 12.2, 1) + # sample_dicts = run.constuct_sample_dicts_for_proceduretype(proceduretype=self) + # output_samples = [next((item for item in sample_dicts if item['row'] == row and item['column'] == column), + # dict(sample_id="", row=row, column=column, background_color="#ffffff")) + # for row in plate_rows + # for column in plate_columns] + # logger.debug(f"Output samples:\n{pformat(output_samples)}") # NOTE: An overly complicated list comprehension create a list of sample locations # NOTE: next will return a blank cell if no value found for row/column env = jinja_template_loading() template = env.get_template("plate_map.html") - html = template.render(plate_rows=self.plate_rows, plate_columns=self.plate_columns, samples=wells, vw=vw) + html = template.render(plate_rows=self.plate_rows, plate_columns=self.plate_columns, samples=sample_dicts, vw=vw) return html + "
" + @property + def ranked_plate(self): + matrix = np.array([[0 for yyy in range(1, self.plate_rows + 1)] for xxx in range(1, self.plate_columns + 1)]) + return {iii: (item[0][1] + 1, item[0][0] + 1) for iii, item in enumerate(np.ndenumerate(matrix), start=1)} + + @property + def total_wells(self): + return self.plate_rows * self.plate_columns class Procedure(BaseClass): id = Column(INTEGER, primary_key=True) diff --git a/src/submissions/backend/db/models/submissions.py b/src/submissions/backend/db/models/submissions.py index 04280a9..2705762 100644 --- a/src/submissions/backend/db/models/submissions.py +++ b/src/submissions/backend/db/models/submissions.py @@ -28,11 +28,13 @@ from openpyxl.drawing.image import Image as OpenpyxlImage from tools import row_map, setup_lookup, jinja_template_loading, rreplace, row_keys, check_key_or_attr, Result, Report, \ report_result, create_holidays_for_year, check_dictionary_inclusion_equality from datetime import datetime, date -from typing import List, Any, Tuple, Literal, Generator, Type +from typing import List, Any, Tuple, Literal, Generator, Type, TYPE_CHECKING from pathlib import Path from jinja2.exceptions import TemplateNotFound from jinja2 import Template from PIL import Image +if TYPE_CHECKING: + from backend.db.models.kits import ProcedureType from . import kittype_procedure @@ -653,6 +655,10 @@ class Run(BaseClass, LogMixin): output_list = [assoc.hitpicked for assoc in self.runsampleassociation] return output_list + @property + def sample_dicts(self) -> List[dict]: + return [dict(sample_id=assoc.sample.sample_id, row=assoc.row, column=assoc.column, background_color="#6ffe1d") for assoc in self.runsampleassociation] + @classmethod def make_plate_map(cls, sample_list: list, plate_rows: int = 8, plate_columns=12) -> str: """ @@ -1273,6 +1279,51 @@ class Run(BaseClass, LogMixin): def allowed_procedures(self): return self.clientsubmission.submissiontype.proceduretype + def get_submission_rank_of_sample(self, sample: Sample|str): + if isinstance(sample, str): + sample = Sample.query(sample_id=sample) + clientsubmissionsampleassoc = next((assoc for assoc in self.clientsubmission.clientsubmissionsampleassociation + if assoc.sample == sample), None) + if clientsubmissionsampleassoc: + return clientsubmissionsampleassoc.submission_rank + else: + return 0 + + def constuct_sample_dicts_for_proceduretype(self, proceduretype: ProcedureType): + plate_dict = proceduretype.ranked_plate + ranked_samples = [] + unranked_samples = [] + for sample in self.sample: + submission_rank = self.get_submission_rank_of_sample(sample=sample) + if submission_rank != 0: + row, column = plate_dict[submission_rank] + ranked_samples.append(dict(sample_id=sample.sample_id, row=row, column=column, submission_rank=submission_rank, background_color="#6ffe1d")) + else: + unranked_samples.append(sample) + possible_ranks = (item for item in list(plate_dict.keys()) if item not in [sample['submission_rank'] for sample in ranked_samples]) + # logger.debug(possible_ranks) + # possible_ranks = (plate_dict[idx] for idx in possible_ranks) + for sample in unranked_samples: + try: + submission_rank = next(possible_ranks) + except StopIteration: + continue + row, column = plate_dict[submission_rank] + ranked_samples.append( + dict(sample_id=sample.sample_id, row=row, column=column, submission_rank=submission_rank, + background_color="#6ffe1d")) + padded_list = [] + for iii in range(1, proceduretype.total_wells+1): + sample = next((item for item in ranked_samples if item['submission_rank']==iii), + dict(sample_id="", row=0, column=0, submission_rank=iii) + ) + padded_list.append(sample) + logger.debug(f"Final padded list:\n{pformat(list(sorted(padded_list, key=itemgetter('submission_rank'))))}") + return list(sorted(padded_list, key=itemgetter('submission_rank'))) + + + + class SampleType(BaseClass): id = Column(INTEGER, primary_key=True) #: primary key diff --git a/src/submissions/backend/validators/pydant.py b/src/submissions/backend/validators/pydant.py index d67f9cd..36c3dd9 100644 --- a/src/submissions/backend/validators/pydant.py +++ b/src/submissions/backend/validators/pydant.py @@ -243,6 +243,8 @@ class PydSample(PydBaseClass): sampletype: str | None = Field(default=None) submission_rank: int | List[int] | None = Field(default=0, validate_default=True) enabled: bool = Field(default=True) + row: int = Field(default=0) + column: int = Field(default=0) @field_validator("sample_id", mode="before") @classmethod @@ -1336,6 +1338,7 @@ class PydElastic(BaseModel, extra="allow", arbitrary_types_allowed=True): class PydProcedure(PydBaseClass, arbitrary_types_allowed=True): proceduretype: ProcedureType | None = Field(default=None) + run: Run | None = Field(default=None) name: dict = Field(default=dict(value="NA", missing=True), validate_default=True) technician: dict = Field(default=dict(value="NA", missing=True)) repeat: bool = Field(default=False) @@ -1344,6 +1347,7 @@ class PydProcedure(PydBaseClass, arbitrary_types_allowed=True): plate_map: str | None = Field(default=None) reagent: list | None = Field(default=[]) reagentrole: dict | None = Field(default={}, validate_default=True) + samples: List[PydSample] = Field(default=[]) @field_validator("name") @classmethod @@ -1379,18 +1383,68 @@ class PydProcedure(PydBaseClass, arbitrary_types_allowed=True): value = {item.name: item.reagents for item in kittype.reagentrole} return value + + def update_kittype_reagentroles(self, kittype: str | KitType): if kittype == self.__class__.model_fields['kittype'].default['value']: return if isinstance(kittype, str): kittype_obj = KitType.query(name=kittype) try: - self.reagentrole = {item.name: item.reagents for item in + self.reagentrole = {item.name: item.get_reagents(kittype=kittype_obj) for item in kittype_obj.get_reagents(proceduretype=self.proceduretype)} except AttributeError: self.reagentrole = {} self.possible_kits.insert(0, self.possible_kits.pop(self.possible_kits.index(kittype))) + def shuffle_samples(self, source_row: int, source_column: int, destination_row: int, destination_column=int): + logger.debug(f"Attempting sample shuffle.") + try: + source_sample = next( + (sample for sample in self.samples if sample.row == source_row and sample.column == source_column)) + except StopIteration: + raise StopIteration("Couldn't find proper sample.") + logger.debug(f"Source Well: {source_row}, {source_column}") + logger.debug(f"Destination Well: {destination_row}, {destination_column}") + updateable_samples = [] + if source_row > destination_row and source_column >= destination_column: + logger.debug(f"Sample was moved ahead.") + movement = "pos" + for sample in self.samples: + if sample.row >= destination_row and sample.column >= destination_column: + if sample.row <= source_row and sample.column <= source_column: + updateable_samples.append(sample) + + elif source_row < destination_row and source_column <= destination_column: + logger.debug(f"Sample was moved back.") + movement = "neg" + for sample in self.samples: + if sample.row <= destination_row and sample.column <= destination_column: + if sample.row >= source_row and sample.column >= source_column: + updateable_samples.append(sample) + else: + logger.debug(f"Don't know what happened.") + logger.debug(f"Samples to be updated: {pformat(updateable_samples)}") + for sample in updateable_samples: + if sample.row == source_row and sample.column == source_column: + sample.row = destination_row + sample.column = destination_column + else: + match movement: + case "pos": + if sample.row + 1 > 8: + sample.column += 1 + sample.row = 1 + else: + sample.row += 1 + case "neg": + if sample.row - 1 <= 0: + sample.column -= 1 + sample.row = 8 + else: + sample.row -= 1 + + class PydClientSubmission(PydBaseClass): # sql_object: ClassVar = ClientSubmission diff --git a/src/submissions/templates/css/styles.css b/src/submissions/templates/css/styles.css index a14b6cf..45488b9 100644 --- a/src/submissions/templates/css/styles.css +++ b/src/submissions/templates/css/styles.css @@ -84,3 +84,29 @@ div.gallery { width: 100%; object-fit: contain; } + +.plate { + display: inline-grid; + grid-auto-flow: column; + place-content: center center; + grid-gap: 2px; +} + +.well { + border: 1px solid #000; + padding: 20px; + display: flex; + text-align: center; + box-shadow: 0 2px 10px rgba(0, 0, 0, 0.1); + transition: all 0.3s ease; + cursor: move; +} + +.well:hover { + box-shadow: 0 4px 25px rgba(0, 0, 0, 0.5); +} + +.grid-item p { + max-width: 100%; + border-radius: 8px; +} \ No newline at end of file diff --git a/src/submissions/templates/plate_map.html b/src/submissions/templates/plate_map.html index 586db7d..d7aa2bb 100644 --- a/src/submissions/templates/plate_map.html +++ b/src/submissions/templates/plate_map.html @@ -1,17 +1,15 @@ -
+
{% for sample in samples %} - {% endfor %} -
\ No newline at end of file + +
diff --git a/src/submissions/templates/procedure_creation.html b/src/submissions/templates/procedure_creation.html new file mode 100644 index 0000000..23d72a3 --- /dev/null +++ b/src/submissions/templates/procedure_creation.html @@ -0,0 +1,124 @@ +{% extends "details.html" %} + + + + {% block head %} + {{ super() }} + New {{ proceduretype['name'] }} for {{ run['plate_number'] }} + {% endblock %} + + + + {% block body %} +

New {{ proceduretype['name'] }} for {{ run['plate_number'] }}



+
+
+
+
+

+ +

+
+
+ {% if procedure['reagentrole'] %} +

+ {% for key, value in procedure['reagentrole'].items() %} +
+ + {% endfor %} + {% endif %} +
+
+
+ {% if plate_map %} +

Plate map:

+ {{ plate_map }} + {% endif %} +
+
+ {% endblock %} + + + \ No newline at end of file diff --git a/src/submissions/tools/__init__.py b/src/submissions/tools/__init__.py index 2f0535c..2fd88e8 100644 --- a/src/submissions/tools/__init__.py +++ b/src/submissions/tools/__init__.py @@ -27,7 +27,6 @@ from sqlalchemy.exc import IntegrityError as sqlalcIntegrityError from pytz import timezone as tz from functools import wraps - timezone = tz("America/Winnipeg") logger = logging.getLogger(f"procedure.{__name__}") @@ -249,6 +248,7 @@ def timer(func): func (__function__): incoming function """ + @wraps(func) def wrapper(*args, **kwargs): start_time = time.perf_counter() @@ -257,6 +257,7 @@ def timer(func): run_time = end_time - start_time print(f"Finished {func.__name__}() in {run_time:.4f} secs") return value + return wrapper @@ -295,7 +296,6 @@ class GroupWriteRotatingFileHandler(handlers.RotatingFileHandler): class CustomFormatter(logging.Formatter): - class bcolors: HEADER = '\033[95m' OKBLUE = '\033[94m' @@ -482,6 +482,7 @@ def setup_lookup(func): elif v is not None: sanitized_kwargs[k] = v return func(*args, **sanitized_kwargs) + return wrapper @@ -532,7 +533,6 @@ def get_application_from_parent(widget): class Result(BaseModel, arbitrary_types_allowed=True): - owner: str = Field(default="", validate_default=True) code: int = Field(default=0) msg: str | Exception @@ -768,6 +768,7 @@ def under_development(func): Result(owner=func.__str__(), code=1, msg=error_msg, status="warning")) return report + return wrapper @@ -856,6 +857,7 @@ def create_holidays_for_year(year: int | None = None) -> List[date]: offset = -d.weekday() # weekday == 0 means Monday output = d + timedelta(offset) return output.date() + if not year: year = date.today().year # NOTE: Includes New Year's day for next year. @@ -900,6 +902,11 @@ def flatten_list(input_list: list): return list(itertools.chain.from_iterable(input_list)) +def create_plate_grid(rows: int, columns: int): + matrix = np.array([[0 for yyy in range(1, columns + 1)] for xxx in range(1, rows + 1)]) + return {iii: (item[0][1]+1, item[0][0]+1) for iii, item in enumerate(np.ndenumerate(matrix), start=1)} + + class classproperty(property): def __get__(self, owner_self, owner_cls): return self.fget(owner_cls) @@ -1293,4 +1300,3 @@ class Settings(BaseSettings, extra="allow"): ctx = Settings() -