Update app.py

app.py

@@ -1,12 +1,15 @@
 import streamlit as st
 import pandas as pd
 import io
+import re
+import numpy as np
+
 
 # =========================
 # Streamlit App Setup
 # =========================
 st.set_page_config(page_title="ASCII ↔ Binary Converter", layout="wide")
-st.title("🔢 ASCII ↔ Binary Converter")
+st.title("ASCII ↔ Binary Converter")
 
 # =========================
 # Voyager ASCII 6-bit Table
@@ -47,7 +50,7 @@ def binary_labels_to_string(bits: list[int]) -> str:
 # =========================
 # Tabs
 # =========================
-tab1, tab2 = st.tabs(["Text → Binary", "Binary → Text"])
+tab1, tab2, tab3 = st.tabs(["Text → Binary", "Binary → Text", "Robot Script"])
 
 # --------------------------------------------------
 # TAB 1: Text → Binary
@@ -63,7 +66,7 @@ with tab1:
 
     col1, col2 = st.columns([2, 1])
     with col1:
-        group_size = st.slider("Select number of positions per group:", min_value=12, max_value=32, value=25)
+        group_size = st.slider("Select number of positions per row:", min_value=12, max_value=32, value=25)
     with col2:
         custom_cols = st.number_input("Or enter custom number:", min_value=1, max_value=128, value=group_size)
     if custom_cols != group_size:
@@ -165,3 +168,199 @@ with tab2:
             st.error(f"Error reading or converting file: {e}")
     else:
         st.info("👆 Upload a file to start the reverse conversion.")
