| """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 |
|
|
