Skills › Scientific Agent Skills › pacsomatic › run_pacsomatic.py

run_pacsomatic.py

891 lines · skills/pacsomatic/scripts/run_pacsomatic.py at 5f2bc03

Open on GitHub

1"""Prepare, validate, and optionally launch nf-core/pacsomatic across platforms."""

3import argparse

4import csv

5import json

6import math

7import os

8import re

9import shlex

10import shutil

11import subprocess

12import sys

13from pathlib import Path

14from urllib.parse import urlparse

16SCHEDULER_CMDS = {

17 "lsf": "bsub",

18 "slurm": "sbatch",

19 "pbs": "qsub",

20 "sge": "qsub",

21}

23# Partial executable inventory only; container-internal entrypoints are not host dependencies.

24PACSOMATIC_HOST_TOOLS = [

25 "pbmm2",

26 "samtools",

27 "mosdepth",

28 "hiphase",

29 "cnvkit.py",

30 "vep",

31 "multiqc",

32]

34MIN_JAVA_MAJOR = 17

35MAX_JAVA_MAJOR = 26

36MIN_NEXTFLOW_VERSION = (24, 4, 2)

37REVIEWED_REVISION = "24c84cb371b0339c1d65a4de9451671945e19772"

40def fail(message):

41 print(f"[ERROR] {message}", file=sys.stderr)

42 raise SystemExit(1)

45def info(message):

46 print(f"[INFO] {message}")

49def warn(message):

50 print(f"[WARN] {message}")

53def is_remote_path(path):

54 parsed = urlparse(path)

55 return parsed.scheme in {"http", "https", "s3", "gs", "ftp"}

58def normalize_walltime_hhmmss(value, label):

59 if re.fullmatch(r"\d{1,3}:[0-5]\d(?::[0-5]\d)?", value):

60 parts = [int(part) for part in value.split(":")]

61 if len(parts) == 2:

62 parts.append(0)

63 if any(parts):

64 return f"{parts[0]:02}:{parts[1]:02}:{parts[2]:02}"

65 fail(f"Invalid {label} format: {value!r}. Use HH:MM or HH:MM:SS")

68def has_any_command(names):

69 return next((name for name in names if shutil.which(name) is not None), "")

72def detect_java_major_version():

73 java = os.environ.get("JAVA_CMD") or (str(Path(os.environ["JAVA_HOME"]) / "bin/java") if os.environ.get("JAVA_HOME") else "java")

74 if shutil.which(java) is None:

75 return None

76 completed = subprocess.run([java, "-version"], capture_output=True, text=True, timeout=30)

77 if completed.returncode:

78 return None

79 raw = f"{completed.stdout}\n{completed.stderr}"

80 match = re.search(r'version\s+"([0-9]+)(?:\.[0-9._]+)?"', raw)

81 if match:

82 return int(match.group(1))

83 return None

86def ensure_no_spaces(label, value):

87 if not value or any(ch.isspace() or ord(ch) < 32 for ch in value):

88 fail(f"{label} must be nonempty and contain no whitespace/control characters.")

91def validate_repo(repo_path):

92 repo_path = repo_path.expanduser().resolve()

93 if not repo_path.exists() or not repo_path.is_dir():

94 fail(f"Repository path does not exist or is not a directory: {repo_path}")

95 entry = repo_path / "main.nf"

96 if not entry.exists():

97 fail(f"Could not find main.nf in repository path: {repo_path}")

98 return repo_path

100

101def ensure_pipeline_repo(args):

102 if not args.repo_path and not args.checkout_dir:

103 return

104

105 if args.repo_path:

106 repo = validate_repo(Path(args.repo_path))

107 if args.pipeline_version:

108 fail("For --repo-path, check out the desired revision yourself and omit --pipeline-version; Nextflow -r is for remote projects.")

109 args.pipeline = str(repo)

110 info(f"Using local pipeline repository: {repo}")

111 return

112

113 if shutil.which("git") is None:

114 fail("git is required to clone pipeline repository")

115

116 checkout_dir = Path(args.checkout_dir).expanduser().resolve()

117 checkout_dir.mkdir(parents=True, exist_ok=True)

118 target = checkout_dir / args.repo_name

119 if not target.exists():

120 cmd = ["git", "clone", args.repo_url, str(target)]

121 info("Cloning repository: " + " ".join(cmd))

122 completed = subprocess.run(cmd)

123 if completed.returncode != 0:

124 fail(f"git clone failed with code {completed.returncode}")

125 if args.pipeline_version:

126 completed = subprocess.run(["git", "-C", str(target), "checkout", "--detach", args.pipeline_version])

