"""Read-only taxonomy navigation, independent of the annotation lookup index.""" from collections import Counter from contextlib import closing import math from html import escape import json from pathlib import Path import sqlite3 import gradio as gr ROOT = Path(__file__).resolve().parent DATABASE = ROOT / "data/taxonomy.sqlite" COMMON_NAMES = ROOT / "data/common_names.json" ICONS = ROOT / "data/icons.json" ICON_DIR = ROOT / "data/icons" # Gradio serves whitelisted files from this route; see build_taxonomy_tab. ICON_ROUTE = "/gradio_api/file=" LICENSE_NAMES = {"https://creativecommons.org/publicdomain/zero/1.0/": "CC0 1.0", "https://creativecommons.org/licenses/by/4.0/": "CC BY 4.0", "https://creativecommons.org/licenses/by/3.0/": "CC BY 3.0", "https://creativecommons.org/publicdomain/mark/1.0/": "Public Domain Mark 1.0"} EUKARYOTA = 2759 EUK_TREE = """WITH RECURSIVE euk(taxid) AS ( SELECT 2759 UNION ALL SELECT t.taxid FROM taxa t JOIN euk e ON t.parent_id=e.taxid WHERE t.taxid != t.parent_id ) """ # Scientific names stay the primary label; English glosses are a secondary line. # See refresh_common_names.py for how the lookup is built. MAX_ENGLISH = 30 MAX_LABEL = 24 COLUMN_LIMIT = 6 MAX_DEPTH = 80 DEFAULT_PATH = "2759/33154" def encode_path(steps): """Serialize selected (taxid, offset) pairs; an offset marks an expanded "Other lineages" aggregate.""" return "/".join(f"{t}~{o}" if o else str(t) for t, o in steps) class Taxonomy: def __init__(self, path=DATABASE, common_names=COMMON_NAMES, icons=ICONS): self.path = Path(path) self.english, self.english_sources = {}, {} if Path(common_names).exists(): payload = json.loads(Path(common_names).read_text()) self.english = {int(t): name for t, name in payload["names"].items()} self.english_sources = payload.get("sources", {}) self.icons, self.icon_credits, self.icon_meta = {}, {}, {} if Path(icons).exists(): payload = json.loads(Path(icons).read_text()) self.icon_meta = {k: v for k, v in payload.items() if k not in ("images", "taxa")} self.icon_credits = payload.get("images", {}) self.icons = {int(t): image for t, image in payload.get("taxa", {}).items() if (ICON_DIR / f"{image}.svg").exists()} self._inherited_icons = {} with closing(self.connect()) as conn: self.metadata = json.loads(conn.execute("SELECT value FROM metadata").fetchone()[0]) self.eukaryote_ids = frozenset(r[0] for r in conn.execute(EUK_TREE + "SELECT taxid FROM euk")) self.eukaryotes = dict(conn.execute("SELECT * FROM taxa WHERE taxid=?", (EUKARYOTA,)).fetchone()) self.coverage = self.metadata.get("coverage") def connect(self): conn = sqlite3.connect(f"{self.path.resolve().as_uri()}?mode=ro", uri=True) conn.row_factory = sqlite3.Row return conn def search(self, query): query = str(query or "").strip() aliases = {"animals": "33208", "plants": "33090", "green plants": "33090", "eukaryotes": "2759"} query = aliases.get(query.casefold(), query) with closing(self.connect()) as conn: if query.lstrip("-").isdigit(): rows = conn.execute(EUK_TREE + "SELECT taxa.* FROM taxa JOIN euk USING(taxid) WHERE taxid=?", (int(query),)).fetchall() else: pattern = "%" + query.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") + "%" rows = conn.execute(EUK_TREE + "SELECT taxa.* FROM taxa JOIN euk USING(taxid) WHERE name LIKE ? ESCAPE '\\' ORDER BY total_count DESC LIMIT 100", (pattern,)).fetchall() return [(f"{r['name']} · {r['total_count']:,} assemblies · taxid {r['taxid']}", str(r['taxid'])) for r in rows] def icon_for(self, conn, taxid): """Nearest icon at or above this taxon, following the lineage to Eukaryota. Returns the image and whether it belongs to this taxon or was borrowed. """ if taxid in self.icons: return self.icons[taxid], False if taxid in self._inherited_icons: return self._inherited_icons[taxid], True walked, current = [], taxid for _ in range(MAX_DEPTH): row = conn.execute("SELECT parent_id FROM taxa WHERE taxid=?", (current,)).fetchone() if row is None or row['parent_id'] == current: break walked.append(current) current = row['parent_id'] if current in self.icons or current in self._inherited_icons: break found = self.icons.get(current) or self._inherited_icons.get(current, "") for step in walked: self._inherited_icons[step] = found return found, True def lineage(self, conn, taxid): rows = [dict(conn.execute("SELECT * FROM taxa WHERE taxid=?", (taxid,)).fetchone())] while rows[-1]['taxid'] != EUKARYOTA: rows.append(dict(conn.execute("SELECT * FROM taxa WHERE taxid=?", (rows[-1]['parent_id'],)).fetchone())) return rows[::-1] def taxon_path(self, taxid): """Path that expands every ancestor of a taxon, for search results and shortcuts.""" taxid = int(taxid) if taxid not in self.eukaryote_ids: raise ValueError("Choose a eukaryotic taxon from this snapshot.") with closing(self.connect()) as conn: return encode_path((r['taxid'], 0) for r in self.lineage(conn, taxid)) def view_taxon(self, taxid): return self.view(self.taxon_path(taxid)) def view(self, path=DEFAULT_PATH): """Render one column per expanded group. A path lists the selected node of each column.""" try: steps = [(int(t), int(o or 0)) for t, _, o in (part.partition("~") for part in str(path).split("/"))] except ValueError as exc: raise ValueError("Choose a eukaryotic taxon from this snapshot.") from exc if steps[0] != (EUKARYOTA, 0) or len(steps) > MAX_DEPTH or any( t not in self.eukaryote_ids or o < 0 for t, o in steps): raise ValueError("Choose a eukaryotic taxon from this snapshot.") with closing(self.connect()) as conn: def taxon(t): return dict(conn.execute("SELECT * FROM taxa WHERE taxid=?", (t,)).fetchone()) def node(row, offset=0, **extra): return dict(taxid=row['taxid'], offset=offset, name=row['name'], label=row['name'], english=self.english.get(row['taxid'], ""), **dict(zip(("icon", "borrowed_icon"), self.icon_for(conn, row['taxid']))), total=row['total_count'], covered=row.get('covered_total', 0), **extra) def remainder(rows, parent, start): return dict(taxid=parent, offset=start, name=f"Other lineages · {len(rows):,} groups (display aggregate)", label=f"Other lineages ({len(rows):,})", english="", icon="", borrowed_icon=False, total=sum(r['total_count'] for r in rows), covered=sum(r.get('covered_total', 0) for r in rows), aggregate=True) def column(step, selected): """Direct lineages of a selected node, keeping the next selection visible.""" parent, offset = step rows = [dict(r) for r in conn.execute("SELECT * FROM taxa WHERE parent_id=? AND taxid!=? ORDER BY total_count DESC, taxid", (parent, parent))] if offset and offset >= len(rows): raise ValueError("This group has no further lineages.") shown, rest = rows[offset:offset + COLUMN_LIMIT], rows[offset + COLUMN_LIMIT:] pinned = [r for r in rest if selected and not selected[1] and r['taxid'] == selected[0]] shown, rest = shown + pinned, [r for r in rest if r not in pinned] nodes = [node(r) for r in shown] if rest: nodes.append(remainder(rest, parent, offset + COLUMN_LIMIT)) row = taxon(parent) if not offset and row['direct_count'] and rows: nodes.append(dict(taxid=parent, offset=0, name="Assemblies assigned directly to this taxon", label="Direct assignments", english="", icon="", borrowed_icon=False, total=row['direct_count'], covered=row.get('covered_direct', 0), direct=True)) return nodes columns = [[node(taxon(EUKARYOTA))]] for depth, step in enumerate(steps[1:], 1): if step[1] and (step[0] != steps[depth - 1][0] or step[1] <= steps[depth - 1][1]) or \ not step[1] and taxon(step[0])['parent_id'] != steps[depth - 1][0]: raise ValueError("Choose a eukaryotic taxon from this snapshot.") columns.append(column(steps[depth - 1], step)) tail = column(steps[-1], None) if tail: columns.append(tail) for depth, nodes in enumerate(columns): for item in nodes: item['path'] = encode_path(steps[:depth] + [(item['taxid'], item['offset'])]) item['selected'] = depth < len(steps) and not item.get('direct') and \ (item['taxid'], item['offset']) == steps[depth] if depth < len(steps) and not any(item['selected'] for item in nodes): raise ValueError("Choose a eukaryotic taxon from this snapshot.") lineage = [taxon(t) for t, _ in steps] shortcuts = {t: encode_path((r['taxid'], 0) for r in self.lineage(conn, t)) for t in (33208, 33090, 4751)} return self.render(columns, steps, lineage, shortcuts, has_children=bool(tail)) def render(self, columns, steps, lineage, shortcuts, has_children): top, chart, spacing, bar = 24, 460, 274, 14 # Silhouettes are not square: they run from tall and thin to long and # flat. Fitting them all into one box turns most into slivers, so each # gets its own box with the same area and its own proportions. icon_area, icon_max, gutter = 23 * 23, 40, 46 width = max(830, 40 + (len(columns) - 1) * spacing + bar + 250) height = top * 2 + chart def stack(nodes): """Sankey layout: bar heights follow assembly counts, with room for every label.""" gap, slot = (12 if len(nodes) > 1 else 0), 46 fixed, scale = set(), 0 for _ in range(len(nodes) + 1): free = chart - gap * (len(nodes) - 1) - slot * len(fixed) flexible = sum(n['total'] for k, n in enumerate(nodes) if k not in fixed) scale = free / flexible if flexible else 0 small = fixed | {k for k, n in enumerate(nodes) if n['total'] * scale < slot} if small == fixed: break fixed = small y = top for k, n in enumerate(nodes): room = slot if k in fixed else n['total'] * scale n['h'] = max(3, n['total'] * scale) n['y'] = y + (room - n['h']) / 2 y += room + gap def clip(text, limit): return text if len(text) <= limit else text[:limit - 2] + "…" def share(n): return n['covered'] / n['total'] if self.coverage and n['total'] else 0 def ribbon(x0, a0, a1, x1, b0, b1, kind): bend = (x0 + x1) / 2 return (f'') flows, nodes = [], [] for depth, column in enumerate(columns): stack(column) x = 40 + depth * spacing for n in column: n['x'] = x for depth, column in enumerate(columns[:-1]): parent = next((n for n in column if n['selected']), None) children = columns[depth + 1] if not parent: continue total = sum(c['total'] for c in children) or 1 cursor = parent['y'] for child in children: a0 = cursor a1 = cursor + parent['h'] * child['total'] / total cursor = a1 f = share(child) am, bm = a0 + f * (a1 - a0), child['y'] + f * child['h'] x0, x1 = parent['x'] + bar, child['x'] if f > 0: flows.append(ribbon(x0, a0, am, x1, child['y'], bm, "annotated")) if f < 1: flows.append(ribbon(x0, am, a1, x1, bm, child['y'] + child['h'], "missing")) for depth, column in enumerate(columns): on_path = depth < len(steps) shown = Counter(n['icon'] for n in column if n['icon']) for n in column: if n['borrowed_icon'] and shown[n['icon']] > 1: n['icon'] = "" # identical borrowed outlines read as "same group" x, y, h = n['x'], n['y'], n['h'] f = share(n) percent = 100 * f title = f"{n['name']} ({n['english']})" if n['english'] else n['name'] detail = (f"{title} · {n['covered']:,} of {n['total']:,} assemblies have published annotations ({percent:.2f}%). Includes partial assemblies." if self.coverage else f"{title} · {n['total']:,} assemblies. Coverage inventory unavailable.") if n.get('aggregate'): detail += " Display aggregate, not a taxonomic clade. Activate to see its lineages." elif not n.get('direct') and not n['selected']: detail += " Activate to expand this group to the right." action = "" if n.get('direct') else f'data-path="{n["path"]}" role="button" tabindex="0"' classes = "tree-node" + (" is-selected" if n['selected'] else " is-sibling" if on_path else "") + \ (" is-direct" if n.get('direct') else "") nodes.append(f'') cy = y + h / 2 nodes.append(f'') green = h * f if green > 0: nodes.append(f'') if green < h: nodes.append(f'') nodes.append(f'') label, english = clip(n['label'], MAX_LABEL), clip(n['english'], MAX_ENGLISH) stats = f"{percent:.1f}% · {n['covered']:,} / {n['total']:,}" if self.coverage else f"{n['total']:,} assemblies" # The gutter is reserved whether or not this node has a silhouette, so # labels stay aligned down the column. if n['icon']: ratio = self.icon_credits.get(n['icon'], {}).get('ratio') or 1.0 icon_h = min(icon_max, max(6, math.sqrt(icon_area / ratio))) icon_w = min(icon_max, max(6, icon_h * ratio)) icon_h = min(icon_max, max(6, icon_w / ratio)) nodes.append(f'') text_x = x + bar + 8 + gutter # Three lines when there is an English gloss, two without, both centred on the bar. head, stat_line = (cy - 10, cy + 16) if english else (cy - 3, cy + 13) nodes.append(f'{escape(label)}') if english: nodes.append(f'{escape(english)}') nodes.append(f'{stats}') svg = (f'' + ''.join(flows) + ''.join(nodes) + '') focus = next(n for n in columns[len(steps) - 1] if n['selected']) total, covered = focus['total'], focus['covered'] percent = 100 * covered / total if self.coverage and total else 0 selected_name = f"Other lineages of {lineage[-1]['name']}" if focus.get('aggregate') else focus['label'] if focus.get('english'): selected_name += f" · {focus['english']}" crumbs = [] for depth, (row, (_, offset)) in enumerate(zip(lineage, steps)): name = "Other lineages" if offset else row['name'] crumbs.append(f'') breadcrumbs = ' / '.join(crumbs) parent = encode_path(steps[:-1] or steps) stamp = (self.coverage or self.metadata)['created_at'][:10] if self.coverage: summary = (f'
{escape(selected_name)} · ASSEMBLY COVERAGE
' f'
{covered:,}of {total:,}
assemblies annotated
' f'{percent:.1f}%
') legend = ('
Annotated' 'Not annotated
') else: summary = f'
{escape(selected_name)}
{total:,}GenBank assemblies
Coverage unavailable
' legend = '
Assemblies · coverage unavailable
' leaf_note = '

