Create app.py

app.py

@@ -0,0 +1,167 @@
+import streamlit as st
+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
+
+# Configure the web page styling
+st.set_page_config(page_title="The Genomic Oracle", page_icon="🧬", layout="wide")
+
+# ==========================================
+# 1. LOAD AI MODELS (CACHED IN RAM)
+# ==========================================
+@st.cache_resource(show_spinner="Booting up the Oracle Network...")
+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
+
+device, tokenizer, clf_coding, model_base, model_promoter = load_oracle_brains()
+
+# ==========================================
+# 2. BIOINFORMATICS PIPELINE
+# ==========================================
+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. STREAMLIT USER INTERFACE
+# ==========================================
+st.title("🧬 The Genomic Oracle")
+st.markdown("### A Deep Learning Cascade for DNA Sequence Classification")
+st.info("💡 **Tip:** Sequences longer than 50 base pairs yield significantly more accurate biological predictions.")
+
+user_input = st.text_area("Enter DNA Sequence:", height=150)
+
+run_mapping = st.checkbox("Query NCBI BLAST for spatial mapping (Takes 1-3 minutes)")
+
+if st.button("Initialize Deep Scan", type="primary"):
+    if len(user_input) < 10:
+        st.error("Sequence too short! Please provide at least 10 base pairs.")
+    else:
+        with st.spinner("Analyzing spatial attention tensors..."):
+            label, conf, raw_scores = analyze_sequence(user_input)
+            
+        st.success("Analysis Complete!")
+        
+        col1, col2 = st.columns(2)
+        with col1:
+            st.subheader("Classification")
+            st.write(f"**{label}**")
+            st.write(f"**Confidence:** {conf:.2%}")
+            
+        with col2:
+            st.subheader("Internal Statistics")
+            st.write(f"- Level 1 (Coding): {raw_scores['Level 1 (Gene)'] if isinstance(raw_scores['Level 1 (Gene)'], str) else f'{raw_scores['Level 1 (Gene)']:.2%}'}")
+            st.write(f"- Level 2 (Promoter): {raw_scores['Level 2 (Promoter)'] if isinstance(raw_scores['Level 2 (Promoter)'], str) else f'{raw_scores['Level 2 (Promoter)']:.2%}'}")
+            st.write(f"- Level 3 (Intergenic): {raw_scores['Level 3 (Intergenic)'] if isinstance(raw_scores['Level 3 (Intergenic)'], str) else f'{raw_scores['Level 3 (Intergenic)']:.2%}'}")
+
+        st.divider()
+
+        if run_mapping:
+            if "Junk" in label:
+                st.warning("Spatial mapping bypassed. Sequence classified as non-functional noise.")
+            else:
+                with st.spinner("Querying NCBI and Ensembl databases..."):
+                    context_type = "CODING" if "Coding" in label else "PROMOTER"
+                    context = get_genomic_context(user_input, context_type)
+                    
+                    st.subheader("Final Mapping Report")
+                    if "error" in context:
+                        st.error(context['error'])
+                    elif "location" in context:
+                        st.write(f"**Location:** {context['location']}")
+                        st.write(f"**Strand:** {context['strand']}")
+                        st.write(f"**Coordinates:** {context['start']:,} - {context['end']:,}")
+                        st.write(f"**Notes:** {context['metadata']}")
+                    else:
+                        st.error("Could not map sequence.")