File size: 2,992 Bytes
c87881a | 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 | """Legacy evidence audit; all operations are read-only until an output is written."""
from __future__ import annotations
import re
from pathlib import Path
from typing import Any
import h5py
import pandas as pd
from .data import FORBIDDEN_MODEL_INPUTS
MIBIG_PATTERN = re.compile(r"^BGC\d{7}$")
def audit_legacy_data(atlas_csv: str | Path, genes_csv: str | Path, embeddings_h5: str | Path) -> dict[str, Any]:
atlas = pd.read_csv(atlas_csv)
genes = pd.read_csv(genes_csv)
labeled = atlas.dropna(subset=["compound_family"]).copy()
target_by_bgc = labeled.set_index("bgc_id")["compound_family"].astype(str).to_dict()
direct_hits: dict[str, int] = {}
if "mibig_id" in genes:
gene_targets = genes["mibig_id"].fillna("").astype(str).str.extract(r"^(BGC\d{7})", expand=False)
matches = gene_targets == genes["bgc_id"].map(target_by_bgc)
direct_hits = matches.groupby(genes["bgc_id"]).sum().astype(int).to_dict()
with h5py.File(embeddings_h5, "r") as handle:
embedding_keys = set(handle.keys())
gene_ids = set(genes["gene_id"].astype(str))
group_sizes = labeled.groupby("compound_family")["bgc_id"].nunique()
return {
"schema_version": 1,
"atlas": {
"rows": int(len(atlas)),
"unique_bgcs": int(atlas["bgc_id"].nunique()),
"unique_genomes": int(atlas["genome_id"].nunique()),
"labeled_bgcs": int(labeled["bgc_id"].nunique()),
"reference_ids": int(labeled["compound_family"].nunique()),
"eligible_groups_at_least_5": int((group_sizes >= 5).sum()),
"labels_matching_mibig_accession": int(
labeled["compound_family"].astype(str).map(lambda value: bool(MIBIG_PATTERN.match(value))).sum()
),
"labels_with_at_least_3_direct_target_hits": int(
sum(direct_hits.get(str(identifier), 0) >= 3 for identifier in labeled["bgc_id"])
),
},
"genes": {
"rows": int(len(genes)),
"unique_gene_ids": int(genes["gene_id"].nunique()),
"duplicate_gene_ids_across_rows": int(len(genes) - genes["gene_id"].nunique()),
"duplicate_composite_keys": int(genes.duplicated(["bgc_id", "gene_id"]).sum()),
},
"embeddings": {
"keys": len(embedding_keys),
"unique_gene_coverage": (len(gene_ids & embedding_keys) / len(gene_ids)) if gene_ids else 0.0,
"missing_unique_gene_ids": len(gene_ids - embedding_keys),
},
"leakage": {
"label_generation_columns_present": sorted(
{"mibig_id", "pident", "qcovs", "evalue"}.intersection(genes.columns)
),
"forbidden_atlas_feature_columns_present": sorted(FORBIDDEN_MODEL_INPUTS.intersection(atlas.columns)),
"finding": "Legacy labels are MIBiG-reference pseudo-labels and label-generating alignment fields must not be model inputs.",
},
}
|