"""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'
This is a terminal taxon in the assembly snapshot. Use the breadcrumb to explore its relatives.
' if not has_children else '' return f'''Follow the branches. Discover where we’ve annotated.