127 if completed.returncode:

128 fail("Could not check out requested pipeline revision in the new clone.")

129 elif args.pipeline_version:

130 result = subprocess.run(["git", "-C", str(target), "rev-parse", "HEAD", args.pipeline_version], capture_output=True, text=True)

131 revisions = result.stdout.splitlines()

132 if result.returncode or len(revisions) != 2 or revisions[0] != revisions[1]:

133 fail("Existing checkout differs from --pipeline-version; select its revision explicitly outside this helper.")

134

135 repo = validate_repo(target)

136 args.pipeline = str(repo)

137 args.pipeline_version = ""

138 info(f"Using cloned pipeline repository: {repo}")

139

140

141def parse_env_list_output(raw_output, env_name):

142 for line in raw_output.splitlines():

143 line = line.strip()

144 if not line or line.startswith("Name") or line.startswith("#"):

145 continue

146 tokens = line.split()

147 path_token = next((token for token in reversed(tokens) if token.startswith("/")), None)

148 if path_token is None:

149 continue

150 prefix = Path(path_token)

151 if prefix.name == env_name:

152 return prefix

153 if tokens and tokens[0] == env_name:

154 return prefix

155 return None

156

157

158def find_conda_env_prefix(env_name):

159 commands = []

160 if shutil.which("mamba"):

161 commands.append(["mamba", "env", "list"])

162 if shutil.which("conda"):

163 commands.append(["conda", "env", "list"])

164

165 for cmd in commands:

166 completed = subprocess.run(cmd, capture_output=True, text=True)

167 combined = f"{completed.stdout}\n{completed.stderr}"

168 prefix = parse_env_list_output(combined, env_name)

169 if prefix is not None and prefix.exists():

170 return prefix

171

172 return None

173

174

175def create_conda_env(env_name, env_file):

176 env_file = Path(env_file).expanduser().resolve()

177 if not env_file.exists():

178 fail(f"Conda environment file does not exist: {env_file}")

179

180 if shutil.which("mamba"):

181 cmd = ["mamba", "env", "create", "-n", env_name, "-f", str(env_file)]

182 elif shutil.which("conda"):

183 cmd = ["conda", "env", "create", "-n", env_name, "-f", str(env_file)]

184 else:

185 fail("Neither mamba nor conda is available to create environment")

186

187 info("Creating conda environment: " + " ".join(cmd))

188 completed = subprocess.run(cmd)

189 if completed.returncode != 0:

190 fail(f"Environment creation failed with code {completed.returncode}")

191

192

193def resolve_runtime(args):

194 if args.use_current_path:

195 return None

196

197 prefix = find_conda_env_prefix(args.conda_env)

198 if prefix is None and args.create_conda_env:

199 if not args.conda_env_file:

200 fail("--create-conda-env requires --conda-env-file; no environment YAML is bundled.")

201 env_file = args.conda_env_file

202 create_conda_env(args.conda_env, env_file)

203 prefix = find_conda_env_prefix(args.conda_env)

204

205 if prefix is None:

206 if args.run:

207 fail(

208 f"Conda environment '{args.conda_env}' was not found. "

209 "Re-run with --create-conda-env or use --use-current-path."

210 )

211 warn(

212 f"Conda environment '{args.conda_env}' was not found. "

213 "Dry-run/generate-only continues; run mode requires a resolvable runtime."

214 )

215 return None

216

217 nextflow_bin = prefix / "bin" / "nextflow"

218 if nextflow_bin.exists():

219 args.nextflow_bin = str(nextflow_bin)

220 info(f"Using Nextflow from conda environment: {nextflow_bin}")

221 else:

222 warn(

223 f"Conda environment found at {prefix}, but nextflow was not found in {prefix / 'bin'}. "

224 "Falling back to --nextflow-bin PATH lookup."

225 )

226

227 return prefix

228

229

230def build_generated_params_content(args, samplesheet_path):

231 lines = [

232 f"input: {json.dumps(samplesheet_path)}",

233 f"outdir: {json.dumps(args.outdir)}",

234 ]

235 if args.fasta:

236 lines.append(f"fasta: {json.dumps(args.fasta)}")

237 elif args.genome:

238 lines.append(f"genome: {json.dumps(args.genome)}")

239 return "\n".join(lines) + "\n"

240

241

242def write_generated_params_file(args, samplesheet_path):

243 config_path = os.path.join(args.outdir, args.config_name)

244 os.makedirs(os.path.dirname(config_path) or ".", exist_ok=True)

245 with open(config_path, "w", encoding="utf-8") as handle:

246 handle.write(build_generated_params_content(args, samplesheet_path))

247 return config_path

