Spaces:

alrobles
/

ecointeraction-extractor

Sleeping

feat: upgrade to lightweight System 1 extractor with compact 2.1MB SQLite backbone

0d7be17 verified 2 days ago

10.9 kB

import re
import json
import time
from typing import Tuple, List, Dict, Any
import gradio as gr

from compact_resolver import CompactTaxonomicResolver, CompactTaxon

# Initialize Compact Backbone Resolver (2.1 MB SQLite, sub-millisecond lookup)
resolver = CompactTaxonomicResolver()

INTERACTION_RO_MAP = {
    "pollinates": {"ro_id": "RO:0002455", "label": "pollinates", "valence": "mutualistic"},
    "disperses_seeds_of": {"ro_id": "RO:0002456", "label": "disperses seeds of", "valence": "mutualistic"},
    "symbiont_of": {"ro_id": "RO:0002440", "label": "symbiont of", "valence": "mutualistic"},
    "mycorrhizal_partner_of": {"ro_id": "RO:0002442", "label": "mycorrhizal partner of", "valence": "mutualistic"},
    "preys_on": {"ro_id": "RO:0002439", "label": "preys on", "valence": "antagonistic"},
    "herbivore_of": {"ro_id": "RO:0002470", "label": "herbivore of", "valence": "antagonistic"},
    "parasite_of": {"ro_id": "RO:0002444", "label": "parasite of", "valence": "antagonistic"},
    "parasitoid_of": {"ro_id": "RO:0002208", "label": "parasitoid of", "valence": "antagonistic"},
    "pathogen_of": {"ro_id": "RO:0002556", "label": "pathogen of", "valence": "antagonistic"},
    "vector_of": {"ro_id": "RO:0002459", "label": "vector of", "valence": "antagonistic"},
    "competes_with": {"ro_id": "RO:0002441", "label": "competes with", "valence": "antagonistic"},
    "allelopathic_to": {"ro_id": "RO:0002555", "label": "allelopathic to", "valence": "antagonistic"},
    "epibiont_on": {"ro_id": "RO:0002553", "label": "epibiont on", "valence": "commensal_facilitative"},
    "phoretic_on": {"ro_id": "RO:0002437", "label": "phoretic on", "valence": "commensal_facilitative"},
    "associated_with": {"ro_id": "RO:0002437", "label": "associated with", "valence": "unspecified"}
}

BINOMIAL_REGEX = re.compile(r"\b([A-Z][a-z]{2,20}\s+[a-z]{2,25}(?:\s+subsp\.\s+[a-z]{2,20}|\s+var\.\s+[a-z]{2,20})?)\b")


def extract_candidate_taxa(text: str) -> List[str]:
    """Finds candidate Linnaean binomials in text."""
    matches = BINOMIAL_REGEX.findall(text)
    # Deduplicate preserving order
    seen = set()
    result = []
    for m in matches:
        clean = m.strip()
        if clean.lower() not in seen:
            seen.add(clean.lower())
            result.append(clean)
    return result


