Create app.py

This is a streamlit to gradio direct conversion from Geonomic's Genomic Classifier Capstone.

app.py

@@ -0,0 +1,239 @@
+import gradio as gr
+import os
+import re
+import requests
+import torch
+import joblib
+import numpy as np
+import torch.nn.functional as F
+from Bio.Blast import NCBIWWW
+from Bio.Blast import NCBIXML
+from transformers import AutoTokenizer, AutoModelForSequenceClassification, AutoModel
+
+# ==========================================
+# 1. LOAD AI MODELS (CACHED IN RAM)
+# ==========================================
+@torch.no_grad()
+def load_oracle_brains():
+    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
+
+    clf = joblib.load("coding_classifier_universal.joblib")
+
+    tokenizer = AutoTokenizer.from_pretrained("DNABERT_Local", trust_remote_code=True)
+    model_base = AutoModel.from_pretrained("DNABERT_Local", trust_remote_code=True).to(device)
+    model_base.eval()
+
+    model_promoter = AutoModelForSequenceClassification.from_pretrained("llm_promoter_classifier_v2", trust_remote_code=True).to(device)
+    model_promoter.eval()
+
+    return device, tokenizer, clf, model_base, model_promoter
+
+# Initialize models once (Gradio handles caching across sessions)
+device, tokenizer, clf_coding, model_base, model_promoter = load_oracle_brains()
+
+# ==========================================
+# 2. BIOINFORMATICS PIPELINE (unchanged logic)
+# ==========================================
+def analyze_sequence(dna_sequence):
+    clean_seq = "".join(dna_sequence.split()).upper()
+    inputs = tokenizer([clean_seq], return_tensors="pt", max_length=300, truncation=True, padding=True)
+    inputs = {k: v.to(device) for k, v in inputs.items()}
+
+    with torch.no_grad():
+        out_base = model_base(**inputs)
+        mask = inputs["attention_mask"].unsqueeze(-1)
+        embedding = (out_base[0] * mask).sum(dim=1) / mask.sum(dim=1).clamp(min=1)
+        vector = embedding.float().cpu().numpy()
+
+    p_coding = clf_coding.predict_proba(vector)[0][1]
+
+    if p_coding >= 0.60:
+        return "GENE (Protein-Coding DNA)", p_coding, {
+            "Level 1 (Gene)": p_coding,
+            "Level 2 (Promoter)": "Skipped",
+            "Level 3 (Intergenic)": "Skipped"
+        }
+
+    with torch.no_grad():
+        outputs_promoter = model_promoter(**inputs)
+        logits = outputs_promoter.logits
+        probs = F.softmax(logits, dim=-1)
+        p_promoter = probs[0][0].item()
+
+    if p_promoter >= 0.50:
+        return "PROMOTER (Regulatory DNA)", p_promoter, {
+            "Level 1 (Gene)": p_coding,
+            "Level 2 (Promoter)": p_promoter,
+            "Level 3 (Intergenic)": 1 - p_promoter
+        }
+
+    return "INTERGENIC (Non-Functional Junk DNA)", 1 - p_promoter, {
+        "Level 1 (Gene)": p_coding,
+        "Level 2 (Promoter)": p_promoter,
+        "Level 3 (Intergenic)": 1 - p_promoter
+    }
+
+def get_genomic_context(sequence, feature_type):
+    try:
+        result_handle = NCBIWWW.qblast("blastn", "nt", sequence,
+                                       entrez_query="Homo sapiens[Organism] AND biomol_genomic[PROP]",
+                                       hitlist_size=1)
+        blast_record = NCBIXML.read(result_handle)
+    except Exception as e:
+        return {"error": f"BLAST Connection Error: {e}"}
+
+    if not blast_record.alignments:
+        return {"error": "No human genome match found for this sequence."}
+
+    alignment = blast_record.alignments[0]
+    hsp = alignment.hsps[0]
+    accession = alignment.accession
+    full_title = alignment.title.split('|')[-1].strip()
+
+    chrom_match = re.search(r"chromosome\s([0-9XYMT]+)", alignment.title, re.IGNORECASE)
+    chrom = chrom_match.group(1) if chrom_match else None
+    location_string = f"Chromosome {chrom}" if chrom else f"Accession {accession} | {full_title}"
+
+    start = hsp.sbjct_start
+    end = hsp.sbjct_end
+    is_forward = (start < end)
+    strand_txt = "Forward (+)" if is_forward else "Reverse (-)"
+
+    if chrom is None:
+        return {"location": location_string, "start": start, "end": end, "strand": strand_txt,
+                "metadata": "BLAST returned a localized record without a chromosome. Ensembl mapping skipped."}
+
+    search_start = min(start, end) if feature_type == "CODING" else (end if is_forward else max(1, end - 15000))
+    search_end = max(start, end) if feature_type == "CODING" else (end + 15000 if is_forward else end)
+
+    try:
+        response = requests.get(
+            f"https://rest.ensembl.org/overlap/region/human/{chrom}:{search_start}-{search_end}?feature=gene",
+            headers={"Content-Type": "application/json"}
+        )
+        response.raise_for_status()
+        genes = response.json()
+    except Exception as e:
+        return {"location": location_string, "start": start, "end": end, "strand": strand_txt,
+                "metadata": f"Ensembl mapping unavailable: {e}"}
+
+    if not genes:
+        gene_desc = "No annotated genes found in this specific region."
+    else:
+        if feature_type == "PROMOTER":
+            genes.sort(key=lambda x: min(abs(x['start'] - end), abs(x['end'] - end)))
+        top_gene = genes[0]
+        name = top_gene.get('external_name', 'Unknown')
+        biotype = top_gene.get('biotype', 'Unknown').replace('_', ' ').title()
+        desc = top_gene.get('description', 'No description available.').split(' [')[0]
+        gene_desc = (f"Matches Gene: {name} | Type: {biotype} | Function: {desc}"
+                     if feature_type == "CODING"
+                     else f"Regulates Downstream Gene: {name} | Type: {biotype} | Function: {desc}")
+
+    return {"location": location_string, "start": start, "end": end, "strand": strand_txt, "metadata": gene_desc}
+
+
+# ==========================================
+# 3. GRADIO INTERFACE
+# ==========================================
+def gradio_inference(dna_sequence, run_mapping):
+    if len(dna_sequence.strip()) < 10:
+        return ("❌ Sequence too short!", "",
+                {"Level 1 (Gene)": "N/A", "Level 2 (Promoter)": "N/A", "Level 3 (Intergenic)": "N/A"}, "",
+                "⚠️ Please enter at least 10 base pairs.")
+
+    # Primary classification
+    label, conf, raw_scores = analyze_sequence(dna_sequence)
+
+    # Format internal statistics
+    def fmt(v):
+        if isinstance(v, str): return v
+        return f"{v:.2%}"
+
+    stats_lines = [
+        f"- Level 1 (Coding): {fmt(raw_scores['Level 1 (Gene)'])}",
+        f"- Level 2 (Promoter): {fmt(raw_scores['Level 2 (Promoter)'])}",
+        f"- Level 3 (Intergenic): {fmt(raw_scores['Level 3 (Intergenic)'])}"
+    ]
+
+    # Context mapping (optional, slow)
+    context_output = ""
+    if run_mapping:
+        if "Junk" in label or "INTERGENIC" in label:
+            context_output = "⚠️ Spatial mapping skipped — sequence classified as non-functional."
+        else:
+            context_type = "CODING" if "Coding" in label else "PROMOTER"
+            try:
+                # Run BLAST + Ensembl lookup (blocking)
+                import time; start_time = time.time()
+                context = get_genomic_context(dna_sequence, context_type)
+                elapsed = int(time.time() - start_time)
+
+                if "error" in context:
+                    context_output = f"❌ Mapping failed: {context['error']}"
+                elif "location" in context:
+                    context_lines = [
+                        f"📍 **Location:** {context['location']}",
+                        f"🧬 **Strand:** {context['strand']}",
+                        f"📏 **Coordinates:** {context['start']:,} – {context['end']:,}",
+                        f"ℹ️ **Notes:** {context['metadata']}"
+                    ]
+                    context_output = "\n".join(context_lines)
+                else:
+                    context_output = "⚠️ Could not map sequence."
+            except Exception as e:
+                context_output = f"❌ Mapping error: {str(e)}"
+    else:
+        context_output = "⏸️ Spatial mapping skipped (disable checkbox to run)."
+
+    summary = (
+        f"✅ Classification complete in ~1–2 sec.\n"
+        f"\n"
+        f"🎯 **Result:** {label}\n"
+        f"📊 **Confidence:** {conf:.2%}"
+    )
+
+    return summary, "\n".join(stats_lines), raw_scores, context_output, "⏳ Mapping query in progress..." if run_mapping else ""
+
+with gr.Blocks(theme=gr.themes.Soft(), title="🧬 The Genomic Oracle") as demo:
+    gr.Markdown("# 🧬 The Genomic Oracle")
+    gr.Markdown("### A Deep Learning Cascade for DNA Sequence Classification")
+    gr.Info("💡 Tip: Sequences longer than 50 bp yield more accurate predictions.")
+
+    with gr.Row():
+        with gr.Column(scale=1):
+            dna_input = gr.Textbox(
+                label="Enter DNA Sequence",
+                placeholder="e.g., ATGCGATCGATCGATCG...",
+                lines=6,
+                elem_id="dna_input"
+            )
+            run_mapping_cb = gr.Checkbox(
+                value=False,
+                label="Query NCBI BLAST for spatial mapping (Takes 1–3 minutes)"
+            )
+            submit_btn = gr.Button("🚀 Initialize Deep Scan", variant="primary")
+
+        with gr.Column(scale=2):
+            output_summary = gr.Textbox(label="✅ Classification Summary", lines=8)
+            stats_panel = gr.Textbox(label="📊 Internal Statistics", lines=6, show_copy_button=True)
+
+            mapping_section = gr.Accordion("📍 Genomic Context (BLAST/Ensembl)", open=False)
+            with mapping_section:
+                context_output = gr.Textbox(
+                    label="Mapping Results",
+                    lines=5,
+                    placeholder="Results will appear here..."
+                )
+
+    # Live feedback
+    info_box = gr.Markdown("", elem_id="info_box")
+
+    submit_btn.click(
+        fn=lambda seq, map: (
+            *gradio_inference(seq, map)[:4],
+            gr.Textbox(visible=True) if map else gr.Textbox(visible=False)
+        ),
+        inputs=[dna_input, run_mapping_cb],
+        outputs=[output_summary, stats_panel, gr.State(), context_output]
+    )