File size: 2,034 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 | """Materialize fail-closed data-quality evidence for a frozen split."""
from __future__ import annotations
from pathlib import Path
from typing import Any
from .artifacts import sha256_file, write_json_immutable
from .config import LoadedConfig
from .data import ALLOWED_MODEL_INPUTS, BGCEmbeddingDataset
from .splits import load_split, split_summary
def validate_frozen_dataset(
config: LoadedConfig,
split_path: str | Path,
output_dir: str | Path,
) -> dict[str, Any]:
assignments = load_split(split_path)
dataset = BGCEmbeddingDataset(
config.resolve_path("data", "embeddings_h5"),
config.resolve_path("data", "atlas_csv"),
assignments,
int(config.values["model"]["esm_dimension"]),
)
output = Path(output_dir)
output.mkdir(parents=True, exist_ok=True)
exclusion_path = output / "missing_embedding_rows.csv"
if exclusion_path.exists():
raise FileExistsError(f"Refusing to overwrite quality artifact: {exclusion_path}")
dataset.missing_gene_rows.to_csv(exclusion_path, index=False)
usable_gene_rows = sum(len(dataset.bgc_to_genes[bgc_id]) for bgc_id in dataset.bgc_ids)
missing_gene_rows = len(dataset.missing_gene_rows)
report: dict[str, Any] = {
"schema_version": 1,
"split_file_sha256": sha256_file(split_path),
"atlas_file_sha256": sha256_file(config.resolve_path("data", "atlas_csv")),
"embeddings_file_sha256": sha256_file(config.resolve_path("data", "embeddings_h5")),
"split": split_summary(assignments),
"usable_bgcs": len(dataset),
"usable_gene_rows": usable_gene_rows,
"missing_embedding_rows": missing_gene_rows,
"embedding_coverage": usable_gene_rows / (usable_gene_rows + missing_gene_rows),
"position_source": "ordered community_atlas.protein_ids",
"model_input_whitelist": sorted(ALLOWED_MODEL_INPUTS),
"zero_gene_fallback": False,
}
write_json_immutable(output / "data_quality.json", report)
return report
|