| """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.", |
| }, |
| } |
|
|
|
|
