File size: 8,131 Bytes
e9d9f3f 8e880a1 e9d9f3f a19b785 e9d9f3f | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 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.") |