diff --git a/bioflow_kernel/stage_image_map.json b/bioflow_kernel/stage_image_map.json index a988e02..330b63d 100644 --- a/bioflow_kernel/stage_image_map.json +++ b/bioflow_kernel/stage_image_map.json @@ -388,68 +388,218 @@ "image": "bioflow-rnaR", "module": "cli.run_expression", "executable": true, - "note": "跨环境: 主 rnaR(R 绘图); 内部 rnaseq_quant 矩阵触发走 rnaseq 镜像子阶段(编排层拆分)" + "note": "跨环境: 主 rnaR(R 绘图); 内部 rnaseq_quant 矩阵触发走 rnaseq 镜像子阶段(编排层拆分)", + "cli_contract": [ + [ + "--workspace", + "workspace" + ], + [ + "--prefix", + "prefix" + ] + ] }, "rnaseq_qc": { "image": "bioflow-rnaseq", "module": "cli.run_rnaseq_qc", - "executable": true + "executable": true, + "cli_contract": [ + [ + "--input", + "input_dir" + ], + [ + "--config", + "samples_yaml" + ], + [ + "--output", + "output_final" + ] + ] }, "rnaseq_align": { "image": "bioflow-rnaseq", "module": "cli.run_rnaseq_align", - "executable": true + "executable": true, + "cli_contract": [ + [ + "--input", + "input_dir" + ], + [ + "--hisat2-index", + "param:hisat2_index" + ], + [ + "--output", + "output_final" + ] + ] }, "rnaseq_quant": { "image": "bioflow-rnaseq", "module": "cli.run_rnaseq_quant", - "executable": true + "executable": true, + "cli_contract": [ + [ + "--input", + "input_dir" + ], + [ + "--gtf", + "param:gtf" + ], + [ + "--output", + "output_final" + ] + ] }, "rnaseq_sample_qc": { "image": "bioflow-rnaseq", "module": "cli.run_rnaseq_sample_qc", - "executable": true + "executable": true, + "cli_contract": [ + [ + "--workspace", + "workspace" + ] + ] }, "rnaseq_wgcna": { "image": "bioflow-rnaseq", "module": "cli.run_rnaseq_wgcna", - "executable": true + "executable": true, + "cli_contract": [ + [ + "--workspace", + "workspace" + ] + ] }, "rnaseq_de": { "image": "bioflow-rnaseq", "module": "cli.run_rnaseq_de", "executable": true, - "note": "跨环境: DE 表在 rnaseq; DESeq2 R 统计/绘图可拆分 r_de_analysis 镜像子阶段(编排层)" + "note": "跨环境: DE 表在 rnaseq; DESeq2 R 统计/绘图可拆分 r_de_analysis 镜像子阶段(编排层)", + "cli_contract": [ + [ + "--input", + "input_dir" + ], + [ + "--config", + "samples_yaml" + ], + [ + "--output", + "output_final" + ] + ] }, "rnaseq_go_enrich": { "image": "bioflow-rnaseq", "module": "cli.run_rnaseq_enrich", - "executable": true + "executable": true, + "cli_contract": [ + [ + "--input", + "input_dir" + ], + [ + "--output", + "output_final" + ], + [ + "--prefix", + "prefix" + ] + ] }, "rnaseq_kegg_enrich": { "image": "bioflow-rnaseq", "module": "cli.run_rnaseq_enrich", - "executable": true + "executable": true, + "cli_contract": [ + [ + "--input", + "input_dir" + ], + [ + "--output", + "output_final" + ], + [ + "--prefix", + "prefix" + ] + ] }, "rnaseq_fgsea": { "image": "bioflow-rnaseq", "module": "cli.run_rnaseq_fgsea", - "executable": true + "executable": true, + "cli_contract": [ + [ + "--workspace", + "workspace" + ] + ] }, "rnaseq_candidate_family": { "image": "bioflow-rnaseq", "module": "cli.run_rnaseq_candidate_family", - "executable": true + "executable": true, + "cli_contract": [ + [ + "--workspace", + "workspace" + ], + [ + "--prefix", + "prefix" + ] + ] }, "rnaseq_report": { "image": "bioflow-rnaseq", "module": "cli.run_rnaseq_report", - "executable": true + "executable": true, + "cli_contract": [ + [ + "--workspace", + "workspace" + ], + [ + "--prefix", + "prefix" + ] + ] }, "rnaseq_enrich": { "image": "bioflow-rnaseq", "module": "cli.run_rnaseq_enrich_v2", - "executable": true + "executable": true, + "cli_contract": [ + [ + "--input", + "input_dir" + ], + [ + "--output", + "output_final" + ], + [ + "--prefix", + "prefix" + ], + [ + "--job-id", + "job_id" + ] + ] }, "enrichment": { "image": "bioflow-rnaseq", diff --git a/bioflow_service/app/routers/biojobs.py b/bioflow_service/app/routers/biojobs.py index 26d925a..d12b9e4 100644 --- a/bioflow_service/app/routers/biojobs.py +++ b/bioflow_service/app/routers/biojobs.py @@ -281,6 +281,13 @@ async def _register_path(record: Any, server_path: str, settings: Any) -> dict[s link.symlink_to(p) except OSError as exc: # noqa: BLE001 logger.warning("登记 symlink 失败（不阻断）: %s -> %s (%s)", p, link, exc) + # rnaseq 系阶段需要 samples.yaml（CLI -c 必填）：从 fastq 自动配对生成（不覆盖手工配置） + try: + from ..services.bioflow.runner import ensure_samples_yaml + if ensure_samples_yaml(input_dir): + logger.info("已自动生成 samples.yaml（fastq 配对）: %s", input_dir) + except Exception as exc: # noqa: BLE001 + logger.warning("samples.yaml 自动生成失败（不阻断登记）: %s", exc) registered = [str(p) for p in files] try: _persist_registered_paths(record, registered) diff --git a/bioflow_service/app/services/bioflow/catalog.py b/bioflow_service/app/services/bioflow/catalog.py index db87890..3458104 100644 --- a/bioflow_service/app/services/bioflow/catalog.py +++ b/bioflow_service/app/services/bioflow/catalog.py @@ -343,12 +343,16 @@ _RNASEQ_ALIGN_PARAMS = [ "help": "样本列表（逗号分隔）"}, {"name": "ref_genome", "type": "string", "default": "", "source": "user", "help": "参考基因组路径/索引名"}, + {"name": "hisat2_index", "type": "string", "default": "", "source": "user", + "help": "HISAT2 索引前缀（容器内可访问路径，如 /bioflow-data/hmm/hisat2/xxx）"}, {"name": "cpu", "type": "choice", "choices": ["4", "8", "16", "32"], "default": "8", "source": "runtime", "help": "容器 CPU 数"}, ] _RNASEQ_QUANT_PARAMS = [ {"name": "samples", "type": "string", "default": "", "source": "user", "help": "样本列表（逗号分隔）"}, + {"name": "gtf", "type": "string", "default": "", "source": "user", + "help": "参考基因组 GTF 注释（容器内可访问路径）"}, {"name": "cpu", "type": "choice", "choices": ["4", "8", "16", "32"], "default": "6", "source": "runtime", "help": "容器 CPU 数"}, ] @@ -361,6 +365,8 @@ _RNASEQ_SAMPLE_QC_PARAMS = [ _RNASEQ_DE_PARAMS = [ {"name": "design", "type": "string", "default": "", "source": "user", "help": "实验设计矩阵文件（对比组声明）"}, + {"name": "prefix", "type": "string", "default": "", "source": "contract", + "help": "分析前缀"}, {"name": "cpu", "type": "choice", "choices": ["4", "8", "16", "32"], "default": "4", "source": "runtime", "help": "容器 CPU 数"}, ] @@ -373,6 +379,8 @@ _RNASEQ_WGCNA_PARAMS = [ _RNASEQ_ENRICH_PARAMS = [ {"name": "gene_list", "type": "string", "default": "", "source": "contract", "help": "差异基因列表（上游 DE 结果）"}, + {"name": "prefix", "type": "string", "default": "", "source": "contract", + "help": "分析前缀"}, {"name": "cpu", "type": "choice", "choices": ["4", "8", "16", "32"], "default": "4", "source": "runtime", "help": "容器 CPU 数"}, ] diff --git a/bioflow_service/app/services/bioflow/kernel.py b/bioflow_service/app/services/bioflow/kernel.py index b10bca2..a8d8cd9 100644 --- a/bioflow_service/app/services/bioflow/kernel.py +++ b/bioflow_service/app/services/bioflow/kernel.py @@ -165,9 +165,13 @@ class Kernel: cmd.append(f"{tag}:{self.cfg.docker_tag_suffix}") cmd.extend(CONTAINER_CMD_PREFIX) cmd.extend(["-m", mod]) - # 通用容器内参数：input-dir/workspace/cpu（各入口 argparse 兼容） - cmd += ["--input-dir", self.cfg.container_input_dir] - cmd += ["--workspace", self.cfg.container_workspace_dir] + # 参数注入：有 cli_contract 的阶段按契约，否则沿用默认 --input-dir/--workspace + contract = entry.get("cli_contract") + if contract: + cmd += self._contract_args(contract, params, job_input_dir) + else: + cmd += ["--input-dir", self.cfg.container_input_dir] + cmd += ["--workspace", self.cfg.container_workspace_dir] for key in sorted(params): value = params[key] if key in ("env", "cpu", "memory"): @@ -185,6 +189,46 @@ class Kernel: cmd.append(str(value)) return cmd + def _contract_args(self, contract: list[Any], params: dict[str, Any], + job_input_dir: str | Path) -> list[str]: + """cli_contract 解析：[(flag, source), ...] → argv。 + + source 取值： + - input_dir/workspace：容器内 /job/input、/job/workspace + - output_final：/job/workspace/final（rnaseq 系 -o 语义：产物目录, + CLI 内部以 output.parent 为 workspace 并建 output.parent/final—— + 故 -o 必须指向 workspace/final,manifest 也落此处与 runner 检索对齐） + - samples_yaml：/job/input/samples.yaml(宿主侧不存在则跳过该 flag, + 交由 CLI 报缺失,错误信息比服务端更可读) + - prefix/job_id：从 params 取值,缺失跳过 + - param:：从 params[name] 取值,缺失跳过(用户经确认表单提供) + """ + args: list[str] = [] + for flag, source in contract: + if source == "input_dir": + args += [flag, self.cfg.container_input_dir] + elif source == "workspace": + args += [flag, self.cfg.container_workspace_dir] + elif source == "output_final": + args += [flag, f"{self.cfg.container_workspace_dir}/final"] + elif source == "samples_yaml": + if (Path(job_input_dir) / "samples.yaml").is_file(): + args += [flag, f"{self.cfg.container_input_dir}/samples.yaml"] + elif source == "prefix": + value = str(params.get("prefix") or "").strip() + if value: + args += [flag, value] + elif source == "job_id": + value = str(params.get("job_id") or "").strip() + if value: + args += [flag, value] + elif str(source).startswith("param:"): + key = str(source).split(":", 1)[1] + value = params.get(key) + if value is not None and str(value).strip(): + args += [flag, str(value)] + return args + def _safe_cpu(self, cpu: str) -> str: if str(cpu).strip() in {c.strip() for c in self.cfg.cpu_allow.split(",")}: return str(cpu).strip() diff --git a/bioflow_service/app/services/bioflow/runner.py b/bioflow_service/app/services/bioflow/runner.py index 2732f54..bd92df3 100644 --- a/bioflow_service/app/services/bioflow/runner.py +++ b/bioflow_service/app/services/bioflow/runner.py @@ -92,6 +92,43 @@ def _stage_input_satisfied(stage: str, job_input_dir: Path) -> tuple[bool, list[ return False, list(DEFAULT_INPUT_HINTS) +def ensure_samples_yaml(job_input_dir: Path) -> bool: + """为 rnaseq 系阶段自动生成 samples.yaml(从 fastq 按样本前缀配对)。 + + 扫描 input/ 下 *.fq.gz/fastq.gz,按 <样本>_f1/_r1 与 _f2/_r2 配对,生成 + {samples: {<样本>: {r1: <名>, r2: <名>}}}。已存在则不覆盖(用户手工配置优先)。 + 返回是否生成了新文件。 + """ + import re as _re + import yaml as _yaml + + yaml_path = job_input_dir / "samples.yaml" + if yaml_path.exists(): + return False + pairs: dict[str, dict[str, str]] = {} + for p in sorted(job_input_dir.rglob("*")): + if not p.is_file() or p.name == "samples.yaml": + continue + m = _re.match(r"^(.+?)[._-](f1|r1|1)(\.(fastq|fq))?(\.gz)?$", p.name, _re.I) + m2 = _re.match(r"^(.+?)[._-](f2|r2|2)(\.(fastq|fq))?(\.gz)?$", p.name, _re.I) + m = m or m2 + if not m: + continue + sample, mate = m.group(1), (m.group(2) or "").lower() + rec = pairs.setdefault(sample, {}) + if mate in ("f1", "r1", "1"): + rec["r1"] = p.name + else: + rec["r2"] = p.name + samples = {s: v for s, v in sorted(pairs.items()) if v.get("r1") and v.get("r2")} + if not samples: + return False + yaml_path.write_text( + _yaml.safe_dump({"samples": samples}, allow_unicode=True), encoding="utf-8" + ) + return True + + # ---- runner ---- ExecutorFn = Callable[..., Awaitable[dict[str, Any]]] diff --git a/bioflow_service/tests/test_kernel.py b/bioflow_service/tests/test_kernel.py index b2d6ab4..792f1f7 100644 --- a/bioflow_service/tests/test_kernel.py +++ b/bioflow_service/tests/test_kernel.py @@ -89,6 +89,44 @@ def test_build_command_kernel_dir_mount(tmp_path): assert "-w" in cmd and "/bioflow-kernel" in cmd +def test_build_command_cli_contract(tmp_path): + """契约阶段：按 cli_contract 注入 -i/-o/-c;无契约阶段保持 --input-dir。""" + from app.services.bioflow.kernel import Kernel, KernelConfig, ImageMap + cfg = KernelConfig(host_uid=1234, data_root=str(tmp_path / "data")) + img = ImageMap( + images={"rnaseq": {"docker_tag": "bioflow-rnaseq"}}, + stages={ + "rnaseq_qc": {"image": "rnaseq", "module": "cli.run_rnaseq_qc", + "cli_contract": [["--input", "input_dir"], + ["--config", "samples_yaml"], + ["--output", "output_final"]]}, + "rnaseq_align": {"image": "rnaseq", "module": "cli.run_rnaseq_align", + "cli_contract": [["--input", "input_dir"], + ["--hisat2-index", "param:hisat2_index"], + ["--output", "output_final"]]}, + "plain": {"image": "rnaseq", "module": "cli.run_plain"}, + }, + ) + k = Kernel(img, cfg) + # 有 samples.yaml 时注入 -c + (tmp_path / "in").mkdir(parents=True, exist_ok=True) + (tmp_path / "in" / "samples.yaml").write_text("samples: {}", encoding="utf-8") + cmd = k.build_command("rnaseq_qc", tmp_path / "in", tmp_path / "ws") + assert "--input" in cmd and "/job/input" in cmd + assert "--config" in cmd and "/job/input/samples.yaml" in cmd + assert "--output" in cmd and "/job/workspace/final" in cmd # rnaseq -o 语义对齐 + assert "--input-dir" not in cmd # 契约阶段不再注入默认形态 + # param: 有值注入、无值跳过 + cmd2 = k.build_command("rnaseq_align", tmp_path / "in", tmp_path / "ws", + params={"hisat2_index": "/bioflow-data/hmm/hisat2/osa"}) + assert "--hisat2-index" in cmd2 and "/bioflow-data/hmm/hisat2/osa" in cmd2 + cmd3 = k.build_command("rnaseq_align", tmp_path / "in", tmp_path / "ws") + assert "--hisat2-index" not in cmd3 + # 无契约阶段回归默认 + cmd4 = k.build_command("plain", tmp_path / "in", tmp_path / "ws") + assert "--input-dir" in cmd4 and "--workspace" in cmd4 + + def test_build_command_params(kernel, tmp_path): cmd = kernel.build_command("identify", tmp_path / "in", tmp_path / "ws", {"prefix": "Ft.SPL", "required_pfams": "PF03110",