M2 and clean up tasks

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")],

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):

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},

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):

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]