(
run_dir: Path,
sample: str,
input_vcf: Path,
annotation_vcf: Path | None,
annotation_columns: str | None,
)
| 266 | |
| 267 | |
| 268 | def annotate_vcf( |
| 269 | run_dir: Path, |
| 270 | sample: str, |
| 271 | input_vcf: Path, |
| 272 | annotation_vcf: Path | None, |
| 273 | annotation_columns: str | None, |
| 274 | ) -> tuple[Path, dict[str, Any]]: |
| 275 | if not annotation_vcf: |
| 276 | return input_vcf, {"ok": True, "changed": False, "reason": "annotation not requested"} |
| 277 | columns = detect_annotation_columns(run_dir, annotation_vcf, annotation_columns) |
| 278 | output_vcf = input_vcf.parent / f"{input_vcf.name.removesuffix('.vcf.gz')}.annotated.vcf.gz" |
| 279 | annotate = run_cmd( |
| 280 | [ |
| 281 | "bcftools", |
| 282 | "annotate", |
| 283 | "-a", |
| 284 | str(annotation_vcf), |
| 285 | "-c", |
| 286 | columns, |
| 287 | "--pair-logic", |
| 288 | "exact", |
| 289 | "-O", |
| 290 | "z", |
| 291 | "-o", |
| 292 | str(output_vcf), |
| 293 | str(input_vcf), |
| 294 | ], |
| 295 | run_dir, |
| 296 | timeout=3600, |
| 297 | ) |
| 298 | result: dict[str, Any] = { |
| 299 | "ok": bool(annotate.get("ok")), |
| 300 | "changed": bool(annotate.get("ok")), |
| 301 | "reason": "annotated from resource VCF" |
| 302 | if annotate.get("ok") |
| 303 | else "bcftools annotate failed", |
| 304 | "annotation_vcf": str(annotation_vcf), |
| 305 | "annotation_columns": columns, |
| 306 | "annotate": annotate, |
| 307 | } |
| 308 | return (output_vcf if annotate.get("ok") else input_vcf), result |
| 309 | |
| 310 | |
| 311 | def normalize_mq_header(run_dir: Path, sample: str, vcf: Path) -> dict[str, Any]: |
no test coverage detected