def extract_and_ground(passage: str, source_taxon: str = "", target_taxon: str = "") -> Tuple[str, str, str]:
    """Extracts ecological interaction and grounds entities against the compact backbone."""
    t0 = time.perf_counter()
    p_lower = passage.lower()

    # Auto-detect taxa candidates if not specified
    candidates = extract_candidate_taxa(passage)
    if not source_taxon and len(candidates) >= 1:
        source_taxon = candidates[0]
    if not target_taxon and len(candidates) >= 2:
        target_taxon = candidates[1]

    # Resolve taxa via compact SQLite backbone (<50 µs each)
    src_res = resolver.resolve(source_taxon) if source_taxon else None
    tgt_res = resolver.resolve(target_taxon) if target_taxon else None

    # Interaction cue detection & RO semantic mapping
    detected_type = "associated_with"
    detected_valence = "unspecified"
    ro_id = "RO:0002437"

    if any(w in p_lower for w in ["pollinat", "floral visit", "pollen transfer", "nectar rob"]):
        detected_type = "pollinates"
        detected_valence = "mutualistic"
        ro_id = "RO:0002455"
    elif any(w in p_lower for w in ["seed dispers", "frugivor", "endozoochor", "dispersal of seed"]):
        detected_type = "disperses_seeds_of"
        detected_valence = "mutualistic"
        ro_id = "RO:0002456"
    elif any(w in p_lower for w in ["mycorrhiz", "arbuscular", "root nodul", "ectomycorrhiz"]):
        detected_type = "mycorrhizal_partner_of"
        detected_valence = "mutualistic"
        ro_id = "RO:0002442"
    elif any(w in p_lower for w in ["prey", "predat", "gut content", "kill rate", "diet of", "consumed"]):
        detected_type = "preys_on"
        detected_valence = "antagonistic"
        ro_id = "RO:0002439"
    elif any(w in p_lower for w in ["herbivor", "defoliat", "grazing on", "feeding on leaves", "caterpillar on"]):
        detected_type = "herbivore_of"
        detected_valence = "antagonistic"
        ro_id = "RO:0002470"
    elif any(w in p_lower for w in ["parasitoid", "host pupa", "host larva", "oviposition in"]):
        detected_type = "parasitoid_of"
        detected_valence = "antagonistic"
        ro_id = "RO:0002208"
    elif any(w in p_lower for w in ["parasit", "infest", "ectoparasit", "endoparasit", "tick on", "flea on"]):
        detected_type = "parasite_of"
        detected_valence = "antagonistic"
        ro_id = "RO:0002444"
    elif any(w in p_lower for w in ["vector", "transmitted by", "arbovirus vector"]):
        detected_type = "vector_of"
        detected_valence = "antagonistic"
        ro_id = "RO:0002459"
    elif any(w in p_lower for w in ["allelopath", "competiti", "growth inhibition"]):
        detected_type = "allelopathic_to"
        detected_valence = "antagonistic"
        ro_id = "RO:0002555"

    # Evidence span locator
    sentences = re.split(r"(?<=[.!?])\s+", passage)
    evidence_span = passage.strip()[:240]
    for s in sentences:
        s_low = s.lower()
        if source_taxon and source_taxon.lower() in s_low and target_taxon and target_taxon.lower() in s_low:
            evidence_span = s.strip()
            break
        elif any(cue in s_low for cue in ["prey", "pollinat", "parasit", "host", "dispers", "symbio", "feed"]):
            evidence_span = s.strip()
            break

    elapsed_ms = (time.perf_counter() - t0) * 1000.0

    # Build output payload
    structured_edge = {
        "source_taxon": {
            "verbatim": source_taxon,
            "canonical": src_res.canonical_name if src_res else source_taxon,
            "gbif_key": src_res.gbif_key if src_res else None,
            "ncbi_taxid": src_res.ncbi_taxid if src_res else None,
            "kingdom": src_res.kingdom if src_res else "Unknown"
        },
        "target_taxon": {
            "verbatim": target_taxon,
            "canonical": tgt_res.canonical_name if tgt_res else target_taxon,
            "gbif_key": tgt_res.gbif_key if tgt_res else None,
            "ncbi_taxid": tgt_res.ncbi_taxid if tgt_res else None,
            "kingdom": tgt_res.kingdom if tgt_res else "Unknown"
        },
        "interaction": {
            "type": detected_type,
            "valence": detected_valence,
            "ro_id": ro_id,
            "standard": "OBO Relations Ontology / IS-1.0"
        },
        "evidence": {
            "verbatim_span": evidence_span,
            "confidence": 0.95 if (detected_type != "associated_with" and src_res and tgt_res) else 0.70
        },
        "metrics": {
            "inference_time_ms": round(elapsed_ms, 2),
            "backbone_engine": "SQLite B-Tree (2.1 MB / 23.9k taxa)"
        }
    }

    # Format Markdown Table for Grounding
    def format_row(role: str, name: str, res: CompactTaxon) -> str:
        if res:
            gbif_link = f"[{res.gbif_key}](https://www.gbif.org/species/{res.gbif_key})" if res.gbif_key else "—"
            ncbi_link = f"[{res.ncbi_taxid}](https://www.ncbi.nlm.nih.gov/Taxonomy/Browser/wwwtax.cgi?id={res.ncbi_taxid})" if res.ncbi_taxid else "—"
            return f"| **{role}** | *{res.canonical_name}* | `{res.kingdom}` | {ncbi_link} | {gbif_link} | ✅ Grounded |"
        else:
            return f"| **{role}** | *{name or 'N/A'}* | `Unknown` | — | — | ⚠️ Unresolved |"

    grounding_table = (
        "| Role | Canonical Taxon | Kingdom | NCBI TaxID | GBIF UsageKey | Status |\n"
        "| :--- | :--- | :--- | :--- | :--- | :--- |\n"
        f"{format_row('Source Taxon', source_taxon, src_res)}\n"
        f"{format_row('Target Taxon', target_taxon, tgt_res)}\n"
    )

    metrics_md = f"⚡ **Latency:** `{elapsed_ms:.2f} ms` | 🎯 **Confidence:** `{structured_edge['evidence']['confidence']:.2f}` | 🏛️ **Standard:** `OBO {ro_id}`"

    return json.dumps(structured_edge, indent=2), grounding_table, metrics_md


