| 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 |
|
|
| |
| st.set_page_config(page_title="The Genomic Oracle", page_icon="🧬", layout="wide") |
|
|
| |
| |
| |
| @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() |
|
|
| |
| |
| |
| 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} |
|
|
| |
| |
| |
| 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.") |
