Function mutect2_plan

plugins/ngs-analysis/scripts/run_dna_somatic_variants.py:261–430

(args: argparse.Namespace, pairs: list[dict[str, str]])

def mutect2_plan(args: argparse.Namespace, pairs: list[dict[str, str]]) -> list[dict[str, Any]]:
    reference = args.reference_fasta.expanduser().resolve()
    commands: list[dict[str, Any]] = []
    for pair in pairs:
        pair_id = pair["pair_id"]
        tumor_bam = Path(pair["tumor_alignment"])
        unfiltered = f"variants/{pair_id}.unfiltered.vcf.gz"
        filtered = f"variants/{pair_id}.filtered.vcf.gz"
        cmd: list[str | Path] = [
            "gatk",
            "Mutect2",
            "-R",
            reference,
            "-I",
            tumor_bam,
            "-tumor",
            pair["tumor_sample"],
        ]
        if pair["normal_alignment"]:
            cmd.extend(["-I", Path(pair["normal_alignment"]), "-normal", pair["normal_sample"]])
        if args.germline_resource:
            cmd.extend(["--germline-resource", args.germline_resource.expanduser().resolve()])
        if args.panel_of_normals:
            cmd.extend(["-pon", args.panel_of_normals.expanduser().resolve()])
        if args.target_bed:
            cmd.extend(["-L", args.target_bed.expanduser().resolve()])
        if args.f1r2_orientation_model:
            cmd.extend(["--f1r2-tar-gz", f"f1r2/{pair_id}.f1r2.tar.gz"])
        cmd.extend(["-O", unfiltered])
        commands.append(command_plan_entry(f"{pair_id}: mutect2", cmd, outputs=[unfiltered]))
        if args.f1r2_orientation_model:
            commands.append(
                command_plan_entry(
                    f"{pair_id}: learn read orientation model",
                    [
                        "gatk",
                        "LearnReadOrientationModel",
                        "-I",
                        f"f1r2/{pair_id}.f1r2.tar.gz",
                        "-O",
                        f"f1r2/{pair_id}.read-orientation-model.tar.gz",
                    ],
                    outputs=[f"f1r2/{pair_id}.read-orientation-model.tar.gz"],
                )
            )
        contamination_args: list[str | Path] = []
        if args.germline_resource:
            pileup_intervals = (
                args.target_bed.expanduser().resolve()
                if args.target_bed
                else args.germline_resource.expanduser().resolve()
            )
            tumor_pileups = f"qc/{pair_id}.tumor.pileups.table"
            commands.append(
                command_plan_entry(
                    f"{pair_id}: tumor pileup summaries",
                    [
                        "gatk",