+
+# --------------------------------------------------
+# TAB 3: Pipetting Command Generator
+# --------------------------------------------------
+
+
+with tab3:
+    st.header("🧪 Pipetting Command Generator")
+    st.markdown("""
+    Upload your sample file (Excel, CSV, or TXT) containing binary mutation data.
+    The app will:
+    - Auto-detect or create `Sample`, `Position#`, `Total edited`, and `Volume per "1"` columns  
+    - Generate pipetting commands dynamically for any number of positions  
+    - Display and allow CSV download for both commands and source volume summaries  
+    """)
+
+    uploaded = st.file_uploader("📤 Upload data file", type=["xlsx", "csv", "txt"])
+    max_per_well_ul = st.number_input("Maximum volume per source well (µL)", min_value=10.0, max_value=1000.0, value=160.0, step=10.0)
+
+    if uploaded is not None:
+        try:
+            # --- Load file ---
+            if uploaded.name.endswith(".xlsx"):
+                sheet_names = pd.ExcelFile(uploaded).sheet_names
+                sheet_choice = st.selectbox("Select sheet:", sheet_names)
+                df = pd.read_excel(uploaded, sheet_name=sheet_choice)
+            elif uploaded.name.endswith(".csv"):
+                df = pd.read_csv(uploaded)
+            else:  # TXT
+                df = pd.read_csv(uploaded, sep="\t")
+
+            st.success(f"✅ Loaded file with {len(df)} rows and {len(df.columns)} columns")
+
+            # --- Clean column names ---
+            df.columns = [str(c).strip() for c in df.columns]
+
+            # --- Ensure Sample column ---
+            if not any(c.lower() == "sample" for c in df.columns):
+                df.insert(0, "Sample", np.arange(1, len(df) + 1))
+                st.info("`Sample` column missing — automatically generated 1..N.")
+
+            # --- Detect Position columns ---
+            position_cols = [c for c in df.columns if re.match(r"(?i)^position\s*\d+", c)]
+            if not position_cols:
+                # If not found, auto-generate for numeric-like columns
+                non_pos_cols = {"sample", "total edited", 'volume per "1"', "volume per 1"}
+                candidate_cols = [c for c in df.columns if c.lower() not in non_pos_cols]
+                if not candidate_cols:
+                    st.error("❌ Could not detect any Position columns.")
+                    st.stop()
+                position_cols = candidate_cols
+                st.info(f"Position columns inferred automatically: {len(position_cols)} detected.")
+
+            # --- Ensure Total edited ---
+            if "Total edited" not in df.columns:
+                df["Total edited"] = df[position_cols].apply(pd.to_numeric, errors="coerce").fillna(0).sum(axis=1).astype(int)
+                st.info("`Total edited` column missing — calculated automatically as sum of 1s per row.")
+
+            # --- Ensure Volume per "1" ---
+            vol_candidates = [c for c in df.columns if "volume per" in c.lower()]
+            if not vol_candidates:
+                df['Volume per "1"'] = 64 / df["Total edited"].replace(0, np.nan)
+                df['Volume per "1"'] = df['Volume per "1"'].fillna(0)
+                st.info('`Volume per "1"` column missing — calculated automatically as 64 / Total edited.')
+                volume_col = 'Volume per "1"'
+            else:
+                volume_col = vol_candidates[0]
+
+            # --- Constants ---
+            ROWS_96 = ["A", "B", "C", "D", "E", "F", "G", "H"]
+            COLS_96 = list(range(1, 13))
+            GROUP_START_COLS = [1, 5, 9]
+            num_positions = len(position_cols)
+
+            def well_name(row_letter, col_number):
+                return f"{row_letter}{col_number}"
+
+            def sample_index_to_plate_and_well(sample_idx):
+                plate_num = ((sample_idx - 1) // 96) + 1
+                within_plate = (sample_idx - 1) % 96
+                row_idx = within_plate // 12
+                col_idx = within_plate % 12
+                return plate_num, well_name(ROWS_96[row_idx], COLS_96[col_idx])
+
+            def source_index_to_group_coords(source_idx):
+                group_idx = source_idx - 1
+                plate_num = 1 + (group_idx // 24)
+                within_plate_group = group_idx % 24
+                row_idx = within_plate_group // 3
+                group_in_row = within_plate_group % 3
+                row_letter = ROWS_96[row_idx]
+                start_col = GROUP_START_COLS[group_in_row]
+                return plate_num, row_letter, start_col
+
+            def source_index_to_wells(source_idx):
+                plate_num, row_letter, start_col = source_index_to_group_coords(source_idx)
+                return [(plate_num, well_name(row_letter, start_col + offset)) for offset in range(4)]
+
+            def pick_tool(volume_ul):
+                return "TS_10" if volume_ul <= 10.0 else "TS_50"
+
+            # --- Core logic ---
+            commands = []
+            source_volume_totals = {}
+            per_source_well_cum = {i: [0.0, 0.0, 0.0, 0.0] for i in range(1, num_positions + 1)}
+            per_source_wells = {i: source_index_to_wells(i) for i in range(1, num_positions + 1)}
+
+            for _, row in df.iterrows():
+                sample_id = int(row["Sample"])
+                vol_per_one = float(row[volume_col])
+                if vol_per_one > max_per_well_ul:
+                    st.warning(f"Volume per '1' ({vol_per_one} µL) exceeds per-well cap of {max_per_well_ul} µL for sample {sample_id}.")
+                dest_plate, dest_well = sample_index_to_plate_and_well(sample_id)
+                tool = pick_tool(vol_per_one)
+
+                for pos_idx, col in enumerate(position_cols, 1):
+                    val = row[col]
+                    try:
+                        is_one = (float(val) == 1.0)
+                    except:
+                        is_one = str(val).strip() == "1"
+                    if not is_one:
+                        continue
+
+                    wells_list = per_source_wells[pos_idx]
+                    cum_list = per_source_well_cum[pos_idx]
+
+                    chosen = None
+                    for w_i, ((plate_num, well_str), current_vol) in enumerate(zip(wells_list, cum_list)):
+                        if current_vol + vol_per_one <= max_per_well_ul:
+                            chosen = (w_i, plate_num, well_str)
+                            break
+                    if chosen is None:
+                        st.error(f"Source {pos_idx} has all 4 wells full. Cannot add {vol_per_one} µL.")
+                        st.stop()
+
+                    w_i, src_plate, src_well = chosen
+                    cum_list[w_i] += vol_per_one
+                    per_source_well_cum[pos_idx] = cum_list
+                    source_volume_totals[(src_plate, src_well)] = source_volume_totals.get((src_plate, src_well), 0.0) + vol_per_one
+
+                    commands.append({
+                        "SourceIdx": pos_idx,
+                        "Source plate": src_plate,
+                        "Source well": src_well,
+                        "Destination plate": dest_plate,
+                        "Destination well": dest_well,
+                        "Volume": round(vol_per_one, 2),
+                        "Tool": tool
+                    })
+
+            # --- Compile results ---
+            commands_df = pd.DataFrame(commands)
+            commands_df = commands_df.sort_values(
+                by=["SourceIdx", "Source plate", "Source well", "Destination plate", "Destination well"],
+                kind="stable"
+            )
+            commands_df = commands_df[["Source plate", "Source well", "Destination plate", "Destination well", "Volume", "Tool"]]
+
+            # --- Source summary ---
+            summary_rows = []
+            for src_idx in range(1, num_positions + 1):
+                for plate_num, well_str in per_source_wells[src_idx]:
+                    total = source_volume_totals.get((plate_num, well_str), 0.0)
+                    summary_rows.append({
+                        "Source": src_idx,
+                        "Source plate": plate_num,
+                        "Source well": well_str,
+                        "Total volume taken (µL)": round(total, 2)
+                    })
+            summary_df = pd.DataFrame(summary_rows)
+
+            st.success(f"✅ Generated {len(commands_df)} commands across {num_positions} positions.")
+
+            st.markdown("### 💧 Pipetting Commands")
+            st.dataframe(commands_df, use_container_width=True, height=400)
+            st.download_button(
+                "⬇️ Download Commands CSV",
+                commands_df.to_csv(index=False),
+                "pipetting_commands.csv",
+                mime="text/csv"
+            )
+
+            st.markdown("### 📊 Source Volume Summary")
+            st.dataframe(summary_df, use_container_width=True, height=400)
+            st.download_button(
+                "⬇️ Download Source Summary CSV",
+                summary_df.to_csv(index=False),
+                "source_volume_summary.csv",
+                mime="text/csv"
+            )
+
+        except Exception as e:
+            st.error(f"❌ Error processing file: {e}")
+    else:
+        st.info("👆 Upload an Excel/CSV/TXT file to start generating pipetting commands.")