248

249

250def build_samplesheet(args, samplesheet_path):

251 rows = [

252 {

253 "patient": args.patient_id,

254 "sample": args.tumor_sample_id,

255 "status": "1",

256 "bam": args.tumor_bam,

257 "pbi": args.tumor_pbi or "",

258 },

259 {

260 "patient": args.patient_id,

261 "sample": args.normal_sample_id,

262 "status": "0",

263 "bam": args.normal_bam,

264 "pbi": args.normal_pbi or "",

265 },

266 ]

267

268 with open(samplesheet_path, "w", newline="", encoding="utf-8") as handle:

269 writer = csv.DictWriter(handle, fieldnames=["patient", "sample", "status", "bam", "pbi"])

270 writer.writeheader()

271 writer.writerows(rows)

272

273

274def resolve_reference_args(args):

275 if args.fasta:

276 return ["--fasta", args.fasta]

277 if args.genome:

278 return ["--genome", args.genome]

279 raise ValueError("Either --fasta or --genome must be provided.")

280

281

282def build_nextflow_command(args, samplesheet_path):

283 cmd = [

284 args.nextflow_bin,

285 "run",

286 args.pipeline,

287 "-profile",

288 args.profile,

289 "--input",

290 samplesheet_path,

291 "--outdir",

292 args.outdir,

293 ]

294 cmd.extend(resolve_reference_args(args))

295

296 if args.pipeline_version:

297 cmd.extend(["-r", args.pipeline_version])

298 if args.nextflow_config:

299 cmd.extend(["-c", args.nextflow_config])

300

301 params_file = args.params_file

302 if not params_file and args.use_generated_params_file and args.generated_params_file:

303 params_file = args.generated_params_file

304 if params_file:

305 cmd.extend(["-params-file", params_file])

306

307 if args.resume:

308 cmd.append("-resume")

309

310 if args.with_report:

311 cmd.extend(["-with-report", args.with_report])

312

313 if args.with_dag:

314 cmd.extend(["-with-dag", args.with_dag])

315

316 if args.extra_args:

317 cmd.extend(shlex.split(args.extra_args))

318

319 return cmd

320

321

322def scheduler_header_lines(args):

323 mem_mb = math.ceil(args.memory_gb * 1024)

324 suffix = "%J" if args.executor == "lsf" else "%j" if args.executor == "slurm" else ""

325 stdout_path = os.path.join(args.logdir or args.outdir, args.stdout_file or f"launcher{suffix}.out")

326 stderr_path = os.path.join(args.logdir or args.outdir, args.stderr_file or f"launcher{suffix}.err")

327 walltime_hhmmss = normalize_walltime_hhmmss(args.walltime, "walltime")

328

329 if args.executor == "lsf":

330 lines = [

331 f"#BSUB -J {args.job_name}",

332 f"#BSUB -n {args.cpus}",

333 f"#BSUB -M {mem_mb}MB",

334 f"#BSUB -W {int(walltime_hhmmss.split(':')[0])}:{int(walltime_hhmmss.split(':')[1]):02}",

335 f"#BSUB -o {stdout_path}",

336 f"#BSUB -e {stderr_path}",

337 ]

338 if args.queue:

339 lines.insert(1, f"#BSUB -q {args.queue}")

340 if args.project:

341 lines.insert(1, f"#BSUB -P {args.project}")

342 return lines

343

344 if args.executor == "slurm":

345 lines = [

346 f"#SBATCH --job-name={args.job_name}",

347 f"#SBATCH --cpus-per-task={args.cpus}",

348 f"#SBATCH --mem={mem_mb}",

349 f"#SBATCH --time={walltime_hhmmss}",

350 f"#SBATCH --output={stdout_path}",

351 f"#SBATCH --error={stderr_path}",

352 ]

353 if args.queue:

354 lines.insert(1, f"#SBATCH --partition={args.queue}")

355 if args.project:

356 lines.insert(1, f"#SBATCH --account={args.project}")

357 return lines

358

359 if args.executor == "pbs":

360 lines = [

361 f"#PBS -N {args.job_name}",

362 f"#PBS -l select=1:ncpus={args.cpus}:mem={mem_mb}mb",

363 f"#PBS -l walltime={walltime_hhmmss}",

364 f"#PBS -o {stdout_path}",

365 f"#PBS -e {stderr_path}",

366 ]

367 if args.queue:

368 lines.insert(1, f"#PBS -q {args.queue}")

369 if args.project:

370 lines.insert(1, f"#PBS -A {args.project}")

371 return lines

372

373 if args.executor == "sge":

