diff --git a/index.py b/index.py index fa46416..896f1c9 100644 --- a/index.py +++ b/index.py @@ -129,6 +129,8 @@ alerts = html.Div( [ dbc.Alert("Error: More than one qPCR file uploaded. Please upload only one.", color="danger",id="alert-fade-too-many-qpcr", dismissable=False, is_open=False), + dbc.Alert(id="alert-find-libraries-fail", color="danger", is_open=False, dismissable=True, children=""), + dbc.Alert(id="alert-fade-duplicate-file", color="danger", is_open=False, dismissable=True, children=""), dbc.Alert( id="alert-tapestation-sampleid", color="warning", @@ -137,8 +139,7 @@ children="", ), dbc.Alert(id="alert-find-libraries-success", color="success", is_open=False, dismissable=True, children=""), - dbc.Alert(id="alert-find-libraries-fail", color="danger", is_open=False, dismissable=True, children=""), - dbc.Alert(id="alert-fade-duplicate-file", color="danger", is_open=False, dismissable=True, children=""), + ], style={"margin": "20px"} ) @@ -368,32 +369,6 @@ def plate_mapper(plate_ids, objects): # --- Handle Plate Upload --- if triggered == "upload-files": - # TODO: Complete Re-Factoring of this function - # Is there already a qPCR plate in state? - qpcr_found = any(p.plate_type == "qPCR" for p in app_state.plate_registry) - - # Check if any uploaded file looks like qPCR (contains 'abs quat' after cleaning) - incoming_has_abs_quat = False - for fname in filenames: - clean_name = fname.lower() - clean_name = re.sub(r'\.ref\.(csv|json)$', '', clean_name) - if "abs quant" in clean_name: - incoming_has_abs_quat = True - break - - # If qPCR already exists and new file has 'abs quat' in name -> block upload - if qpcr_found and incoming_has_abs_quat: - warnings = collect_missing_sample_warnings(app_state) - show_warning = bool(warnings) - warning_msg = build_warning_msg(warnings) - return ( - True, "Error: A qPCR file has already been uploaded. Only one such file is allowed.", - app_state.to_json(), show_warning, warning_msg, - False, "", False, "", None, False, "", no_update, no_update - ) - - - # print("uploaded_files",app_state.uploaded_files_base64) # --- Deduplication logic --- duplicates = [] files_to_upload = [] @@ -411,12 +386,12 @@ def plate_mapper(plate_ids, objects): warning_msg = build_warning_msg(warnings) if len(duplicates) == 1: - duplicate_msg = f"Warning: Upload not successful! File '{duplicates[0]}' has already been uploaded." + duplicate_msg = f"Upload failed: The file {duplicates[0]} has already been uploaded." else: duplicate_msg = html.Div([ html.Strong("Warning: Upload not successful!"), html.Br(), - "The following file(s) have already been uploaded and were skipped:", + "The following file(s) have already been uploaded:", html.Ul([html.Li(fname) for fname in duplicates]) ]) @@ -427,11 +402,23 @@ def plate_mapper(plate_ids, objects): ) + temp_state = copy.deepcopy(app_state) # Work on a copy first; only commit if all uploads succeed! if files_to_upload: for content, fname in zip(files_to_upload, filenames_to_upload): - app_state.uploaded_files_base64[fname] = content - handle_plate_upload(files_to_upload, filenames_to_upload, app_state) - + upload_status, temp_state, failure_msg = handle_plate_upload(content, fname, temp_state) + if upload_status: + temp_state.uploaded_files_base64[fname] = content + else: + warnings = collect_missing_sample_warnings(app_state) + show_warning = bool(warnings) + warning_msg = build_warning_msg(warnings) + return ( + True, failure_msg, + app_state.to_json(), show_warning, warning_msg, + False, "", False, "", None, False, "", no_update, no_update + ) + + app_state = temp_state warnings = collect_missing_sample_warnings(app_state) show_warning = bool(warnings) warning_msg = build_warning_msg(warnings) @@ -489,7 +476,10 @@ def plate_mapper(plate_ids, objects): else: # Plate View selected_data = selected_data_map.get(plate_id) plate = app_state.plate_by_id.get(plate_id) - selected_wells = [p['customdata'][0] for p in selected_data['points']] + if not selected_data: + selected_wells = [] + else: + selected_wells = [p['customdata'][0] for p in selected_data['points']] if selected_wells: to_remove = set(selected_wells) plate.plate_wells = [ @@ -833,7 +823,10 @@ def update_plate_buttons(app_data_state_json, selected_tab): df = plate.dataset(app_state, tidy=False) # Only consider rows where sample_type == "PhiX" - phix_rows = df[df.get("sample_type") == "PhiX"] + if "sample_type" in df.columns: + phix_rows = df[df["sample_type"] == "PhiX"] + else: + phix_rows = pd.DataFrame() # Prepare dropdown options dropdown_options = [] @@ -877,18 +870,6 @@ def update_plate_buttons(app_data_state_json, selected_tab): id={"type": "dummy-flagging-output", "index": plate_id}, style={"display": "none"}, ), - - # --- Remove Flagged Samples Button --- - dbc.Button( - "Remove Flagged Samples", - id={"type": "remove-flagged-btn", "index": plate_id}, - color="danger", # make this action visually distinct - className="w-100 py-2 fw-semibold small shadow-sm", - ), - html.Div( - id={"type": "dummy-remove-output", "index": plate_id}, - style={"display": "none"}, - ), ], className="p-2 bg-light rounded shadow-sm", # optional card-like background ) @@ -904,19 +885,6 @@ def on_adjust_flagging_clicked(n_clicks): print(f"Adjust flagging condition button is pressed for plate {plate_id}") return "" - -@app.callback( - Output({"type": "dummy-remove-output", "index": MATCH}, "children"), - Input({"type": "remove-flagged-btn", "index": MATCH}, "n_clicks"), - prevent_initial_call=True -) -def on_remove_flagged_clicked(n_clicks): - plate_id = ctx.triggered_id["index"] - print(f"Remove flagged samples button is pressed for plate {plate_id}") - return "" - - - # @app.callback( # [Output({"type": "flagging-modal", "index": MATCH}, "is_open"), # Output({"type": "flagging-modal-body", "index": MATCH}, "children")], diff --git a/models.py b/models.py index 403f12c..52c690b 100644 --- a/models.py +++ b/models.py @@ -7,6 +7,22 @@ import numpy as np from enum import Enum +# Signature columns for file type detection +SIGNATURE_COLUMNS = { + "QPCR": { + "Sample Name", "Gene Name", "Cq", "Concentration", "Position", "Call" + }, + "HSD1000": { + "FileName", "WellId", "Sample Description", "From [bp]", "To [bp]", "Conc. [pg/µl]" + }, + "D1000": { + "FileName", "WellId", "Sample Description", "From [bp]", "To [bp]", "Conc. [ng/µl]" + }, + "D5000": { + "FileName", "WellId", "Sample Description", "From [bp]", "To [bp]", "Conc. [ng/µl]" + }, +} + ###################################################### # SampleType Enum ###################################################### @@ -329,19 +345,6 @@ def safe_float(val): - - - - - - - - - - - - - ###################################################### # App Specific Part ###################################################### @@ -398,15 +401,6 @@ def create_from_raw_data(cls, name: str, raw_data: str): # Needs to be more general! class AppSpecificFile(PlateFile): - FILE_PATTERNS = [ - (re.compile(r'hsd1000', re.IGNORECASE), "HSD1000"), - (re.compile(r'd1000', re.IGNORECASE), "D1000"), - (re.compile(r'd5000', re.IGNORECASE), "D5000"), - (re.compile(r'abs quant', re.IGNORECASE), "QPCR"), - (re.compile(r'qpcr', re.IGNORECASE), "QPCR"), - ] - - def __init__(*args, **kwargs): raise NotImplementedError("Use the factory method create_from_raw_data to instantiate AppSpecificFile.") @@ -423,7 +417,7 @@ def create_from_raw_data(cls, name: str, raw_data: str): It detects the file type and initializes the appropriate subclass. """ - file_type = cls.detect_file_type(name) + file_type = cls.detect_file_type_from_content(raw_data, name) match file_type: case "D1000" | "D5000": @@ -576,16 +570,35 @@ def create_from_raw_data(cls, name: str, raw_data: str): case _: raise ValueError(f"Unknown file type for file: {name}. Please check the file name or implement a new file type handler.") - @classmethod - def detect_file_type(cls, name: str): - clean_name = name.lower() - clean_name = re.sub(r'\.ref\.(csv|json)$', '', clean_name) - for pattern, file_str in cls.FILE_PATTERNS: - if pattern.search(clean_name): - print(f"Detected file type: {file_str} for file: {name}") - return file_str - raise ValueError(f"Unknown file type for file: {name}") + def detect_file_type_from_content(cls, raw_data: str, file_name: str = "") -> str: + try: + df = pd.read_csv(io.StringIO(raw_data), sep=None, engine="python", encoding="utf-8-sig") + df.columns = [c.replace('\ufeff', '').strip() for c in df.columns] + file_columns = set(df.columns) + except Exception as e: + raise ValueError( + "File is not in the correct format (CSV/TXT with a valid delimiter). " + "Please upload only qPCR or Tapestation files (D1000, D5000, HSD1000). " + f"Details: {str(e)}" + ) from e + + if SIGNATURE_COLUMNS["QPCR"] <= file_columns: + return "QPCR" + if SIGNATURE_COLUMNS["HSD1000"] <= file_columns: + return "HSD1000" + if SIGNATURE_COLUMNS["D1000"] <= file_columns: + clean_name = file_name.lower() + if "d5000" in clean_name: + return "D5000" + elif "d1000" in clean_name: + return "D1000" + else: + return "D1000" + raise ValueError( + "File is not in the correct format. Please upload only qPCR or Tapestation files (D1000, D5000, HSD1000)." + ) + def to_dataframe(self): diff --git a/utils/plate_render_utils.py b/utils/plate_render_utils.py index 040e1b7..a7170a3 100644 --- a/utils/plate_render_utils.py +++ b/utils/plate_render_utils.py @@ -18,6 +18,33 @@ "Library": "#43a047", # green } +TAPESTATION_HIDE_COLS = { + 'plate_id', + 'plate_type', + 'automatic_region', + 'file_sample_identifier', + '% of Total', + 'Area', + 'Conc. [ng/µl]', + 'Conc. [pg/µl]', + 'From [bp]', + 'To [bp]', +} + +QPCR_HIDE_COLS = { + 'plate_id', + 'plate_type', + 'replicate_group', + 'file_sample_identifier', + 'Concentration', + 'Concentration Error', + 'Concentration Mean', + 'Cq Error', + 'EPF', + 'Slope', + 'Standard', +} + # constants for the plate plot GRAPH_W, GRAPH_H, CELL_R = 560, 400, 30 @@ -332,6 +359,12 @@ def _is_num(s): } column_defs = [checkbox_col] + [c for c in column_defs if c["field"] not in ("__error__", "__warning__")] + if plate.plate_type != 'qPCR': + column_defs = [col for col in column_defs if col.get('field') not in TAPESTATION_HIDE_COLS] + + elif plate.plate_type == 'qPCR': + column_defs = [col for col in column_defs if col.get('field') not in QPCR_HIDE_COLS] + # --- AgGrid --- grid = dag.AgGrid( id={"type": "datatable-plate", "index": pid}, diff --git a/utils/plates_callback_utils.py b/utils/plates_callback_utils.py index 8501221..6bdd985 100644 --- a/utils/plates_callback_utils.py +++ b/utils/plates_callback_utils.py @@ -60,29 +60,38 @@ def find_and_populate_samples_by_sample_id( print("Finished populating sample info via sample IDs.") -def handle_plate_upload(contents, filenames, app_state): - for content, name in zip(contents, filenames): - - content_type, content_string = content.split(',') - decoded = base64.b64decode(content_string) - try: - decoded_str = decoded.decode("utf-8") - except UnicodeDecodeError: - decoded_str = decoded.decode("ISO-8859-1") - +def handle_plate_upload(content, name, app_state): + content_type, content_string = content.split(',') + decoded = base64.b64decode(content_string) + try: + decoded_str = decoded.decode("utf-8") + except UnicodeDecodeError: + decoded_str = decoded.decode("ISO-8859-1") + + try: uploaded_file = AppSpecificFile.create_from_raw_data(name=name, raw_data=decoded_str) + if uploaded_file.file_type_name == "qPCR": + if any(p.plate_type == "qPCR" for p in app_state.plate_registry): + failure_msg = "Upload failed: Only one qPCR plate can be uploaded." + return False, app_state, failure_msg app_state, plate = Plate.from_file(input_file=uploaded_file, app_state=app_state) - - # here we make the new functon to automatically populate sample_type app_state, plate = populate_sample_types(app_state, plate) if plate.plate_type in ("D1000", "D5000", "HSD1000"): plate = populate_samples_by_identifier_lookup(plate, app_state) - #print("intial app state sample_registry:", app_state.sample_registry) + # print("intial app state sample_registry:", app_state.sample_registry) # print("intial app state plate_registry:", app_state.plate_registry) # print("Added plate:", plate.plate_id) + except Exception as e: + print("Error in process_plate_upload:", e) + failure_msg = f"Upload failed: The file {name} is not valid. Please upload only qPCR or Tapestation files (D1000, D5000, HSD1000)." + return False, app_state, failure_msg + + return True, app_state, "" + + def handle_plate_delete(plate_id_to_delete, app_state): diff --git a/utils/warning_utils.py b/utils/warning_utils.py index 3746a0f..9650035 100644 --- a/utils/warning_utils.py +++ b/utils/warning_utils.py @@ -32,8 +32,8 @@ def collect_missing_sample_warnings(app_state): def build_warning_msg(warnings): instruction = ( - "To resolve these warnings: Please enter the correct sample_id values manually in the table for the affected wells, " - "then click 'Find Libraries'. The application will retrieve the corresponding sample information from B-Fabric and update the table automatically." + "To resolve these warnings: Please enter the correct sample_id values manually in the table for the affected wells." + "The application will retrieve the corresponding sample information from B-Fabric and update the table automatically." ) # Only show instruction if there are real warnings items = [html.Li(msg) for msg in warnings]