This is a terminal taxon in the assembly snapshot. Use the breadcrumb to explore its relatives.

' if not has_children else '' return f'''
HUGGING FACE BIO / GENOME ATLAS

The tree of eukaryotic life.

Follow the branches. Discover where we’ve annotated.

{summary}
{breadcrumbs}
{svg}
{leaf_note}
{legend} Click a group to expand it to the right · Swipe to explore Updated {stamp}
Bar and flow heights are proportional to assembly counts within each column; each column is rescaled to fill the height, so compare sizes within a column. Coverage counts assemblies with published annotations, including partial assemblies.
''' def credit_lines(taxonomy): """One line per artist, as CC BY asks, with the licences their images carry.""" artists = {} for image in set(taxonomy.icons.values()): meta = taxonomy.icon_credits.get(image, {}) name = meta.get("attribution") or meta.get("contributor") or "Unknown artist" entry = artists.setdefault(name, {"licenses": set(), "pages": []}) entry["licenses"].add(LICENSE_NAMES.get(meta.get("license"), meta.get("license") or "unknown licence")) entry["pages"].append(meta.get("page")) lines = [] for name in sorted(artists, key=str.casefold): entry = artists[name] count = len(entry["pages"]) lines.append(f"- [{name}]({entry['pages'][0]}) — {', '.join(sorted(entry['licenses']))}" f"{f' — {count} silhouettes' if count > 1 else ''}") return lines def build_taxonomy_tab(): """Render the tree and its controls inside the Genome atlas tab.""" if not DATABASE.exists(): gr.Markdown("The eukaryotic tree will appear when the taxonomy snapshot is available.") return taxonomy = Taxonomy() if taxonomy.icons: # The chart references each silhouette by URL, so the browser caches it # once instead of the file riding along in every re-render. gr.set_static_paths(paths=[ICON_DIR]) tree = gr.HTML(taxonomy.view(), css_template=(ROOT / "atlas.css").read_text(), js_on_load=(ROOT / "atlas.js").read_text(), apply_default_css=False, elem_id="eukaryotic-atlas") def explore_tree(evt: gr.EventData): try: return taxonomy.view(evt.route) except (ValueError, TypeError, AttributeError) as exc: raise gr.Error("Choose a eukaryotic group in this snapshot.") from exc tree.click(explore_tree, outputs=tree, show_progress="minimal", concurrency_limit=1) with gr.Accordion("Find a lineage", open=False, elem_classes="atlas-disclosure"): with gr.Row(): query = gr.Textbox(label="Eukaryotic taxon", placeholder="Scientific name, taxon ID, or a group such as animals") find = gr.Button("Search lineages", elem_classes="action-button") matches = gr.Dropdown(choices=[], label="Matching lineages", interactive=True) status = gr.Markdown() def taxonomy_search(text): choices = taxonomy.search(text) if str(text or '').strip() else [] return gr.Dropdown(choices=choices, value=None), f"{len(choices)} results (up to 100, largest first)." if choices else "No matching eukaryotic taxa in this snapshot." def taxonomy_view(taxid): return taxonomy.view_taxon(taxid) if taxid else taxonomy.view() matches.input(taxonomy_view, matches, tree) find.click(taxonomy_search, query, [matches, status]) query.submit(taxonomy_search, query, [matches, status]) if taxonomy.icons: with gr.Accordion("Silhouette credits", open=False, elem_classes="atlas-disclosure"): gr.Markdown("\n".join(["Silhouettes from [PhyloPic](https://www.phylopic.org), used unmodified. " "Public-domain images need no credit; the CC BY ones do, and every artist is listed here. " "`data/icons.json` maps each taxon to its image.", ""] + credit_lines(taxonomy))) with gr.Accordion("About the tree and coverage", open=False, elem_classes="atlas-disclosure"): gr.Markdown("This tree follows **NCBI Taxonomy within Eukaryota**. Animals and fungi share the Opisthokonta branch; " "green plants include green algae. ‘Other lineages’ combines the remaining siblings for display and can be expanded. " "Each expanded group adds a column on the right with up to six direct lineages plus any remainder, so every ancestor stays visible. " "The opening view also expands the animal/fungal branch. Bars and flows are drawn like a Sankey diagram: " "heights follow **actual assembly counts** within each column, split into annotated and not-annotated assemblies.\n\n" "Each group shows its scientific name with a plain-English name underneath where one is available. These come from " "NCBI Taxonomy's common names plus a curated set of short glosses for the large unranked clades NCBI leaves unnamed, " "such as Opisthokonta (animals and fungi) or Ecdysozoa (moulting animals). The glosses describe the living members of a " "clade in everyday words; they are not formal taxonomic synonyms, so the scientific name stays the primary label.\n\n" "Coverage means the fraction of current GenBank assembly versions with published annotations, including partial assemblies. " "It does not measure the fraction of bases annotated. Parent counts include descendants once each. " "Exact accession versions are matched. The Database tab searches a dated snapshot of all published annotation files.\n\n" "Sources: [NCBI assembly summary](https://ftp.ncbi.nlm.nih.gov/genomes/ASSEMBLY_REPORTS/assembly_summary_genbank.txt), " "[NCBI Taxonomy](https://ftp.ncbi.nlm.nih.gov/pub/taxonomy/taxdump.tar.gz) for both the tree and its common names, " "and [published annotations](https://huggingface.co/buckets/HuggingFaceBio/genbank-annotations).") if taxonomy.icons: gr.Markdown("Silhouettes come from [PhyloPic](https://www.phylopic.org), resolved from each taxon's NCBI ID and " "stored unmodified. Only public-domain (CC0) and CC BY 4.0 images are used; NonCommercial and " "ShareAlike images are filtered out at build time. A group without a silhouette of its own borrows " "its nearest illustrated ancestor, so a fungal genus shows the fungal outline.") coverage = taxonomy.coverage or {} gr.JSON({"taxonomy_updated": taxonomy.metadata['created_at'], "coverage_updated": coverage.get('created_at'), "silhouettes": taxonomy.icon_meta or None, "display_scope": "Eukaryota (NCBI taxid 2759) and its descendants only", "inventory_method": coverage.get('method'), "published_versions_outside_current_GenBank_snapshot": coverage.get('unmatched_assembly_versions'), "sources": taxonomy.metadata['sources']}, label="Snapshot provenance")