374 lines = [

375 f"#$ -N {args.job_name}",

376 f"#$ -pe smp {args.cpus}",

377 f"#$ -l h_vmem={math.ceil(mem_mb / args.cpus)}M",

378 f"#$ -l h_rt={walltime_hhmmss}",

379 f"#$ -o {stdout_path}",

380 f"#$ -e {stderr_path}",

381 ]

382 if args.queue:

383 lines.insert(1, f"#$ -q {args.queue}")

384 if args.project:

385 lines.insert(1, f"#$ -P {args.project}")

386 return lines

387

388 return []

389

390

391def default_script_path(args):

392 suffix = args.executor if args.executor in SCHEDULER_CMDS else "local"

393 return os.path.join(args.outdir, f"run_pacsomatic.{suffix}.sh")

394

395

396def write_launch_script(args, script_path, nextflow_cmd):

397 quoted_cmd = " ".join(shlex.quote(token) for token in nextflow_cmd)

398 mkdir_targets = [args.outdir, args.workdir]

399 if args.logdir:

400 mkdir_targets.append(args.logdir)

401

402 lines = ["#!/usr/bin/env bash"]

403 lines.extend(scheduler_header_lines(args))

404 lines.extend([

405 "",

406 "set -euo pipefail",

407 f"mkdir -p {' '.join(shlex.quote(path) for path in mkdir_targets)}",

408 f"cd {shlex.quote(args.outdir)}",

409 f"export NXF_WORK={shlex.quote(args.workdir)}",

410 ])

411

412 if args.nxf_opts:

413 lines.append(f"export NXF_OPTS={shlex.quote(args.nxf_opts)}")

414

415 if args.singularity_cache:

416 lines.append(f"export NXF_SINGULARITY_CACHEDIR={shlex.quote(args.singularity_cache)}")

417

418 if args.runtime_prefix:

419 lines.append(f"export PATH={shlex.quote(str(Path(args.runtime_prefix) / 'bin'))}:$PATH")

420 lines.append(f"export CONDA_PREFIX={shlex.quote(str(args.runtime_prefix))}")

421

422 lines.append("")

423 if args.module_load:

424 lines.append(args.module_load)

425 lines.append(quoted_cmd)

426

427 with open(script_path, "w", encoding="utf-8") as handle:

428 handle.write("\n".join(lines) + "\n")

429

430 os.chmod(script_path, 0o755)

431

432

433def verify_bam_and_index(label, bam_path, pbi_path):

434 ensure_no_spaces(f"{label} BAM", bam_path)

435 if not bam_path.endswith(".bam"):

436 fail(f"{label} input must end in .bam; this helper targets unaligned PacBio HiFi BAM, not CRAM.")

437 if pbi_path:

438 ensure_no_spaces(f"{label} PBI", pbi_path)

439 if not pbi_path.endswith(".pbi"):

440 fail(f"{label} PacBio index must end in .pbi; BAI/CSI are different index formats.")

441 if not is_remote_path(pbi_path):

442 require_file(f"{label} PBI", pbi_path)

443 if is_remote_path(bam_path):

444 info(f"{label} BAM is remote; accessibility and content were not verified.")

445 return

446 require_file(f"{label} BAM", bam_path)

447

448

449def require_file(label, path):

450 if not Path(path).is_file() or Path(path).stat().st_size == 0:

451 fail(f"{label} must be a nonempty regular file: {path}")

452

453

454#: Characters that would let --module-load smuggle something other than a module

455#: command into the generated launch script. Everything else written into that

456#: script goes through shlex.quote(); this argument is emitted as a bare shell

457#: line, so it is constrained here instead.

458_SHELL_METACHARACTERS = set("$`|><&(){}[]*?!~\n\r\\\"'")

459

460

461def normalize_module_load(raw):

462 """Validate --module-load and return it as a safe shell line.

463

464 Accepts one or more `module ...` commands separated by `&&` or `;` -- the

465 documented shape, e.g. "module purge && module load nextflow/26.04.6".

466 Rejects anything else so a caller-supplied string cannot become arbitrary

467 shell in the launch script the operator later executes.

468 """

469 if not raw or not raw.strip():

470 return ""

471

472 segments = [seg.strip() for seg in re.split(r"&&|;", raw) if seg.strip()]

473 if not segments:

474 fail("--module-load contained no command.")

475

476 normalized = []

477 for segment in segments:

478 if any(ch in _SHELL_METACHARACTERS for ch in segment):

479 fail(

480 f"--module-load segment {segment!r} contains shell metacharacters. "

481 "Only plain 'module ...' commands are accepted."

482 )

483 try:

484 tokens = shlex.split(segment)

485 except ValueError as exc:

486 fail(f"--module-load segment {segment!r} could not be parsed: {exc}")

487 if not tokens or tokens[0] != "module":

488 fail(

489 f"--module-load segment {segment!r} does not start with 'module'. "

490 "Pass module commands only, e.g. 'module load nextflow/26.04.6'."

491 )

492 normalized.append(" ".join(shlex.quote(token) for token in tokens))

493

494 return " && ".join(normalized)

495

496

497def validate_inputs(args):

498 if args.dry_run and (args.run or args.submit):

499 fail("--dry-run cannot be combined with --run or --submit.")

500 if args.dry_run and (args.create_conda_env or args.checkout_dir):

501 fail("--dry-run does not clone repositories or create environments; prepare these separately.")

502 args.module_load = normalize_module_load(args.module_load)

503 ensure_no_spaces("patient-id", args.patient_id)

504 ensure_no_spaces("tumor-sample-id", args.tumor_sample_id)

505 ensure_no_spaces("normal-sample-id", args.normal_sample_id)

506 if args.tumor_sample_id == args.normal_sample_id:

507 fail("tumor-sample-id and normal-sample-id must be different.")

508

509 if not args.fasta and not args.genome:

510 fail("Either --fasta or --genome must be provided.")

511

512 if args.fasta and args.genome:

513 fail("Choose exactly one of --fasta or --genome.")

514 if args.cpus < 1 or not math.isfinite(args.memory_gb * 1024) or args.memory_gb <= 0:

515 fail("--cpus must be positive and --memory-gb must be finite and positive.")

516 walltime = normalize_walltime_hhmmss(args.walltime, "walltime")

517 if args.executor == "lsf" and not walltime.endswith(":00"):

518 fail("LSF walltime has minute precision; provide HH:MM or HH:MM:00.")

519 if args.executor in SCHEDULER_CMDS:

520 for label in ("job_name", "project", "queue", "outdir", "logdir", "stdout_file", "stderr_file"):

521 value = getattr(args, label)

522 if value and not re.fullmatch(r"[A-Za-z0-9_./%+,:@=-]+", value):

523 fail(f"--{label.replace('_', '-')} has characters unsupported in scheduler directives.")

524 try:

525 extra = shlex.split(args.extra_args)

526 except ValueError as exc:

527 fail(f"Invalid --extra-args quoting: {exc}")

528 managed = {"--input", "--outdir", "--genome", "--fasta", "-r", "-profile", "-params-file", "-c", "-work-dir", "-w"}

529 if any(token.split("=", 1)[0] in managed for token in extra):

530 fail("--extra-args cannot override managed input, reference, revision, config, profile or work paths.")

531

532 verify_bam_and_index("tumor", args.tumor_bam, args.tumor_pbi)

533 verify_bam_and_index("normal", args.normal_bam, args.normal_pbi)

534 if args.tumor_bam == args.normal_bam or (

535 not is_remote_path(args.tumor_bam) and not is_remote_path(args.normal_bam)

536 and os.path.samefile(args.tumor_bam, args.normal_bam)

537 ):

538 fail("Tumor and normal BAMs must be different files.")

539

540 if args.fasta:

541 if not re.fullmatch(r"\S+\.fn?a(sta)?(\.gz)?", args.fasta):

542 fail("FASTA path must match upstream .fa/.fna/.fasta, optionally .gz, without whitespace.")

543 if not is_remote_path(args.fasta):

544 require_file("Reference FASTA", args.fasta)

545 for label in ("params_file", "nextflow_config"):

546 if getattr(args, label):

547 require_file(label, getattr(args, label))

548 if args.genome:

549 ensure_no_spaces("genome", args.genome)

550

551

552def ensure_runtime_tools(args):

553 nextflow_missing = shutil.which(args.nextflow_bin) is None

554 if args.run and nextflow_missing:

555 fail(f"Could not find Nextflow executable: {args.nextflow_bin}")

556 if not args.run and nextflow_missing:

557 warn(

558 f"Nextflow executable not found ({args.nextflow_bin}). "

559 "Dry-run and generate-only modes can continue without Nextflow."

560 )

561 if not nextflow_missing:

562 result = subprocess.run([args.nextflow_bin, "-version"], capture_output=True, text=True, timeout=60)

563 match = re.search(r"version\s+(\d+)\.(\d+)\.(\d+)", result.stdout + result.stderr)

564 if result.returncode or not match:

565 if args.run:

566 fail("Nextflow -version failed or could not be parsed in the selected runtime.")

567 warn("Nextflow version was not verified.")

568 elif tuple(map(int, match.groups())) < MIN_NEXTFLOW_VERSION:

569 if args.run:

570 fail("Reviewed pacsomatic revision requires Nextflow >=24.04.2.")

571 warn("Nextflow is older than the reviewed pipeline's minimum 24.04.2.")

572 else:

573 info(f"Nextflow version: {'.'.join(match.groups())}")

574

575 if args.run and args.executor in SCHEDULER_CMDS:

576 submit_cmd = SCHEDULER_CMDS[args.executor]

577 if shutil.which(submit_cmd) is None:

578 fail(f"{submit_cmd} is required for --run with --executor {args.executor}")

579

580

581def ensure_dependency_tools(args):

582 profile_items = [item.strip().lower() for item in args.profile.split(",") if item.strip()]

583 missing_run = []

584 missing_warn = []

585 missing_host = []

586

587 java_major = detect_java_major_version()

588 if java_major is None:

589 if args.run:

590 missing_run.append("java")

591 else:

592 missing_warn.append("java")

593 elif not MIN_JAVA_MAJOR <= java_major <= MAX_JAVA_MAJOR:

594 if args.run:

595 missing_run.append(f"java {MIN_JAVA_MAJOR}-{MAX_JAVA_MAJOR}")

596 else:

597 missing_warn.append(f"java {MIN_JAVA_MAJOR}-{MAX_JAVA_MAJOR}")

598

599 if "docker" in profile_items and not has_any_command(["docker"]):

600 if args.run:

601 missing_run.append("docker")

602 else:

603 missing_warn.append("docker")

604

605 for profile, executable in (("singularity", "singularity"), ("apptainer", "apptainer"), ("podman", "podman"), ("shifter", "shifter"), ("charliecloud", "ch-run"), ("mamba", "mamba")):

606 if profile in profile_items and not has_any_command([executable]):

607 if args.run:

608 missing_run.append(executable)

609 else:

610 missing_warn.append(executable)

611

612 if "conda" in profile_items and not has_any_command(["conda", "mamba"]):

613 if args.run:

614 missing_run.append("conda|mamba")

615 else:

616 missing_warn.append("conda|mamba")

617

618 # No upstream profile named local; task executor and runtime profile are separate.

619 if args.check_host_bio_tools:

620 for tool in PACSOMATIC_HOST_TOOLS:

621 if shutil.which(tool) is None:

622 missing_host.append(tool)

623

624 if missing_host:

625 msg = (

626 "Missing host bioinformatics tools requested by --check-host-bio-tools: "

627 + ", ".join(sorted(set(missing_host)))

628 )

629 if args.run and args.strict_host_bio_tools:

630 fail(msg)

631 warn(msg)

632

633 if missing_warn:

634 warn(

635 "Potentially missing dependency tools for selected profile/runtime: "

636 + ", ".join(sorted(set(missing_warn)))

637 + ". Dry-run/generate-only continues."

638 )

639

640 if missing_run:

641 fail(

642 "Missing dependency tools for selected profile/runtime: "

643 + ", ".join(sorted(set(missing_run)))

644 )

645

646

647def submit_command_for_executor(executor, script_path):

648 """Return (argv, stdin_path) for submitting the launch script.

649

650 lsf reads the script from stdin (`bsub < script`); every other executor

651 takes it as an argument. Returning an argv list rather than a shell string

652 keeps submission off a shell entirely, so a script path containing shell

653 metacharacters cannot extend the command that runs.

654 """

655 if executor == "lsf":

656 return ["bsub"], script_path

657 if executor == "slurm":

658 return ["sbatch", script_path], None

659 if executor in {"pbs", "sge"}:

660 return ["qsub", script_path], None

661 return ["bash", script_path], None

662

663

664def format_submit_command(argv, stdin_path):

665 """Render a submission as a copy-pasteable shell line, for display only."""

666 line = shlex.join(argv)

667 if stdin_path:

668 line += f" < {shlex.quote(stdin_path)}"

669 return line

670

671

672def extract_job_id(executor, output):

673 patterns = {

674 "lsf": r"<([0-9]+)>",

675 "slurm": r"Submitted batch job\s+([0-9]+)",

676 "pbs": r"^([0-9]+(?:\.[A-Za-z0-9_.-]+)?)$",

677 "sge": r"job\s+([0-9]+)",

678 }

679 pattern = patterns.get(executor)

680 if not pattern:

681 return None

682

683 for line in output.splitlines():

684 match = re.search(pattern, line.strip())

685 if match:

686 return match.group(1)

687 return None

688

689

690def execute_launch(args, script_path):

691 argv, stdin_path = submit_command_for_executor(args.executor, script_path)

692 if args.executor not in SCHEDULER_CMDS:

693 completed = subprocess.run(argv)

694 if completed.returncode:

695 fail(f"Execution failed with exit code {completed.returncode}")

696 info("Local launcher exited successfully; review pipeline completion and QC outputs.")

697 return

698 if stdin_path:

699 with open(stdin_path, "rb") as handle:

700 completed = subprocess.run(argv, stdin=handle, text=True, capture_output=True)

701 else:

702 completed = subprocess.run(argv, text=True, capture_output=True)

703 stdout = (completed.stdout or "").strip()

704 stderr = (completed.stderr or "").strip()

705

706 if completed.returncode != 0:

707 if stdout:

708 print(stdout)

709 if stderr:

710 print(stderr, file=sys.stderr)

711 fail(f"Execution failed with exit code {completed.returncode}")

712

713 print("--- Scheduler Accepted Submission (pipeline completion not verified) ---")

714 if stdout:

715 print(stdout)

716 if stderr:

717 print(stderr, file=sys.stderr)

718

719 job_id = extract_job_id(args.executor, stdout)

720 if job_id:

721 info(f"Detected job id: {job_id}")

722

723

724def parse_args():

725 parser = argparse.ArgumentParser(

726 description="Prepare and optionally launch nf-core/pacsomatic from matched tumor/normal BAMs"

727 )

728

729 parser.add_argument("--tumor-bam", required=True, help="Tumor BAM path")

730 parser.add_argument("--normal-bam", required=True, help="Normal BAM path")

731 parser.add_argument("--tumor-pbi", default="", help="Tumor BAM index (.pbi) path, optional")

732 parser.add_argument("--normal-pbi", default="", help="Normal BAM index (.pbi) path, optional")

733 parser.add_argument("--patient-id", required=True, help="Patient ID for samplesheet")

734 parser.add_argument("--tumor-sample-id", required=True, help="Tumor sample ID")

735 parser.add_argument("--normal-sample-id", required=True, help="Normal sample ID")

736

737 parser.add_argument("--fasta", default="", help="Reference FASTA path")

738 parser.add_argument("--genome", default="", help="Reference genome key, e.g. GRCh38")

739 parser.add_argument("--outdir", required=True, help="Output directory")

740 parser.add_argument("--workdir", default="", help="Nextflow work directory, default: <outdir>/work")

741 parser.add_argument("--logdir", default="", help="Optional scheduler logs directory; if unset use outdir")

742 parser.add_argument("--stdout-file", default="", help="Scheduler stdout filename; default uses %%J for LSF, %%j for Slurm")

743 parser.add_argument("--stderr-file", default="", help="Scheduler stderr filename; default uses %%J for LSF, %%j for Slurm")

744 parser.add_argument("--samplesheet", default="", help="Samplesheet path, default: <outdir>/samplesheet.csv")

745 parser.add_argument("--script-path", default="", help="Launch script path, default: <outdir>/run_pacsomatic.<executor>.sh")

746

747 parser.add_argument("--pipeline", default="nf-core/pacsomatic", help="Pipeline name or repo")

748 parser.add_argument("--repo-path", default="", help="Optional local pipeline repository path (contains main.nf)")

749 parser.add_argument("--repo-url", default="https://github.com/nf-core/pacsomatic.git", help="Pipeline repository URL when cloning")

750 parser.add_argument("--checkout-dir", default="", help="Directory where pipeline repo should be cloned")

751 parser.add_argument("--repo-name", default="pacsomatic", help="Folder name inside checkout-dir for cloned repo")

752 parser.add_argument("--pipeline-version", default="", help="Remote revision for -r; defaults to the reviewed pacsomatic commit")

753 parser.add_argument("--nextflow-bin", default="nextflow", help="Nextflow executable path")

754 parser.add_argument("--profile", default="singularity", help="Nextflow profile, e.g. singularity or singularity,institute")

755 parser.add_argument("--params-file", default="", help="Path to Nextflow -params-file (yaml/json)")

756 parser.add_argument("--nextflow-config", default="", help="Infrastructure/process config passed via -c; not a pipeline params file")

757 parser.add_argument("--overwrite", action="store_true", help="Replace existing helper artifacts after review (never input files)")

758 parser.add_argument("--config-name", default="pacsomatic.params.generated.yaml", help="Generated params YAML filename under outdir")

759 parser.add_argument("--use-generated-params-file", action="store_true", help="Use generated params YAML via -params-file when params-file is not provided")

760 parser.add_argument("--resume", action="store_true", help="Add -resume to Nextflow command")

761 parser.add_argument("--with-report", default="", help="Path for Nextflow -with-report output")