CUSTOM_CSS = """
.metric-card {
    background-color: #f8fafc;
    border-radius: 8px;
    padding: 12px 16px;
    margin-bottom: 12px;
}
"""

with gr.Blocks(title="EcoInteraction Extractor", css=CUSTOM_CSS) as demo:
    gr.Markdown("# 🌿 EcoInteraction Extractor")
    gr.Markdown(
        "Sub-millisecond System 1 Biological Interaction Extractor with Compact Linnaean Backbone Grounding. "
        "Standardized to **IS-1.0** and **OBO Relations Ontology**."
    )

    with gr.Row():
        with gr.Column(scale=5):
            passage_in = gr.Textbox(
                label="Scientific Literature Passage",
                value="In subalpine meadows, Bombus bimaculatus visits flowers of Pedicularis groenlandica to extract nectar and facilitate pollen transfer.",
                lines=5,
                placeholder="Paste an excerpt, abstract sentence, or observational note..."
            )
            with gr.Row():
                src_in = gr.Textbox(label="Source Taxon (Auto-detected if blank)", placeholder="e.g. Bombus bimaculatus")
                tgt_in = gr.Textbox(label="Target Taxon (Auto-detected if blank)", placeholder="e.g. Pedicularis groenlandica")
            btn = gr.Button("Extract & Ground Interaction", variant="primary")

            gr.Examples(
                examples=[
                    [
                        "Adult females of Ixodes ricinus were collected while feeding on Apodemus flavicollis in deciduous temperate woodlands.",
                        "Ixodes ricinus",
                        "Apodemus flavicollis"
                    ],
                    [
                        "In subalpine meadows, Bombus bimaculatus visits flowers of Pedicularis groenlandica to extract nectar and facilitate pollen transfer.",
                        "Bombus bimaculatus",
                        "Pedicularis groenlandica"
                    ],
                    [
                        "Falco peregrinus was observed repeatedly preying upon Columba livia within urban foraging territories.",
                        "Falco peregrinus",
                        "Columba livia"
                    ]
                ],
                inputs=[passage_in, src_in, tgt_in]
            )

        with gr.Column(scale=5):
            metrics_display = gr.Markdown("⚡ **Ready for extraction**")
            gr.Markdown("### 🧬 Taxonomic Grounding")
            table_out = gr.Markdown()
            gr.Markdown("### 📦 Structured IS-1.0 Interaction Edge")
            json_out = gr.Code(label="Normalized JSON Output", language="json")

    btn.click(
        extract_and_ground,
        inputs=[passage_in, src_in, tgt_in],
        outputs=[json_out, table_out, metrics_display]
    )

