| """Compact, tracked provenance for untracked external benchmark assets.""" |
|
|
| from __future__ import annotations |
|
|
| import json |
| import subprocess |
| from pathlib import Path |
| from typing import Any |
|
|
| import pandas as pd |
| from Bio import SeqIO |
|
|
| from .artifacts import file_record, sha256_json, write_json_immutable |
|
|
|
|
| def _git_head(path: Path) -> str: |
| completed = subprocess.run( |
| ["git", "rev-parse", "HEAD"], cwd=path, text=True, capture_output=True, check=False |
| ) |
| return completed.stdout.strip() if completed.returncode == 0 else "unavailable" |
|
|
|
|
| def _fasta_count(path: Path) -> int: |
| with path.open("r", encoding="utf-8") as handle: |
| return sum(1 for _ in SeqIO.parse(handle, "fasta")) |
|
|
|
|
| def build_external_manifest( |
| external_root: str | Path, |
| output_path: str | Path, |
| benchmark_commit: str, |
| archive_url: str, |
| archive_sha256: str, |
| supersedes: str | None = None, |
| ) -> dict[str, Any]: |
| root = Path(external_root).resolve() |
| benchmark = root / "bgc-clustering-benchmark" |
| processed = root / "processed" |
| key_inputs = [ |
| benchmark / "source_data/NPAtlas_bm_v1.tsv", |
| benchmark / "tanimoto_results/NPAtlas_bm_v1.tsv", |
| benchmark / "bgc_similarities/bigscape_similarity_score_1.csv", |
| ] |
| licenses = [ |
| benchmark / "LICENSE", |
| benchmark / "source_data/LICENSE", |
| root / "BGC-MLM/LICENSE", |
| root / "BGC-MAP/README.md", |
| ] |
| processed_files = [ |
| processed / "gold_bgc_product_mapping.csv", |
| processed / "all_bgc_product_metadata.csv", |
| processed / "ambiguous_product_exclusions.csv", |
| processed / "external_proteins.fasta", |
| processed / "external_atlas.csv", |
| processed / "external_sequence_provenance.json", |
| ] |
| required = key_inputs + licenses + processed_files |
| if missing := [str(path) for path in required if not path.is_file()]: |
| raise FileNotFoundError(f"External manifest inputs are missing: {missing}") |
|
|
| gold = pd.read_csv(processed / "gold_bgc_product_mapping.csv") |
| metadata = pd.read_csv(processed / "all_bgc_product_metadata.csv") |
| exclusions = pd.read_csv(processed / "ambiguous_product_exclusions.csv") |
| atlas = pd.read_csv(processed / "external_atlas.csv") |
| with (processed / "external_sequence_provenance.json").open("r", encoding="utf-8") as handle: |
| sequence_provenance = json.load(handle) |
| records = { |
| "key_inputs": [file_record(path, root) for path in key_inputs], |
| "licenses": [file_record(path, root) for path in licenses], |
| "processed_outputs": [file_record(path, root) for path in processed_files], |
| } |
| manifest = { |
| "schema_version": 1, |
| "benchmark": { |
| "repository": "https://github.com/aswalker-lab/BGC-clustering-benchmark", |
| "archive_commit": benchmark_commit, |
| "archive_url": archive_url, |
| "archive_sha256": archive_sha256, |
| "local_git_head_informational": _git_head(benchmark), |
| }, |
| "released_baselines": { |
| "BGC-MLM": { |
| "repository": "https://github.com/aswalker-lab/BGC-MLM", |
| "commit": _git_head(root / "BGC-MLM"), |
| }, |
| "BGC-MAP": { |
| "repository": "https://github.com/EvoCatalysis/BGC_annotation", |
| "commit": _git_head(root / "BGC-MAP"), |
| }, |
| }, |
| "license_note": ( |
| "Benchmark source data are CC BY-NC 4.0; repository code and baseline " |
| "licenses remain governed by their recorded upstream license files." |
| ), |
| "counts": { |
| "all_mapped_bgcs": int(metadata["bgc_id"].nunique()), |
| "unambiguous_gold_bgcs": int(gold["bgc_id"].nunique()), |
| "ambiguous_exclusions": int(exclusions["bgc_id"].nunique()), |
| "sequence_bgcs": int(atlas["bgc_id"].nunique()), |
| "proteins": _fasta_count(processed / "external_proteins.fasta"), |
| "source_genbank_files": int(sequence_provenance["source_files"]), |
| }, |
| "records": records, |
| "records_sha256": sha256_json(records), |
| } |
| if supersedes is not None: |
| manifest["supersedes"] = supersedes |
| write_json_immutable(output_path, manifest) |
| return manifest |
|
|