762 parser.add_argument("--with-dag", default="", help="Path for Nextflow -with-dag output")

763 parser.add_argument("--extra-args", default="", help="Extra raw args appended to Nextflow command")

764

765 parser.add_argument(

766 "--executor",

767 default="local",

768 choices=["local", "none", "lsf", "slurm", "pbs", "sge"],

769 help="Outer launcher backend only; task executor is set separately by profile or --nextflow-config (pbs uses PBS Pro headers)",

770 )

771 parser.add_argument("--job-name", default="pacsomatic", help="Scheduler job name")

772 parser.add_argument("--project", default="", help="Scheduler project/account when supported")

773 parser.add_argument("--queue", default="", help="Scheduler queue/partition when supported")

774 parser.add_argument("--cpus", type=int, default=16, help="Driver CPU slots/threads (not pipeline task resources)")

775 parser.add_argument("--memory-gb", type=float, default=64.0, help="Driver memory in GiB (not pipeline task resources)")

776 parser.add_argument("--walltime", default="48:00", help="Requested walltime in HH:MM or HH:MM:SS")

777 parser.add_argument("--nxf-opts", default="", help="Optional NXF_OPTS, e.g. '-Xms1g -Xmx4g'")

778 parser.add_argument("--singularity-cache", default="", help="Optional NXF_SINGULARITY_CACHEDIR")

779 parser.add_argument("--conda-env", default="pacsomatic-nextflow", help="Conda environment name used to resolve Nextflow runtime")

780 parser.add_argument("--conda-env-file", default="", help="Conda environment YAML used with --create-conda-env")

781 parser.add_argument("--create-conda-env", action="store_true", help="Create conda environment when missing")

782 parser.add_argument("--use-current-path", action="store_true", help="Use current PATH and skip conda runtime resolution")

783 parser.add_argument(

784 "--module-load",

785 default="",

786 help="Optional module command prefix, e.g. 'module load nextflow/26.04.6'",

787 )

788

789 parser.add_argument(

790 "--check-host-bio-tools",

791 action="store_true",

792 help="Check pacsomatic host bioinformatics tools on PATH (mainly for non-container local runs)",

793 )

794 parser.add_argument(

795 "--strict-host-bio-tools",

796 action="store_true",

797 help="Fail in --run mode if --check-host-bio-tools finds missing host tools",

798 )

799

800 parser.add_argument("--dry-run", action="store_true", help="Validate inputs and write artifacts without execution")

801 parser.add_argument("--run", action="store_true", help="Execute or submit the generated launch script")

802 parser.add_argument("--submit", action="store_true", help="Backward-compatible alias for --run")

803

804 return parser.parse_args()

805

806

807def normalize_paths(args):

808 for field in ("tumor_bam", "normal_bam", "tumor_pbi", "normal_pbi", "fasta", "outdir", "workdir", "logdir", "samplesheet", "script_path", "params_file", "nextflow_config", "with_report", "with_dag", "singularity_cache"):

809 value = getattr(args, field)

810 if value and not is_remote_path(value):

811 setattr(args, field, str(Path(value).expanduser().resolve()))

812 for field in ("outdir", "workdir", "logdir", "samplesheet", "script_path", "params_file", "nextflow_config"):

813 if is_remote_path(getattr(args, field)):

814 fail(f"--{field.replace('_', '-')} must be local for this helper.")

815 if not args.outdir:

816 fail("--outdir must be nonempty.")

817 args.workdir = args.workdir or str(Path(args.outdir) / "work")

818

819

820def validate_artifact_targets(args, samplesheet_path, launch_script_path):

821 artifacts = [Path(samplesheet_path).resolve(), Path(launch_script_path).resolve(), Path(args.outdir, args.config_name).resolve()]

822 if Path(args.config_name).name != args.config_name or not args.config_name:

823 fail("--config-name must be a filename under outdir.")

824 inputs = [Path(getattr(args, key)).resolve() for key in ("tumor_bam", "normal_bam", "tumor_pbi", "normal_pbi", "fasta", "params_file", "nextflow_config") if getattr(args, key) and not is_remote_path(getattr(args, key))]

825 if len(set(artifacts)) != len(artifacts):

826 fail("Helper artifact paths must be distinct.")

827 for path in artifacts:

828 if path in inputs or any(path.exists() and item.exists() and os.path.samefile(path, item) for item in inputs):

829 fail("Helper artifact path aliases an input file.")

830 if path.exists() and (not args.overwrite or not path.is_file()):

831 fail(f"Artifact exists: {path}. Review it, then use --overwrite to replace helper artifacts.")

832

833

834def main():

835 args = parse_args()

836