if __name__ == "__main__":
    demo.launch()

	import re
	import json
	import time
	from typing import Tuple, List, Dict, Any
	import gradio as gr

	from compact_resolver import CompactTaxonomicResolver, CompactTaxon

	# Initialize Compact Backbone Resolver (2.1 MB SQLite, sub-millisecond lookup)
	resolver = CompactTaxonomicResolver()

	INTERACTION_RO_MAP = {
	"pollinates": {"ro_id": "RO:0002455", "label": "pollinates", "valence": "mutualistic"},
	"disperses_seeds_of": {"ro_id": "RO:0002456", "label": "disperses seeds of", "valence": "mutualistic"},
	"symbiont_of": {"ro_id": "RO:0002440", "label": "symbiont of", "valence": "mutualistic"},
	"mycorrhizal_partner_of": {"ro_id": "RO:0002442", "label": "mycorrhizal partner of", "valence": "mutualistic"},
	"preys_on": {"ro_id": "RO:0002439", "label": "preys on", "valence": "antagonistic"},
	"herbivore_of": {"ro_id": "RO:0002470", "label": "herbivore of", "valence": "antagonistic"},
	"parasite_of": {"ro_id": "RO:0002444", "label": "parasite of", "valence": "antagonistic"},
	"parasitoid_of": {"ro_id": "RO:0002208", "label": "parasitoid of", "valence": "antagonistic"},
	"pathogen_of": {"ro_id": "RO:0002556", "label": "pathogen of", "valence": "antagonistic"},
	"vector_of": {"ro_id": "RO:0002459", "label": "vector of", "valence": "antagonistic"},
	"competes_with": {"ro_id": "RO:0002441", "label": "competes with", "valence": "antagonistic"},
	"allelopathic_to": {"ro_id": "RO:0002555", "label": "allelopathic to", "valence": "antagonistic"},
	"epibiont_on": {"ro_id": "RO:0002553", "label": "epibiont on", "valence": "commensal_facilitative"},
	"phoretic_on": {"ro_id": "RO:0002437", "label": "phoretic on", "valence": "commensal_facilitative"},
	"associated_with": {"ro_id": "RO:0002437", "label": "associated with", "valence": "unspecified"}
	}

	BINOMIAL_REGEX = re.compile(r"\b([A-Z][a-z]{2,20}\s+[a-z]{2,25}(?:\s+subsp\.\s+[a-z]{2,20}\|\s+var\.\s+[a-z]{2,20})?)\b")


	def extract_candidate_taxa(text: str) -> List[str]:
	"""Finds candidate Linnaean binomials in text."""
	matches = BINOMIAL_REGEX.findall(text)
	# Deduplicate preserving order
	seen = set()
	result = []
	for m in matches:
	clean = m.strip()
	if clean.lower() not in seen:
	seen.add(clean.lower())
	result.append(clean)
	return result


	def extract_and_ground(passage: str, source_taxon: str = "", target_taxon: str = "") -> Tuple[str, str, str]:
	"""Extracts ecological interaction and grounds entities against the compact backbone."""
	t0 = time.perf_counter()
	p_lower = passage.lower()

	# Auto-detect taxa candidates if not specified
	candidates = extract_candidate_taxa(passage)
	if not source_taxon and len(candidates) >= 1:
	source_taxon = candidates[0]
	if not target_taxon and len(candidates) >= 2:
	target_taxon = candidates[1]

	# Resolve taxa via compact SQLite backbone (<50 µs each)
	src_res = resolver.resolve(source_taxon) if source_taxon else None
	tgt_res = resolver.resolve(target_taxon) if target_taxon else None

	# Interaction cue detection & RO semantic mapping
	detected_type = "associated_with"
	detected_valence = "unspecified"
	ro_id = "RO:0002437"

	if any(w in p_lower for w in ["pollinat", "floral visit", "pollen transfer", "nectar rob"]):
	detected_type = "pollinates"
	detected_valence = "mutualistic"
	ro_id = "RO:0002455"
	elif any(w in p_lower for w in ["seed dispers", "frugivor", "endozoochor", "dispersal of seed"]):
	detected_type = "disperses_seeds_of"
	detected_valence = "mutualistic"
	ro_id = "RO:0002456"
	elif any(w in p_lower for w in ["mycorrhiz", "arbuscular", "root nodul", "ectomycorrhiz"]):
	detected_type = "mycorrhizal_partner_of"
	detected_valence = "mutualistic"
	ro_id = "RO:0002442"
	elif any(w in p_lower for w in ["prey", "predat", "gut content", "kill rate", "diet of", "consumed"]):
	detected_type = "preys_on"
	detected_valence = "antagonistic"
	ro_id = "RO:0002439"
	elif any(w in p_lower for w in ["herbivor", "defoliat", "grazing on", "feeding on leaves", "caterpillar on"]):
	detected_type = "herbivore_of"
	detected_valence = "antagonistic"
	ro_id = "RO:0002470"
	elif any(w in p_lower for w in ["parasitoid", "host pupa", "host larva", "oviposition in"]):
	detected_type = "parasitoid_of"
	detected_valence = "antagonistic"
	ro_id = "RO:0002208"
	elif any(w in p_lower for w in ["parasit", "infest", "ectoparasit", "endoparasit", "tick on", "flea on"]):
	detected_type = "parasite_of"
	detected_valence = "antagonistic"
	ro_id = "RO:0002444"
	elif any(w in p_lower for w in ["vector", "transmitted by", "arbovirus vector"]):
	detected_type = "vector_of"
	detected_valence = "antagonistic"
	ro_id = "RO:0002459"
	elif any(w in p_lower for w in ["allelopath", "competiti", "growth inhibition"]):
	detected_type = "allelopathic_to"
	detected_valence = "antagonistic"
	ro_id = "RO:0002555"

	# Evidence span locator
	sentences = re.split(r"(?<=[.!?])\s+", passage)
	evidence_span = passage.strip()[:240]
	for s in sentences:
	s_low = s.lower()
	if source_taxon and source_taxon.lower() in s_low and target_taxon and target_taxon.lower() in s_low:
	evidence_span = s.strip()
	break
	elif any(cue in s_low for cue in ["prey", "pollinat", "parasit", "host", "dispers", "symbio", "feed"]):
	evidence_span = s.strip()
	break

	elapsed_ms = (time.perf_counter() - t0) * 1000.0

	# Build output payload
	structured_edge = {
	"source_taxon": {
	"verbatim": source_taxon,
	"canonical": src_res.canonical_name if src_res else source_taxon,
	"gbif_key": src_res.gbif_key if src_res else None,
	"ncbi_taxid": src_res.ncbi_taxid if src_res else None,
	"kingdom": src_res.kingdom if src_res else "Unknown"
	},
	"target_taxon": {
	"verbatim": target_taxon,
	"canonical": tgt_res.canonical_name if tgt_res else target_taxon,
	"gbif_key": tgt_res.gbif_key if tgt_res else None,
	"ncbi_taxid": tgt_res.ncbi_taxid if tgt_res else None,
	"kingdom": tgt_res.kingdom if tgt_res else "Unknown"
	},
	"interaction": {
	"type": detected_type,
	"valence": detected_valence,
	"ro_id": ro_id,
	"standard": "OBO Relations Ontology / IS-1.0"
	},
	"evidence": {
	"verbatim_span": evidence_span,
	"confidence": 0.95 if (detected_type != "associated_with" and src_res and tgt_res) else 0.70
	},
	"metrics": {
	"inference_time_ms": round(elapsed_ms, 2),
	"backbone_engine": "SQLite B-Tree (2.1 MB / 23.9k taxa)"
	}
	}

	# Format Markdown Table for Grounding
	def format_row(role: str, name: str, res: CompactTaxon) -> str:
	if res:
	gbif_link = f"[{res.gbif_key}](https://www.gbif.org/species/{res.gbif_key})" if res.gbif_key else "—"
	ncbi_link = f"[{res.ncbi_taxid}](https://www.ncbi.nlm.nih.gov/Taxonomy/Browser/wwwtax.cgi?id={res.ncbi_taxid})" if res.ncbi_taxid else "—"
	return f"\| {role} \| {res.canonical_name} \| `{res.kingdom}` \| {ncbi_link} \| {gbif_link} \| ✅ Grounded \|"
	else:
	return f"\| {role} \| {name or 'N/A'} \| `Unknown` \| — \| — \| ⚠️ Unresolved \|"

	grounding_table = (
	"\| Role \| Canonical Taxon \| Kingdom \| NCBI TaxID \| GBIF UsageKey \| Status \|\n"
	"\| :--- \| :--- \| :--- \| :--- \| :--- \| :--- \|\n"
	f"{format_row('Source Taxon', source_taxon, src_res)}\n"
	f"{format_row('Target Taxon', target_taxon, tgt_res)}\n"
	)

	metrics_md = f"⚡ Latency: `{elapsed_ms:.2f} ms` \| 🎯 Confidence: `{structured_edge['evidence']['confidence']:.2f}` \| 🏛️ Standard: `OBO {ro_id}`"

	return json.dumps(structured_edge, indent=2), grounding_table, metrics_md


	CUSTOM_CSS = """
	.metric-card {
	background-color: #f8fafc;
	border-radius: 8px;
	padding: 12px 16px;
	margin-bottom: 12px;
	}
	"""

	with gr.Blocks(title="EcoInteraction Extractor", css=CUSTOM_CSS) as demo:
	gr.Markdown("# 🌿 EcoInteraction Extractor")
	gr.Markdown(
	"Sub-millisecond System 1 Biological Interaction Extractor with Compact Linnaean Backbone Grounding. "
	"Standardized to IS-1.0 and OBO Relations Ontology."
	)

	with gr.Row():
	with gr.Column(scale=5):
	passage_in = gr.Textbox(
	label="Scientific Literature Passage",
	value="In subalpine meadows, Bombus bimaculatus visits flowers of Pedicularis groenlandica to extract nectar and facilitate pollen transfer.",
	lines=5,
	placeholder="Paste an excerpt, abstract sentence, or observational note..."
	)
	with gr.Row():
	src_in = gr.Textbox(label="Source Taxon (Auto-detected if blank)", placeholder="e.g. Bombus bimaculatus")
	tgt_in = gr.Textbox(label="Target Taxon (Auto-detected if blank)", placeholder="e.g. Pedicularis groenlandica")
	btn = gr.Button("Extract & Ground Interaction", variant="primary")

	gr.Examples(
	examples=[
	[
	"Adult females of Ixodes ricinus were collected while feeding on Apodemus flavicollis in deciduous temperate woodlands.",
	"Ixodes ricinus",
	"Apodemus flavicollis"
	],
	[
	"In subalpine meadows, Bombus bimaculatus visits flowers of Pedicularis groenlandica to extract nectar and facilitate pollen transfer.",
	"Bombus bimaculatus",
	"Pedicularis groenlandica"
	],
	[
	"Falco peregrinus was observed repeatedly preying upon Columba livia within urban foraging territories.",
	"Falco peregrinus",
	"Columba livia"
	]
	],
	inputs=[passage_in, src_in, tgt_in]
	)

	with gr.Column(scale=5):
	metrics_display = gr.Markdown("⚡ Ready for extraction")
	gr.Markdown("### 🧬 Taxonomic Grounding")
	table_out = gr.Markdown()
	gr.Markdown("### 📦 Structured IS-1.0 Interaction Edge")
	json_out = gr.Code(label="Normalized JSON Output", language="json")

	btn.click(
	extract_and_ground,
	inputs=[passage_in, src_in, tgt_in],
	outputs=[json_out, table_out, metrics_display]
	)

	if __name__ == "__main__":
	demo.launch()