diff --git a/database/scripts/load_bgcs.py b/database/scripts/load_bgcs.py index 0963e4b..f6f88fc 100644 --- a/database/scripts/load_bgcs.py +++ b/database/scripts/load_bgcs.py @@ -12,9 +12,16 @@ provide (MIBiG 4.0's GBKs dropped the ACCESSION.VERSION suffix) -- so it's looked up here instead, from the same JSON-derived mibig_versions.json extract_mibig_compounds.py produces for load_compounds.py. + +BGCs are deduplicated per source .gbk file via `file_hash` (parse_gbks.py's +sha256 of the whole file's raw text): a region's entry id is derived from that +hash plus its own readout id, and a file whose hash is already stored (checked +via RetroMolDuckDB.bgc_content_hash_exists) is skipped outright, so rerunning +the pipeline over an already-ingested file doesn't redo any fingerprinting work. """ import argparse +import hashlib import json import logging from pathlib import Path @@ -43,6 +50,7 @@ def run( added = 0 skipped = 0 + skipped_existing_file = 0 db = RetroMolDuckDB.open(db_path) try: @@ -53,6 +61,12 @@ def run( continue entry = json.loads(line) + file_hash = entry.get("file_hash") + + if file_hash and db.bgc_content_hash_exists(file_hash): + skipped_existing_file += 1 + continue + readout = LinearReadout.from_dict(entry["readout"]) names, tokens = bgc_primary_sequence(readout, ruleset) @@ -64,20 +78,27 @@ def run( accession = entry.get("accession") name = f"{accession} ({readout.id})" if accession else readout.id url = mibig_url(accession, versions.get(accession)) if accession else None + entry_id = hashlib.sha256(f"{file_hash}:{readout.id}".encode("utf-8")).hexdigest() db.add_entry( + entry_id=entry_id, name=name, + database_name="MIBiG", url=url, raw=entry.get("raw_gbk") if include_raw_gbk else None, entry_type="bgc", primary_sequence=names, fingerprint=fp, + content_hash=file_hash, ) added += 1 finally: db.close() - log.info("load_bgcs: added=%d skipped=%d", added, skipped) + log.info( + "load_bgcs: added=%d skipped=%d skipped_existing_file=%d", + added, skipped, skipped_existing_file, + ) def main() -> None: diff --git a/database/scripts/load_compounds.py b/database/scripts/load_compounds.py index 645ffec..b22a5ab 100644 --- a/database/scripts/load_compounds.py +++ b/database/scripts/load_compounds.py @@ -4,8 +4,13 @@ compound's tailoring events (glycosylation, methylation -- anything that shows up as its own disconnected single-node path) -- is merged into the one primary sequence stored for that compound: longest path first, ties broken lexicographically, joined -by TOKEN_LINK (see common.primary_sequence_from_result). One compound -> one db -entry, always. `raw` is the original input SMILES. +by TOKEN_LINK (see common.primary_sequence_from_result). `raw` is the original input +SMILES. + +Compounds are deduplicated on `result.submission.inchikey` (stereo-aware, computed by +retromol.model.submission.Submission with keep_stereo=True). The same molecule +appearing in both NPAtlas and MIBiG lands as one database entry with two source +records (see RetroMolDuckDB.add_entry) rather than two separate rows. """ import argparse @@ -31,6 +36,8 @@ log = logging.getLogger(__name__) +DATABASE_NAMES = {"npatlas": "NPAtlas", "mibig": "MIBiG"} + def _npatlas_name_and_url(props: dict) -> tuple[str | None, str | None]: npaid_key = find_key_ci(props, ["npaid"]) @@ -102,7 +109,9 @@ def run( fp = fingerprinter.encode(tokens) db.add_entry( + entry_id=result.submission.inchikey, name=name, + database_name=DATABASE_NAMES[source], url=url, raw=result.submission.smiles, entry_type="compound", diff --git a/database/scripts/parse_gbks.py b/database/scripts/parse_gbks.py index 9088051..cb33997 100644 --- a/database/scripts/parse_gbks.py +++ b/database/scripts/parse_gbks.py @@ -29,6 +29,7 @@ """ import argparse +import hashlib import itertools import json import logging @@ -86,6 +87,7 @@ def _process_file(path_str: str) -> tuple[list[dict], str | None]: path = Path(path_str) try: raw_gbk = path.read_text() + file_hash = hashlib.sha256(raw_gbk.encode("utf-8")).hexdigest() regions = parse_antismash_gbk(path, AntiSmashOptions()) entries: list[dict] = [] @@ -105,6 +107,7 @@ def _process_file(path_str: str) -> tuple[list[dict], str | None]: "accession": accession, "file_name": region.file_name, "raw_gbk": raw_gbk, + "file_hash": file_hash, "readout": readout.to_dict(), }) diff --git a/gui/src/client/src/components/workspace/AlignmentGrid.tsx b/gui/src/client/src/components/workspace/AlignmentGrid.tsx index 3103a52..2e46c8c 100644 --- a/gui/src/client/src/components/workspace/AlignmentGrid.tsx +++ b/gui/src/client/src/components/workspace/AlignmentGrid.tsx @@ -12,7 +12,6 @@ import Collapse from "@mui/material/Collapse"; import Stack from "@mui/material/Stack"; import Tooltip from "@mui/material/Tooltip"; import Typography from "@mui/material/Typography"; -import MuiLink from "@mui/material/Link"; import ExpandMoreIcon from "@mui/icons-material/ExpandMore"; import ExpandLessIcon from "@mui/icons-material/ExpandLess"; import DownloadIcon from "@mui/icons-material/Download"; @@ -236,7 +235,6 @@ export function ResultRow({ sendingToUploads?: boolean; }) { const [expanded, setExpanded] = React.useState(false); - const theme = useTheme(); const canOfferSend = result.origin === "database" && !!onSendToUploads; const hasRaw = !!result.raw; @@ -262,22 +260,9 @@ export function ResultRow({ - {result.url ? ( - - {result.name} - - ) : ( - - {result.name} - - )} + + {result.name} + + {result.sources.length > 0 && ( + + {result.sources.map((source, idx) => + source.url ? ( + + ) : ( + + ) + )} + + )} = ({ session, display: "flex", alignItems: "center", gap: 1.5, - flexWrap: "wrap", + minWidth: 0, }} > {`reconstructed backbone ${idx+1}`} {override && ( - + )} - + @@ -695,20 +695,20 @@ export const WorkspaceDiscovery: React.FC = ({ session, display: "flex", alignItems: "center", gap: 1.5, - flexWrap: "wrap", + minWidth: 0, }} > {region.id} - + diff --git a/gui/src/client/src/features/discovery/types.ts b/gui/src/client/src/features/discovery/types.ts index caff6de..1cc3baf 100644 --- a/gui/src/client/src/features/discovery/types.ts +++ b/gui/src/client/src/features/discovery/types.ts @@ -85,6 +85,11 @@ export const DiscoveryResultSchema = z.object({ // database entry can be resubmitted for parsing under the current rule set. Null for // upload-origin BGC candidates and for any entry that predates `raw` being stored. raw: z.string().nullable().default(null), + // Every (name, source database) this entry is known under -- e.g. a compound present + // in both NPAtlas and MIBiG carries two entries here. Empty for upload-origin results. + sources: z + .array(z.object({ name: z.string(), databaseName: z.string(), url: z.string().nullable() })) + .default([]), fingerprintSimilarity: z.number(), primarySequence: z.array(z.string()), inverted: z.boolean(), diff --git a/gui/src/server/routes/discovery.py b/gui/src/server/routes/discovery.py index c4a0a0a..8a43ef9 100644 --- a/gui/src/server/routes/discovery.py +++ b/gui/src/server/routes/discovery.py @@ -720,6 +720,14 @@ def rank_key(align_score: float, target_self_score: float | None) -> float: # offer "send back to Uploads" so a retrieved entry can be reparsed under # the current rule set without the user having to re-supply it by hand. "raw": candidate.entry.raw, + # Every (name, database) this entry is known under -- e.g. a compound + # present in both NPAtlas and MIBiG carries two sources. Empty for + # upload-origin candidates, which aren't backed by any source database + # (see Entry.sources / duckdb.py). + "sources": [ + {"name": s.name, "databaseName": s.database_name, "url": s.url} + for s in candidate.entry.sources + ], "fingerprintSimilarity": candidate.similarity, "primarySequence": [_denormalize_for_display(t) for t in oriented_target], "inverted": inverted, diff --git a/gui/src/server/routes/jobs.py b/gui/src/server/routes/jobs.py index 69a3c54..2959e7c 100644 --- a/gui/src/server/routes/jobs.py +++ b/gui/src/server/routes/jobs.py @@ -97,22 +97,23 @@ def search_compound_by_name(): rows = db.con.execute( """ SELECT - min(id) AS id, - name, - url, - raw - FROM entries - WHERE type = 'compound' - AND raw IS NOT NULL - AND lower(name) LIKE lower(?) - GROUP BY name, url, raw + es.entry_id AS id, + es.name AS name, + es.database_name AS database_name, + es.url AS url, + e.raw AS raw + FROM entry_sources es + JOIN entries e ON e.id = es.entry_id + WHERE e.type = 'compound' + AND e.raw IS NOT NULL + AND lower(es.name) LIKE lower(?) ORDER BY CASE - WHEN lower(name) = lower(?) THEN 0 - WHEN lower(name) LIKE lower(?) THEN 1 + WHEN lower(es.name) = lower(?) THEN 0 + WHEN lower(es.name) LIKE lower(?) THEN 1 ELSE 2 END, - name + es.name LIMIT ? """, [like, q, f"{q}%", limit], @@ -126,11 +127,11 @@ def search_compound_by_name(): { "name": name, "smiles": raw, - "databaseName": "RetroMol", + "databaseName": database_name, "databaseIdentifier": entry_id, "url": url, } - for entry_id, name, url, raw in rows + for entry_id, name, database_name, url, raw in rows if name and raw ] diff --git a/src/retromol_database/duckdb.py b/src/retromol_database/duckdb.py index 0e76812..8d0de2d 100644 --- a/src/retromol_database/duckdb.py +++ b/src/retromol_database/duckdb.py @@ -1,8 +1,6 @@ -import hashlib -import json -from dataclasses import dataclass +from dataclasses import dataclass, field from pathlib import Path -from typing import Iterable, Iterator, Literal, Sequence +from typing import Iterator, Literal, Sequence import duckdb import numpy as np @@ -14,15 +12,29 @@ FINGERPRINT_SIZE = 1024 +@dataclass(frozen=True) +class EntrySource: + name: str + database_name: str + url: str | None + + @dataclass(frozen=True) class Entry: id: str + # Primary display name/url -- the first-ever-inserted source for this entry (see + # `sources` below and entry_sources.seq). Kept as plain fields (not derived via a + # property) so every existing read site and the synthetic upload-candidate Entry(...) + # constructions in routes/discovery.py keep working unchanged. name: str url: str | None raw: str | None type: EntryType primary_sequence: list[str] fingerprint: list[float] + # Every (name, database_name, url) a source has contributed for this entry, ordered + # by insertion. Empty for synthetic (non-database) entries, e.g. session uploads. + sources: list[EntrySource] = field(default_factory=list) @dataclass(frozen=True) @@ -69,25 +81,6 @@ def _normalize_fingerprint(fingerprint: Sequence[float] | np.ndarray) -> list[fl return fp.astype(float).tolist() -def make_entry_id( - *, - name: str, - url: str | None, - raw: str | None, - entry_type: str, - primary_sequence: Sequence[str], -) -> str: - payload = { - "name": name, - "url": url, - "raw": raw, - "type": entry_type, - "primary_sequence": list(primary_sequence), - } - raw = json.dumps(payload, sort_keys=True, separators=(",", ":")) - return hashlib.sha256(raw.encode("utf-8")).hexdigest() - - class RetroMolDuckDB: def __init__(self, path: str | Path, *, read_only: bool = False) -> None: self.path = Path(path).expanduser() @@ -128,95 +121,88 @@ def close(self) -> None: self.con.close() def create_schema(self) -> None: + self.con.execute("CREATE SEQUENCE IF NOT EXISTS entry_sources_seq") self.con.execute( f""" CREATE TABLE IF NOT EXISTS entries ( id VARCHAR PRIMARY KEY, - name VARCHAR NOT NULL, - url VARCHAR, - raw VARCHAR, type VARCHAR NOT NULL CHECK (type IN ('compound', 'bgc')), + raw VARCHAR, + content_hash VARCHAR, primary_sequence VARCHAR[] NOT NULL, fingerprint FLOAT[{FINGERPRINT_SIZE}] NOT NULL ) """ ) + self.con.execute( + """ + CREATE TABLE IF NOT EXISTS entry_sources ( + seq BIGINT DEFAULT nextval('entry_sources_seq'), + entry_id VARCHAR NOT NULL, + name VARCHAR NOT NULL, + database_name VARCHAR NOT NULL, + url VARCHAR, + PRIMARY KEY (entry_id, database_name, name) + ) + """ + ) def add_entry( self, *, + entry_id: str, name: str, + database_name: str, url: str | None, raw: str | None, entry_type: str, primary_sequence: Sequence[str], fingerprint: Sequence[float] | np.ndarray, - entry_id: str | None = None, + content_hash: str | None = None, ) -> str: + """ + Add (or extend) an entry. + + `entry_id` is the caller-supplied molecular identity -- an InChIKey for a + compound, or a hash of the source .gbk file's content plus region id for a + bgc (see database/scripts/load_compounds.py / load_bgcs.py). If an entry + with this id already exists, its stored `raw`/`primary_sequence`/ + `fingerprint` are left untouched (they describe the same molecule/BGC + either way) and only a new `(name, database_name, url)` source row is + added -- or, if that exact (entry_id, database_name, name) combination was + already recorded, its url is refreshed. + """ entry_type = _normalize_entry_type(entry_type) sequence = _normalize_primary_sequence(primary_sequence) fp = _normalize_fingerprint(fingerprint) - entry_id: str = entry_id or make_entry_id( - name=name, - url=url, - raw=raw, - entry_type=entry_type, - primary_sequence=sequence, - ) - self.con.execute( """ - INSERT OR REPLACE INTO entries ( - id, - name, - url, - raw, - type, - primary_sequence, - fingerprint - ) - VALUES (?, ?, ?, ?, ?, ?, ?) + INSERT INTO entries (id, type, raw, content_hash, primary_sequence, fingerprint) + VALUES (?, ?, ?, ?, ?, ?) + ON CONFLICT (id) DO NOTHING """, - [entry_id, name, url, raw, entry_type, sequence, fp], + [entry_id, entry_type, raw, content_hash, sequence, fp], ) - return entry_id - - def add_entries(self, entries: Iterable[Entry]) -> int: - rows = [ - ( - entry.id, - entry.name, - entry.url, - entry.raw, - _normalize_entry_type(entry.type), - _normalize_primary_sequence(entry.primary_sequence), - _normalize_fingerprint(entry.fingerprint), - ) - for entry in entries - ] - - if not rows: - return 0 - - self.con.executemany( + self.con.execute( """ - INSERT OR REPLACE INTO entries ( - id, - name, - url, - raw, - type, - primary_sequence, - fingerprint - ) - VALUES (?, ?, ?, ?, ?, ?, ?) + INSERT INTO entry_sources (entry_id, name, database_name, url) + VALUES (?, ?, ?, ?) + ON CONFLICT (entry_id, database_name, name) DO UPDATE SET url = excluded.url """, - rows, + [entry_id, name, database_name, url], ) - return len(rows) + return entry_id + + def bgc_content_hash_exists(self, content_hash: str) -> bool: + """Whether a bgc entry sourced from a .gbk file with this content hash already exists.""" + row = self.con.execute( + "SELECT 1 FROM entries WHERE type = 'bgc' AND content_hash = ? LIMIT 1", + [content_hash], + ).fetchone() + return row is not None def count(self) -> int: return int(self.con.execute("SELECT count(*) FROM entries").fetchone()[0]) @@ -268,9 +254,14 @@ def stats(self) -> DatabaseStats: url_row = self.con.execute( """ SELECT - count(*) FILTER (WHERE url IS NOT NULL), - count(*) FILTER (WHERE url IS NULL) - FROM entries + count(*) FILTER (WHERE has_url), + count(*) FILTER (WHERE NOT has_url) + FROM ( + SELECT e.id, bool_or(es.url IS NOT NULL) AS has_url + FROM entries e + LEFT JOIN entry_sources es ON es.entry_id = e.id + GROUP BY e.id + ) """ ).fetchone() with_source_url_count = int(url_row[0]) @@ -287,10 +278,32 @@ def stats(self) -> DatabaseStats: without_source_url_count=without_source_url_count, ) + def _sources_for_entry_ids(self, entry_ids: Sequence[str]) -> dict[str, list[EntrySource]]: + """Batch-fetch every (name, database_name, url) source for the given entry ids, ordered by insertion.""" + if not entry_ids: + return {} + + rows = self.con.execute( + """ + SELECT entry_id, name, database_name, url + FROM entry_sources + WHERE entry_id IN (SELECT UNNEST(?)) + ORDER BY entry_id, seq + """, + [list(entry_ids)], + ).fetchall() + + out: dict[str, list[EntrySource]] = {} + for entry_id, name, database_name, url in rows: + out.setdefault(str(entry_id), []).append( + EntrySource(name=str(name), database_name=str(database_name), url=url) + ) + return out + def get_entry(self, entry_id: str) -> Entry | None: row = self.con.execute( """ - SELECT id, name, url, raw, type, primary_sequence, fingerprint + SELECT id, raw, type, primary_sequence, fingerprint FROM entries WHERE id = ? """, @@ -300,19 +313,21 @@ def get_entry(self, entry_id: str) -> Entry | None: if row is None: return None - return _entry_from_row(row) + sources = self._sources_for_entry_ids([entry_id]).get(entry_id, []) + return _entry_from_row(row, sources) def iter_entries(self) -> Iterator[Entry]: rows = self.con.execute( """ - SELECT id, name, url, raw, type, primary_sequence, fingerprint + SELECT id, raw, type, primary_sequence, fingerprint FROM entries ORDER BY id """ ).fetchall() + sources_by_id = self._sources_for_entry_ids([str(row[0]) for row in rows]) for row in rows: - yield _entry_from_row(row) + yield _entry_from_row(row, sources_by_id.get(str(row[0]), [])) def closest( self, @@ -339,8 +354,6 @@ def closest( f""" SELECT id, - name, - url, raw, type, primary_sequence, @@ -354,10 +367,12 @@ def closest( params, ).fetchall() + sources_by_id = self._sources_for_entry_ids([str(row[0]) for row in rows]) + return [ SearchResult( - entry=_entry_from_row(row[:7]), - similarity=float(row[7]), + entry=_entry_from_row(row[:5], sources_by_id.get(str(row[0]), [])), + similarity=float(row[5]), ) for row in rows ] @@ -368,15 +383,17 @@ def export_parquet(self, path: str | Path) -> None: [str(path)], ) -def _entry_from_row(row) -> Entry: +def _entry_from_row(row, sources: list[EntrySource]) -> Entry: + primary = sources[0] if sources else None return Entry( id=str(row[0]), - name=str(row[1]), - url=row[2], # don't turn into str, might be None - raw=row[3], - type=_normalize_entry_type(row[4]), - primary_sequence=list(row[5]), - fingerprint=list(row[6]), + name=primary.name if primary else str(row[0]), + url=primary.url if primary else None, + raw=row[1], + type=_normalize_entry_type(row[2]), + primary_sequence=list(row[3]), + fingerprint=list(row[4]), + sources=sources, )