File size: 5,726 Bytes
21b75e3 | 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 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 | from __future__ import annotations
import json
import pandas as pd
from deepgenopix.cli import (
build_parser,
build_experiment_matrix,
main,
list_presets,
parse_overrides,
runnable_preset_names,
)
from deepgenopix.notebook_support import discover_raw_inputs, discover_split_parquet_root
def test_parse_overrides_requires_json_object():
try:
parse_overrides('["baseline_v1"]')
except ValueError as exc:
assert "JSON object" in str(exc)
else:
raise AssertionError("expected ValueError")
def test_list_presets_excludes_blocked_by_default():
preset_names = {row["preset"] for row in list_presets()}
assert "baseline_v1" in preset_names
assert "zero_flank_v1" not in preset_names
assert "walk9_v1" in preset_names
assert "walk6_v1" in preset_names
def test_runnable_preset_names_include_walk_presets():
assert runnable_preset_names() == [
"baseline_v1",
"stride2_v1",
"stride8_v1",
"latent128_v1",
"latent768_v1",
"layers4_v1",
"stem5_v1",
"stem7_v1",
"walk9_v1",
"walk6_v1",
]
def test_build_experiment_matrix_prefers_recovered_parquet(tmp_path):
raw_dir = tmp_path / "data" / "raw"
raw_dir.mkdir(parents=True)
(raw_dir / "te_seqdata.parquet").write_bytes(b"")
(raw_dir / "te_seqdata.recovered.parquet").write_bytes(b"")
matrix = build_experiment_matrix(repo_root=tmp_path)
assert [row["preset"] for row in matrix] == runnable_preset_names()
assert matrix[0]["compare_against"] is None
assert matrix[1]["compare_against"] == "baseline_v1"
assert matrix[2]["bp_per_token"] == 96
assert matrix[0]["raw_parquet"] == "data/raw/te_seqdata.recovered.parquet"
assert "recovered.parquet" in matrix[0]["raw_input_policy"]
def test_build_experiment_matrix_prefers_split_dataset(tmp_path):
raw_dir = tmp_path / "data" / "raw"
for split in ("train", "val", "test"):
split_dir = raw_dir / split
split_dir.mkdir(parents=True, exist_ok=True)
pd.DataFrame({"sequence": ["A" * 12], "family": ["fam_a"]}).to_parquet(
split_dir / "te_seqdata.parquet",
index=False,
)
(raw_dir / "split_summary.json").write_text(json.dumps({"counts": {"train": 1, "val": 1, "test": 1}}), encoding="utf-8")
matrix = build_experiment_matrix(repo_root=tmp_path)
assert matrix[0]["raw_parquet"] is None
assert matrix[0]["raw_split_root"] == "data/raw"
assert matrix[0]["raw_split_summary"] == "data/raw/split_summary.json"
assert "split_summary.json" in matrix[0]["raw_input_policy"]
def test_discover_raw_inputs_prefers_recovered_parquet(tmp_path):
raw_dir = tmp_path / "data" / "raw"
raw_dir.mkdir(parents=True)
(raw_dir / "te_seqdata_with_biostats.parquet").write_bytes(b"")
(raw_dir / "te_seqdata.parquet").write_bytes(b"")
(raw_dir / "te_seqdata.recovered.parquet").write_bytes(b"")
raw_fasta, raw_parquet = discover_raw_inputs(raw_dir)
assert raw_fasta is None
assert raw_parquet == raw_dir / "te_seqdata.recovered.parquet"
def test_discover_split_parquet_root_returns_summary_path(tmp_path):
raw_dir = tmp_path / "data" / "raw"
for split in ("train", "val", "test"):
split_dir = raw_dir / split
split_dir.mkdir(parents=True, exist_ok=True)
pd.DataFrame({"sequence": ["A" * 12], "family": ["fam_a"]}).to_parquet(
split_dir / "te_seqdata.parquet",
index=False,
)
summary_path = raw_dir / "split_summary.json"
summary_path.write_text("{}", encoding="utf-8")
split_root, split_summary = discover_split_parquet_root(raw_dir)
assert split_root == raw_dir
assert split_summary == summary_path
def test_cli_prep_subcommand_writes_split(tmp_path):
source = tmp_path / "te_seqdata.recovered.parquet"
output_root = tmp_path / "te_split"
pd.DataFrame(
{
"sequence": ["A" * 12, "C" * 12, "G" * 12, "T" * 12],
"family": ["fam_a", "fam_a", "fam_b", "fam_b"],
}
).to_parquet(source, index=False)
exit_code = main(
[
"prep",
"--input",
str(source),
"--output-root",
str(output_root),
"--json",
]
)
assert exit_code == 0
assert (output_root / "split_summary.json").exists()
assert (output_root / "train" / "te_seqdata.parquet").exists()
def test_cli_etl_subcommand_writes_lmdb(tmp_path):
split_root = tmp_path / "te_split"
output_dir = tmp_path / "processed"
for split, rows in {
"train": [{"sequence": "A" * 12, "family": "fam_a"}],
"val": [{"sequence": "C" * 12, "family": "fam_b"}],
"test": [{"sequence": "G" * 12, "family": "fam_c"}],
}.items():
split_dir = split_root / split
split_dir.mkdir(parents=True, exist_ok=True)
pd.DataFrame(rows).to_parquet(split_dir / "te_seqdata.parquet", index=False)
exit_code = main(
[
"etl",
"--split-root",
str(split_root),
"--output-dir",
str(output_dir),
"--pixel-stride-bp",
"9",
"--json",
]
)
assert exit_code == 0
assert (output_dir / "tensors.lmdb").exists()
registry = pd.read_csv(output_dir / "registry.csv")
assert registry["pixel_stride_bp"].tolist() == [9, 9, 9]
def test_cli_help_includes_quant_and_variant_commands():
help_text = build_parser().format_help()
assert "quant-train" in help_text
assert "quant-validate" in help_text
assert "quantify" in help_text
assert "variant" in help_text
|