From af34db97d4a1f48f79be3983169da8d7ae0e97b3 Mon Sep 17 00:00:00 2001 From: Qing Date: Tue, 14 Jul 2026 22:22:44 -0400 Subject: [PATCH 1/2] fix: stabilize annotation CLI integration --- .github/workflows/installation-profiles.yml | 29 +- Docs/Annotation_CLI.md | 15 ++ Docs/Annotation_Result_Schema.md | 5 + metax/cli/annotate.py | 51 +++- .../peptide_annotator/annotation_workflow.py | 175 +++++++++--- metax/peptide_annotator/output_paths.py | 26 ++ metax/peptide_annotator/pep_table_to_otf.py | 40 +-- metax/peptide_annotator/peptable_annotator.py | 12 +- metax/peptide_annotator/subprocess_utils.py | 86 ++++++ metax/peptide_annotator/unit_specific_otf.py | 10 +- tests/test_annotation_cli.py | 249 +++++++++++++++++- tests/test_annotation_subprocess_cleanup.py | 53 ++++ 12 files changed, 663 insertions(+), 88 deletions(-) create mode 100644 metax/peptide_annotator/output_paths.py create mode 100644 metax/peptide_annotator/subprocess_utils.py create mode 100644 tests/test_annotation_subprocess_cleanup.py diff --git a/.github/workflows/installation-profiles.yml b/.github/workflows/installation-profiles.yml index b21ad22e..359fab02 100644 --- a/.github/workflows/installation-profiles.yml +++ b/.github/workflows/installation-profiles.yml @@ -18,7 +18,7 @@ jobs: run: | python -c "import sys, metax, metax.cli.annotate, metax.cli.report; assert not any(name.startswith('PyQt') for name in sys.modules); assert 'metax.report' not in sys.modules" python -m metax.cli.annotate --help - python -m pytest tests/test_annotation_cli.py -q + python -m pytest tests/test_annotation_cli.py tests/test_annotation_subprocess_cleanup.py tests/test_pep_table_to_otf_dataframe.py tests/test_unit_specific_manifest.py -q headless-analysis-profiles: runs-on: ubuntu-latest @@ -34,13 +34,15 @@ jobs: - run: python -m pip install ".[${{ matrix.extra }}]" pytest - name: Verify Analyzer profile if: matrix.extra == 'analyzer' - run: python -c "from metax.taxafunc_analyzer.analyzer import TaxaFuncAnalyzer; import sys; assert not any(name.startswith('PyQt') for name in sys.modules)" + run: | + python -c "from metax.taxafunc_analyzer.analyzer import TaxaFuncAnalyzer; import sys; assert not any(name.startswith('PyQt') for name in sys.modules)" + python -m pytest tests/test_analyzer.py tests/test_annotation_cli.py -q - name: Verify Report profile if: matrix.extra == 'report' run: | python -c "import metax.report; import sys; assert not any(name.startswith('PyQt') for name in sys.modules)" python -m metax.cli.report --help - - run: python -m pytest tests/test_annotation_cli.py -q + python -m pytest tests/test_auto_report_config.py tests/test_auto_report_workflow.py tests/test_annotation_cli.py -q gui-and-full-metadata: runs-on: ubuntu-latest @@ -58,4 +60,23 @@ jobs: env: QT_QPA_PLATFORM: offscreen run: python -c "import PyQt5, qtawesome, qt_material; import metax.gui.main_gui" - - run: python -m pytest tests/test_annotation_cli.py -q + - name: Run shared backend and GUI annotation tests + env: + QT_QPA_PLATFORM: offscreen + run: python -m pytest tests/test_annotation_cli.py tests/test_unit_specific_gui_settings_dialog.py -q + + windows-annotation: + runs-on: windows-latest + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: "3.11" + - run: python -m pip install --upgrade pip + - run: python -m pip install . pytest + - name: Verify lightweight Windows annotation contract + shell: pwsh + run: | + python -c "import sys, metax, metax.cli.annotate; assert not any(name.startswith('PyQt') for name in sys.modules)" + python -m metax.cli.annotate --help + python -m pytest tests/test_annotation_cli.py tests/test_annotation_subprocess_cleanup.py tests/test_pep_table_to_otf_dataframe.py tests/test_unit_specific_manifest.py -q diff --git a/Docs/Annotation_CLI.md b/Docs/Annotation_CLI.md index 8d43bb25..ce8825b1 100644 --- a/Docs/Annotation_CLI.md +++ b/Docs/Annotation_CLI.md @@ -33,6 +33,11 @@ Use `--selection-mode provided` with `--selected-genomes` or `--genome-list-file` to supply genomes directly. Use `automatic` for MetaX's internal genome ranking, or `metaumbra-only` to stop after genome scoring. +When `--selection-mode` is omitted, selected genome IDs or a genome-list file +imply `provided`; otherwise digest directories imply `metaumbra`; otherwise the +mode is `automatic`. Explicit modes take precedence, and contradictory genome +sources are rejected instead of ignored. + Exactly one peptide-mapping source is required for OTF generation: `--peptide-db` or `--digested-genome-folders`. @@ -81,6 +86,10 @@ Run it with: metax-annotate --config annotation.yaml ``` +Relative paths loaded from a configuration file are resolved from the +configuration file's directory. Relative paths supplied directly on the command +line remain relative to the current working directory. + ## Result JSON When `--result-json` is supplied, MetaX writes schema `2.0` atomically. The @@ -92,6 +101,9 @@ run, inputs, parameters, stages, genome_selection, metrics, outputs, diagnostics Use `run.status` and `run.exit_code` for the outcome, `outputs` for generated files, `metrics` for mapping/QC values, and `diagnostics` for warnings or errors. +Scientific outputs are safely renamed when their requested path already exists, +and `outputs` reports the actual paths. The result JSON itself keeps its requested +stable path and is atomically replaced. See [Annotation Result Schema](Annotation_Result_Schema.md) for the complete field reference and examples. @@ -106,6 +118,9 @@ field reference and examples. | 5 | Annotation or external scoring failed | | 130 | Execution was cancelled | +On interruption, MetaX terminates active MetaUmbra or digested-scan process +trees before writing the cancellation result JSON. + ## Python subprocess Invoke MetaX with the same Python environment in which it is installed: diff --git a/Docs/Annotation_Result_Schema.md b/Docs/Annotation_Result_Schema.md index 23dd8623..824d62a8 100644 --- a/Docs/Annotation_Result_Schema.md +++ b/Docs/Annotation_Result_Schema.md @@ -77,6 +77,7 @@ annotation workflow API remains `1.0`. }, "genome_selection": { "method": "metaumbra", + "source": "metax_temp/OTF_metaumbra_genome_presence.tsv", "qvalue_cutoff": 0.05, "genomes_evaluated": 4744, "genomes_selected": 222, @@ -132,6 +133,10 @@ annotation workflow API remains `1.0`. Fields that do not apply to the selected mode are omitted. Output entries are only written for artifacts that were actually produced. +`parameters.selection_mode` and `genome_selection.method` contain the same +effective selection mode. `genome_selection.source` identifies explicit genome +IDs, a genome-list file, MetaUmbra output, or automatic genome ranking. + ## Unit-specific additions Unit-specific results use `run.mode: "unit-specific"` and add: diff --git a/metax/cli/annotate.py b/metax/cli/annotate.py index b40e9b20..b27e5d45 100644 --- a/metax/cli/annotate.py +++ b/metax/cli/annotate.py @@ -2,6 +2,7 @@ import argparse import json +import os import platform import sys import time @@ -38,6 +39,16 @@ def error(self, message: str) -> None: ANNOTATION_RESULT_SCHEMA_VERSION = "2.0" +_CONFIG_PATH_FIELDS = { + "peptide_table", + "unit_specific_manifest", + "taxafunc_db", + "output", + "peptide_db", + "genome_list_file", + "result_json", +} +_CONFIG_MULTI_PATH_FIELDS = {"digested_genome_folders"} def _decode_separator(value: str) -> str: @@ -82,8 +93,38 @@ def _flatten_config(raw_config: Mapping[str, Any]) -> dict[str, Any]: return flattened +def _resolve_config_path_value(value: Any, config_dir: Path) -> Any: + if value is None or not isinstance(value, (str, Path)): + return value + raw_value = os.path.expanduser(str(value)) + if not raw_value.strip(): + return raw_value + candidate = Path(raw_value) + return raw_value if candidate.is_absolute() else str(config_dir / candidate) + + +def _resolve_config_paths(config: Mapping[str, Any], config_dir: Path) -> dict[str, Any]: + resolved = dict(config) + for field_name in _CONFIG_PATH_FIELDS: + if field_name in resolved: + resolved[field_name] = _resolve_config_path_value( + resolved[field_name], config_dir + ) + for field_name in _CONFIG_MULTI_PATH_FIELDS: + if field_name not in resolved: + continue + value = resolved[field_name] + if isinstance(value, (str, Path)): + resolved[field_name] = _resolve_config_path_value(value, config_dir) + elif value is not None: + resolved[field_name] = [ + _resolve_config_path_value(item, config_dir) for item in value + ] + return resolved + + def load_config_file(path: str | Path) -> dict[str, Any]: - config_path = Path(path) + config_path = Path(path).expanduser() if not config_path.is_file(): raise FileNotFoundError(f"Configuration file not found: {config_path}") suffix = config_path.suffix.lower() @@ -112,7 +153,11 @@ def load_config_file(path: str | Path) -> dict[str, Any]: loaded = {} if not isinstance(loaded, Mapping): raise AnnotationConfigurationError("Annotation configuration must be a mapping") - config = _flatten_config(loaded) + config_path = config_path.resolve() + config = _resolve_config_paths( + _flatten_config(loaded), + config_path.parent, + ) requested_api = str( config.pop("workflow_api_version", config.pop("api_version", ANNOTATION_WORKFLOW_API_VERSION)) ) @@ -626,6 +671,8 @@ def main(argv: list[str] | None = None) -> int: stage_started = time.perf_counter() parser = build_parser(config) args = parser.parse_args(argv) + if pre_args.config: + args.config = config["config_path"] mode = args.mode _validate_scientific_parameters(args) result_json_path = args.result_json diff --git a/metax/peptide_annotator/annotation_workflow.py b/metax/peptide_annotator/annotation_workflow.py index bf83af4a..bcb73bde 100644 --- a/metax/peptide_annotator/annotation_workflow.py +++ b/metax/peptide_annotator/annotation_workflow.py @@ -13,12 +13,14 @@ import pandas as pd from metax.peptide_annotator.pep_table_to_otf import peptideProteinsMapper +from metax.peptide_annotator.output_paths import available_output_path from metax.peptide_annotator.peptide_table_prepare import ( is_diann_parquet, is_parquet_path, prepare_diann_parquet_for_direct_otf, read_parquet_columns, ) +from metax.peptide_annotator.subprocess_utils import run_streaming_subprocess ANNOTATION_WORKFLOW_API_VERSION = "1.0" @@ -120,6 +122,53 @@ def _normalise_genomes(values: str | Iterable[str] | None) -> list[str]: return genomes +def resolve_global_selection_mode( + selection_mode: str | None, + *, + selected_genomes: Iterable[str], + genome_list_path: str | Path | None, + digested_genome_folders: Iterable[str], +) -> str: + """Resolve and validate the effective global genome-selection mode.""" + selected_genomes = list(selected_genomes) + has_selected_genomes = bool(selected_genomes) + has_genome_list = genome_list_path is not None + has_digest_folders = bool(list(digested_genome_folders)) + + if has_selected_genomes and has_genome_list: + raise AnnotationConfigurationError( + "Use either selected_genomes or genome_list_file, not both" + ) + + effective_mode = selection_mode + if effective_mode is None: + if has_selected_genomes or has_genome_list: + effective_mode = "provided" + elif has_digest_folders: + effective_mode = "metaumbra" + else: + effective_mode = "automatic" + + if effective_mode not in GLOBAL_SELECTION_MODES: + raise AnnotationConfigurationError( + f"selection_mode must be one of {sorted(GLOBAL_SELECTION_MODES)}" + ) + if effective_mode == "provided" and not ( + has_selected_genomes or has_genome_list + ): + raise AnnotationConfigurationError( + "selection_mode='provided' requires selected_genomes or genome_list_file" + ) + if effective_mode != "provided" and ( + has_selected_genomes or has_genome_list + ): + raise AnnotationConfigurationError( + f"selection_mode={effective_mode!r} cannot be combined with " + "selected_genomes or genome_list_file" + ) + return effective_mode + + def _file_descriptor(path: str | Path, *, format_name: str | None = None) -> dict[str, object]: file_path = Path(path) descriptor: dict[str, object] = {"path": str(file_path)} @@ -205,18 +254,28 @@ def ensure_metaumbra_available( return ".".join(map(str, actual)) -def run_metaumbra_scoring( +def _build_metaumbra_scoring_command( *, peptide_table_path: str, digested_genome_folders: str | Iterable[str], output_path: str, peptide_col: str, - peptide_score_col: str = "Evidence", - peptide_error_col: str = "Q.Value", - single_peptide_error_rate_upper_bound: float = 0.3, -) -> dict[str, object]: + peptide_score_col: str, + peptide_error_col: str, + single_peptide_error_rate_upper_bound: float, +) -> list[str]: digest_dirs = _normalise_paths(digested_genome_folders) - command = [ + if not digest_dirs: + raise AnnotationConfigurationError( + "MetaUmbra scoring requires at least one digested genome directory" + ) + unsupported = [path for path in digest_dirs if re.search(r"[,;，；]", path)] + if unsupported: + raise AnnotationConfigurationError( + "MetaUmbra directory paths cannot contain comma or semicolon delimiters: " + + ", ".join(unsupported) + ) + return [ sys.executable, "-m", "metaumbra", @@ -224,7 +283,7 @@ def run_metaumbra_scoring( "--peptide-table", peptide_table_path, "--genome-digest-dirs", - *digest_dirs, + ";".join(digest_dirs), "--output", output_path, "--peptide-seq-col", @@ -237,33 +296,41 @@ def run_metaumbra_scoring( str(single_peptide_error_rate_upper_bound), ] + +def run_metaumbra_scoring( + *, + peptide_table_path: str, + digested_genome_folders: str | Iterable[str], + output_path: str, + peptide_col: str, + peptide_score_col: str = "Evidence", + peptide_error_col: str = "Q.Value", + single_peptide_error_rate_upper_bound: float = 0.3, +) -> dict[str, object]: + command = _build_metaumbra_scoring_command( + peptide_table_path=peptide_table_path, + digested_genome_folders=digested_genome_folders, + output_path=output_path, + peptide_col=peptide_col, + peptide_score_col=peptide_score_col, + peptide_error_col=peptide_error_col, + single_peptide_error_rate_upper_bound=single_peptide_error_rate_upper_bound, + ) + print("Launching MetaUmbra scoring in an isolated process:") print(" ".join(f'"{part}"' if " " in str(part) else str(part) for part in command)) project_root, env = _metaumbra_environment() metaumbra_version = ensure_metaumbra_available(env=env) creationflags = getattr(subprocess, "CREATE_NO_WINDOW", 0) - process = subprocess.Popen( + return_code, last_lines = run_streaming_subprocess( command, cwd=project_root, - stdout=subprocess.PIPE, - stderr=subprocess.STDOUT, - text=True, - encoding="utf-8", - errors="replace", - bufsize=1, env=env, creationflags=creationflags, + max_captured_lines=80, + emit_line=print, ) - last_lines: list[str] = [] - assert process.stdout is not None - for line in process.stdout: - print(line.rstrip("\n")) - last_lines.append(line) - if len(last_lines) > 80: - last_lines = last_lines[-80:] - - return_code = process.wait() if return_code != 0: tail = "".join(last_lines[-30:]) raise RuntimeError( @@ -318,17 +385,21 @@ def __init__( selected_genome_source: str | None = None, ) -> None: self.peptide_table_path = Path(peptide_table_path) - self.output_path = Path(output_path) + self.requested_output_path = Path(output_path) + self.output_path = available_output_path(self.requested_output_path) self.taxafunc_anno_db_path = ( Path(taxafunc_anno_db_path) if taxafunc_anno_db_path else None ) self.db_path = Path(db_path) if db_path else None self.digested_genome_folders = _normalise_paths(digested_genome_folders) - self.selection_mode = selection_mode or ( - "metaumbra" if self.digested_genome_folders else "automatic" - ) self.selected_genomes = _normalise_genomes(selected_genomes) self.genome_list_path = Path(genome_list_path) if genome_list_path else None + self.selection_mode = resolve_global_selection_mode( + selection_mode, + selected_genomes=self.selected_genomes, + genome_list_path=self.genome_list_path, + digested_genome_folders=self.digested_genome_folders, + ) self.peptide_col = peptide_col self.intensity_col_prefix = intensity_col_prefix self.table_separator = table_separator @@ -367,10 +438,6 @@ def info_path(self) -> Path: return self.output_path.with_name(f"{self.output_path.stem}_info.txt") def _validate(self) -> None: - if self.selection_mode not in GLOBAL_SELECTION_MODES: - raise AnnotationConfigurationError( - f"selection_mode must be one of {sorted(GLOBAL_SELECTION_MODES)}" - ) if not self.peptide_table_path.is_file(): raise FileNotFoundError(f"Peptide table not found: {self.peptide_table_path}") if not self.output_path.parent.is_dir(): @@ -390,17 +457,37 @@ def _validate(self) -> None: raise AnnotationConfigurationError( "Exactly one of db_path or digested_genome_folders is required" ) - if self.db_path is not None and not self.db_path.is_file(): + if ( + self.db_path is not None + and self.selection_mode != "metaumbra-only" + and not self.db_path.is_file() + ): raise FileNotFoundError(f"Peptide database not found: {self.db_path}") for digest_dir in self.digested_genome_folders: if not Path(digest_dir).is_dir(): raise FileNotFoundError(f"Digested genome directory not found: {digest_dir}") - if self.selection_mode in {"metaumbra", "metaumbra-only"} and not self.digested_genome_folders: + if ( + self.selection_mode in {"metaumbra", "metaumbra-only"} + and not self.digested_genome_folders + ): raise AnnotationConfigurationError( f"selection_mode={self.selection_mode!r} requires digested_genome_folders" ) if self.genome_list_path is not None and not self.genome_list_path.is_file(): raise FileNotFoundError(f"Genome list not found: {self.genome_list_path}") + if self.selection_mode == "metaumbra-only": + ignored_inputs = [] + if self.taxafunc_anno_db_path is not None: + ignored_inputs.append("taxafunc_anno_db_path") + if self.db_path is not None: + ignored_inputs.append("db_path") + if ignored_inputs: + warning_message = ( + "selection_mode='metaumbra-only' ignores OTF annotation inputs: " + + ", ".join(ignored_inputs) + ) + if warning_message not in self.warnings: + self.warnings.append(warning_message) if not 0 <= self.lca_threshold <= 1: raise AnnotationConfigurationError("lca_threshold must be between 0 and 1") if not 0 < self.protein_peptide_coverage_cutoff <= 1: @@ -482,6 +569,11 @@ def run(self) -> GlobalOTFRunResult: stages: dict[str, object] = {} stage_started = time.perf_counter() self._validate() + if self.output_path != self.requested_output_path: + print( + "Output file already exists; this run will write to: " + f"{self.output_path}" + ) stages["validation"] = { "status": "success", "duration_seconds": _elapsed(stage_started), @@ -588,6 +680,7 @@ def run(self) -> GlobalOTFRunResult: stage_started = time.perf_counter() selected_genomes = self._resolve_selected_genomes() + provided_genomes_input_count = len(selected_genomes) stages["genome_selection_input"] = { "status": "success", "duration_seconds": _elapsed(stage_started), @@ -599,7 +692,13 @@ def run(self) -> GlobalOTFRunResult: genome_selection: dict[str, object] = {"method": self.selection_mode} selection_source = self.selected_genome_source or ( - str(self.genome_list_path) if self.genome_list_path else None + str(self.genome_list_path) + if self.genome_list_path + else "selected_genomes" + if self.selected_genomes + else "automatic_genome_ranking" + if self.selection_mode == "automatic" + else None ) if selection_source: genome_selection["source"] = selection_source @@ -631,6 +730,8 @@ def run(self) -> GlobalOTFRunResult: outputs["genome_presence"] = _file_descriptor( scoring_output, format_name="tsv" ) + selection_source = str(scoring_output) + genome_selection["source"] = selection_source software["metaumbra_version"] = str(scoring_result["metaumbra_version"]) metaumbra_metadata = { "metaumbra_genome_presence_path": str(scoring_output), @@ -690,8 +791,10 @@ def run(self) -> GlobalOTFRunResult: selection_metadata = { "workflow": "Peptide Direct to OTFs (MetaUmbra)", "genome_selection_method": self.selection_mode, - "metaumbra_scoring_run": self.selection_mode == "metaumbra", - "selected_genomes_input_count": len(selected_genomes), + "metaumbra_scoring_run": ( + self.selection_mode in {"metaumbra", "metaumbra-only"} + ), + "selected_genomes_input_count": provided_genomes_input_count, **metaumbra_metadata, **input_metadata, } diff --git a/metax/peptide_annotator/output_paths.py b/metax/peptide_annotator/output_paths.py new file mode 100644 index 00000000..41e1973f --- /dev/null +++ b/metax/peptide_annotator/output_paths.py @@ -0,0 +1,26 @@ +from __future__ import annotations + +from datetime import datetime +from pathlib import Path + + +def available_output_path(requested_path: str | Path) -> Path: + """Return the requested path or a non-existing timestamped sibling. + + Scientific outputs are preserved when the requested target already exists. + Stable automation files such as result JSON are handled separately and may + intentionally use atomic replacement at their exact requested path. + """ + requested = Path(requested_path) + if not requested.exists(): + return requested + + timestamp = datetime.now().strftime("%Y%m%d%H%M%S") + candidate = requested.with_name(f"{requested.stem}_{timestamp}{requested.suffix}") + counter = 2 + while candidate.exists(): + candidate = requested.with_name( + f"{requested.stem}_{timestamp}_{counter}{requested.suffix}" + ) + counter += 1 + return candidate diff --git a/metax/peptide_annotator/pep_table_to_otf.py b/metax/peptide_annotator/pep_table_to_otf.py index e69bdf30..6ffe45c7 100644 --- a/metax/peptide_annotator/pep_table_to_otf.py +++ b/metax/peptide_annotator/pep_table_to_otf.py @@ -86,6 +86,8 @@ def _ensure_project_root_on_syspath() -> None: _ensure_project_root_on_syspath() +from metax.peptide_annotator.subprocess_utils import run_streaming_subprocess + # NOTE: Avoid importing GUI/Matplotlib-related modules at import-time. # This file can be imported inside multiprocessing workers on Windows; importing # Qt/Matplotlib backends there can trigger repeated backend initialization or @@ -561,26 +563,14 @@ def _query_peptide_proteins_nested_via_subprocess( env.setdefault("PYTHONIOENCODING", "utf-8") creationflags = getattr(subprocess, "CREATE_NO_WINDOW", 0) print("[UnitSpecificDigestedScan/Subprocess] Launching isolated process.", flush=True) - proc = subprocess.Popen( + return_code, last_lines = run_streaming_subprocess( cmd, cwd=str(repo_root), env=env, - stdout=subprocess.PIPE, - stderr=subprocess.STDOUT, - text=True, - encoding="utf-8", - errors="replace", - bufsize=1, creationflags=creationflags, + max_captured_lines=50, + emit_line=lambda line: print(line, flush=True), ) - last_lines: list[str] = [] - assert proc.stdout is not None - for line in proc.stdout: - print(line.rstrip("\n"), flush=True) - last_lines.append(line) - if len(last_lines) > 50: - last_lines = last_lines[-50:] - return_code = proc.wait() if return_code != 0: tail = "".join(last_lines[-20:]) raise RuntimeError( @@ -1411,28 +1401,14 @@ def _query_peptide_proteins_via_subprocess( except Exception: creationflags = 0 - proc = subprocess.Popen( + rc, last_lines = run_streaming_subprocess( cmd, cwd=str(repo_root), env=env, - stdout=subprocess.PIPE, - stderr=subprocess.STDOUT, - text=True, - encoding="utf-8", - errors="replace", - bufsize=1, creationflags=creationflags, + max_captured_lines=50, + emit_line=print, ) - - last_lines: list[str] = [] - assert proc.stdout is not None - for line in proc.stdout: - print(line.rstrip("\n")) - last_lines.append(line) - if len(last_lines) > 50: - last_lines = last_lines[-50:] - - rc = proc.wait() if rc != 0: tail = "".join(last_lines[-20:]) raise RuntimeError(f"Digested scan subprocess failed (exit={rc}). Last output:\n{tail}") diff --git a/metax/peptide_annotator/peptable_annotator.py b/metax/peptide_annotator/peptable_annotator.py index bb3794a5..365c6ea6 100644 --- a/metax/peptide_annotator/peptable_annotator.py +++ b/metax/peptide_annotator/peptable_annotator.py @@ -12,11 +12,13 @@ try: from ..utils.version import __version__ + from .output_paths import available_output_path from .proteins_to_taxafunc import Pep2TaxaFunc from .convert_id_to_name import add_pathway_name_to_df, add_ec_name_to_df, add_ko_name_to_df, add_kegg_module_to_df, add_go_name_to_df except ImportError: print("ImportError occurred, trying alternative imports...") __version__ = "Test version" + from output_paths import available_output_path from proteins_to_taxafunc import Pep2TaxaFunc from convert_id_to_name import add_pathway_name_to_df, add_ec_name_to_df, add_ko_name_to_df, add_kegg_module_to_df, add_go_name_to_df @@ -374,15 +376,7 @@ def save_result(self, df): print(f'Output directory did not exist, created: {dir_path}') if os.path.exists(self.output_path): - counter = 1 - base_name = os.path.splitext(os.path.basename(self.output_path))[0] - ext = os.path.splitext(self.output_path)[-1] - new_output_path = os.path.join(dir_path, f'{base_name}_{pd.Timestamp.now().strftime("%Y%m%d%H%M%S")}{ext}') - - while os.path.exists(new_output_path): - counter += 1 - new_output_path = os.path.join(dir_path, f'{base_name}_{counter}{ext}') - self.output_path = new_output_path + self.output_path = str(available_output_path(self.output_path)) print(f'Output file already exists, saved as: {self.output_path}') # get metadata for the running diff --git a/metax/peptide_annotator/subprocess_utils.py b/metax/peptide_annotator/subprocess_utils.py new file mode 100644 index 00000000..166ffedc --- /dev/null +++ b/metax/peptide_annotator/subprocess_utils.py @@ -0,0 +1,86 @@ +from __future__ import annotations + +import os +import signal +import subprocess +from collections.abc import Callable, Mapping, Sequence +from os import PathLike +from typing import Any + + +def terminate_process( + process: subprocess.Popen[Any], + *, + timeout: float = 5.0, + process_group: bool = False, +) -> None: + """Best-effort termination for an active child and its process group/tree.""" + if process.poll() is not None: + return + try: + if os.name == "nt": + subprocess.run( + ["taskkill", "/PID", str(process.pid), "/T", "/F"], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + timeout=timeout, + check=False, + creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0), + ) + elif process_group: + os.killpg(process.pid, signal.SIGTERM) + else: + process.terminate() + process.wait(timeout=timeout) + except subprocess.TimeoutExpired: + if os.name != "nt" and process_group: + os.killpg(process.pid, signal.SIGKILL) + else: + process.kill() + process.wait() + except OSError: + if process.poll() is None: + process.kill() + process.wait() + return + + +def run_streaming_subprocess( + command: Sequence[str], + *, + cwd: str | PathLike[str] | None = None, + env: Mapping[str, str] | None = None, + creationflags: int = 0, + max_captured_lines: int = 50, + emit_line: Callable[[str], None] | None = None, +) -> tuple[int, list[str]]: + """Run a command without a shell, stream output, and clean up on interruption.""" + process = subprocess.Popen( + list(command), + cwd=cwd, + env=dict(env) if env is not None else None, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + text=True, + encoding="utf-8", + errors="replace", + bufsize=1, + creationflags=creationflags, + start_new_session=os.name != "nt", + ) + captured: list[str] = [] + try: + assert process.stdout is not None + for line in process.stdout: + if emit_line is not None: + emit_line(line.rstrip("\n")) + captured.append(line) + if len(captured) > max_captured_lines: + captured = captured[-max_captured_lines:] + return process.wait(), captured + except BaseException: + terminate_process(process, process_group=os.name != "nt") + raise + finally: + if process.stdout is not None: + process.stdout.close() diff --git a/metax/peptide_annotator/unit_specific_otf.py b/metax/peptide_annotator/unit_specific_otf.py index 0e3bed0b..18ac8500 100644 --- a/metax/peptide_annotator/unit_specific_otf.py +++ b/metax/peptide_annotator/unit_specific_otf.py @@ -17,6 +17,7 @@ peptideProteinsMapper, query_peptide_proteins_from_digested_genome_folders_nested, ) +from metax.peptide_annotator.output_paths import available_output_path from metax.peptide_annotator.peptide_table_prepare import ( has_diann_core_columns, is_parquet_path, @@ -160,7 +161,8 @@ def __init__( self.peptide_table_path = Path(peptide_table_path) self.unit_specific_manifest_path = Path(unit_specific_manifest_path) self.taxafunc_anno_db_path = Path(taxafunc_anno_db_path) - self.output_path = Path(output_path) + self.requested_output_path = Path(output_path) + self.output_path = available_output_path(self.requested_output_path) self.db_path = db_path self.digested_genome_folders = digested_genome_folders self.genome_threshold = genome_threshold @@ -712,6 +714,12 @@ def run( return_dataframe: bool = False, ) -> UnitSpecificOTFRunResult | pd.DataFrame: started_at = datetime.now() + if self.output_path != self.requested_output_path: + print( + "Output file already exists; this run will write to: " + f"{self.output_path}", + flush=True, + ) manifest = load_unit_specific_manifest( self.unit_specific_manifest_path, genome_threshold=self.genome_threshold, diff --git a/tests/test_annotation_cli.py b/tests/test_annotation_cli.py index 9d4746fd..fa4606f6 100644 --- a/tests/test_annotation_cli.py +++ b/tests/test_annotation_cli.py @@ -1,5 +1,6 @@ import json import ast +import shutil import sqlite3 import subprocess import sys @@ -144,6 +145,7 @@ def test_global_annotation_cli_reports_renamed_output_when_target_exists( requested_output = tmp_path / "global_otf.tsv" requested_output.write_text("stale output\n", encoding="utf-8") result_json = tmp_path / "global_result.json" + result_json.write_text("stale result\n", encoding="utf-8") exit_code = main( [ @@ -293,7 +295,7 @@ def all_in_one(self, **kwargs): monkeypatch.setattr(annotation_workflow, "run_metaumbra_scoring", fake_scoring) monkeypatch.setattr(annotation_workflow, "peptideProteinsMapper", FakeMapper) - GlobalOTFAnnotator( + run_result = GlobalOTFAnnotator( peptide_table_path=str(peptide_table), output_path=str(output), taxafunc_anno_db_path=str(taxafunc_db), @@ -314,6 +316,11 @@ def all_in_one(self, **kwargs): assert captured["metaumbra_single_peptide_error_rate_upper_bound"] == 0.01 assert captured["metaumbra_genome_qvalue_cutoff"] == 0.05 assert captured["selected_genomes_from_metaumbra_count"] == 1 + assert captured["selected_genomes_input_count"] == 0 + assert run_result.genome_selection["method"] == "metaumbra" + assert run_result.genome_selection["source"].endswith( + "_metaumbra_genome_presence.tsv" + ) def test_unit_specific_annotation_cli_writes_summary_contract( @@ -347,6 +354,7 @@ def test_unit_specific_annotation_cli_writes_summary_contract( encoding="utf-8", ) output = tmp_path / "unit_otf.tsv" + output.write_text("stale unit output\n", encoding="utf-8") result_json = tmp_path / "unit_result.json" exit_code = main( @@ -369,10 +377,18 @@ def test_unit_specific_annotation_cli_writes_summary_contract( ) assert exit_code == ExitCode.SUCCESS + assert output.read_text(encoding="utf-8") == "stale unit output\n" result = json.loads(result_json.read_text(encoding="utf-8")) assert result["run"]["mode"] == "unit-specific" - assert result["outputs"]["otf"]["path"] == str(output) + actual_output = Path(result["outputs"]["otf"]["path"]) + assert actual_output != output + assert actual_output.is_file() assert Path(result["outputs"]["annotation_summary"]["path"]).is_file() + unit_summary = Path(result["outputs"]["unit_summary"]["path"]) + assert unit_summary.is_file() + assert unit_summary.parent == actual_output.with_name( + f"{actual_output.stem}_artifacts" + ) assert "per_unit_directory" not in result["outputs"] assert result["metrics"]["units"] == {"completed": 2, "skipped": 0} assert result["genome_selection"]["method"] == "unit_specific_manifest" @@ -453,13 +469,238 @@ def test_yaml_config_is_parsed_and_cli_values_can_override(tmp_path): config = load_config_file(config_path) assert config["mode"] == "global" - assert config["output"] == "configured.tsv" - assert config["result_json"] == "result.json" + assert config["output"] == str(tmp_path / "configured.tsv") + assert config["result_json"] == str(tmp_path / "result.json") + assert config["peptide_table"] == str(tmp_path / "input.tsv") + assert config["peptide_db"] == str(tmp_path / "peptides.db") + assert config["taxafunc_db"] == str(tmp_path / "taxafunc.db") assert config["selected_genomes"] == ["g1", "g2"] parsed = build_parser(config).parse_args(["--output", "overridden.tsv"]) assert parsed.output == "overridden.tsv" +@pytest.mark.parametrize( + ("selection_mode", "selected_genomes", "genome_list", "digest_dirs", "expected"), + [ + (None, ["g1"], None, [], "provided"), + (None, [], "genomes.txt", [], "provided"), + (None, [], None, ["digests"], "metaumbra"), + (None, [], None, [], "automatic"), + ("automatic", [], None, ["digests"], "automatic"), + ], +) +def test_global_selection_mode_resolution( + selection_mode, + selected_genomes, + genome_list, + digest_dirs, + expected, +): + assert annotation_workflow.resolve_global_selection_mode( + selection_mode, + selected_genomes=selected_genomes, + genome_list_path=genome_list, + digested_genome_folders=digest_dirs, + ) == expected + + +@pytest.mark.parametrize( + ("selection_mode", "selected_genomes", "genome_list", "digest_dirs"), + [ + ("provided", [], None, []), + ("automatic", ["g1"], None, []), + ("metaumbra", ["g1"], None, ["digests"]), + ("metaumbra-only", [], "genomes.txt", ["digests"]), + ("provided", ["g1"], "genomes.txt", []), + ], +) +def test_global_selection_mode_rejects_contradictory_inputs( + selection_mode, + selected_genomes, + genome_list, + digest_dirs, +): + with pytest.raises(AnnotationConfigurationError): + annotation_workflow.resolve_global_selection_mode( + selection_mode, + selected_genomes=selected_genomes, + genome_list_path=genome_list, + digested_genome_folders=digest_dirs, + ) + + +def test_metaumbra_mode_requires_digest_directories_when_run(tmp_path): + peptide_table = tmp_path / "peptides.tsv" + peptide_table.write_text("Sequence\tEvidence\tQ.Value\nPEPA\t1\t0.01\n", encoding="utf-8") + + annotator = GlobalOTFAnnotator( + peptide_table_path=str(peptide_table), + output_path=str(tmp_path / "genome_presence.tsv"), + selection_mode="metaumbra-only", + ) + + with pytest.raises( + AnnotationConfigurationError, + match="requires digested_genome_folders", + ): + annotator._validate() + + +@pytest.mark.parametrize("folder_count", [1, 2]) +def test_metaumbra_command_encodes_digest_directories_as_one_argument( + tmp_path, + folder_count, +): + digest_dirs = [ + tmp_path / f"digest folder {index}" for index in range(1, folder_count + 1) + ] + command = annotation_workflow._build_metaumbra_scoring_command( + peptide_table_path=str(tmp_path / "peptides with spaces.tsv"), + digested_genome_folders=[str(path) for path in digest_dirs], + output_path=str(tmp_path / "output with spaces.tsv"), + peptide_col="Sequence", + peptide_score_col="Evidence", + peptide_error_col="Q.Value", + single_peptide_error_rate_upper_bound=0.3, + ) + + option_index = command.index("--genome-digest-dirs") + assert command[option_index + 1] == ";".join(str(path) for path in digest_dirs) + assert command[option_index + 2] == "--output" + + +def test_config_paths_follow_config_directory_and_cli_overrides_follow_cwd( + annotation_inputs, + tmp_path, + monkeypatch, +): + peptide_table, peptide_db, taxafunc_db = annotation_inputs + config_dir = tmp_path / "portable config" + config_dir.mkdir() + results_dir = config_dir / "results" + results_dir.mkdir() + for source in (peptide_table, peptide_db, taxafunc_db): + shutil.copy2(source, config_dir / source.name) + + config_path = config_dir / "annotation.json" + config_path.write_text( + json.dumps( + { + "mode": "global", + "inputs": { + "peptide_table": peptide_table.name, + "peptide_db": peptide_db.name, + "taxafunc_db": taxafunc_db.name, + }, + "output": { + "otf": "results/configured.tsv", + "result_json": "results/configured.json", + }, + "options": {"selected_genomes": ["g1"]}, + } + ), + encoding="utf-8", + ) + launch_dir = tmp_path / "different working directory" + launch_dir.mkdir() + monkeypatch.chdir(launch_dir) + + assert main(["--config", str(config_path)]) == ExitCode.SUCCESS + configured_output = results_dir / "configured.tsv" + configured_result = results_dir / "configured.json" + assert configured_output.is_file() + result = json.loads(configured_result.read_text(encoding="utf-8")) + assert result["parameters"]["selection_mode"] == "provided" + assert result["genome_selection"]["source"] == "selected_genomes" + assert result["outputs"]["otf"]["path"] == str(configured_output) + assert result["inputs"]["config"]["path"] == str(config_path) + + assert main( + [ + "--config", + str(config_path), + "--output", + "cli-override.tsv", + "--result-json", + "cli-override.json", + ] + ) == ExitCode.SUCCESS + assert (launch_dir / "cli-override.tsv").is_file() + override_result = json.loads( + (launch_dir / "cli-override.json").read_text(encoding="utf-8") + ) + assert override_result["outputs"]["otf"]["path"] == "cli-override.tsv" + + +def test_config_path_resolution_preserves_absolute_paths_and_resolves_lists( + tmp_path, +): + config_dir = tmp_path / "config" + config_dir.mkdir() + absolute_digest = tmp_path / "absolute digest" + absolute_digest.mkdir() + absolute_table = tmp_path / "absolute.tsv" + config_path = config_dir / "annotation.json" + config_path.write_text( + json.dumps( + { + "inputs": { + "peptide_table": str(absolute_table), + "digested_genome_folders": [ + "relative digest", + str(absolute_digest), + ], + } + } + ), + encoding="utf-8", + ) + + config = load_config_file(config_path) + + assert config["peptide_table"] == str(absolute_table) + assert config["digested_genome_folders"] == [ + str(config_dir / "relative digest"), + str(absolute_digest), + ] + + +def test_metaumbra_only_preserves_existing_output_and_reports_actual_path( + monkeypatch, + tmp_path, +): + peptide_table = tmp_path / "peptides.tsv" + peptide_table.write_text( + "Sequence\tEvidence\tQ.Value\nPEPA\t1\t0.01\n", + encoding="utf-8", + ) + digest_dir = tmp_path / "digests" + digest_dir.mkdir() + requested_output = tmp_path / "genome_presence.tsv" + requested_output.write_text("stale genome presence\n", encoding="utf-8") + + def fake_scoring(**kwargs): + pd.DataFrame( + {"genome_id": ["g1"], "qvalue": [0.01]} + ).to_csv(kwargs["output_path"], sep="\t", index=False) + return {"output": kwargs["output_path"], "metaumbra_version": "1.3.7"} + + monkeypatch.setattr(annotation_workflow, "run_metaumbra_scoring", fake_scoring) + result = GlobalOTFAnnotator( + peptide_table_path=str(peptide_table), + output_path=str(requested_output), + digested_genome_folders=str(digest_dir), + selection_mode="metaumbra-only", + ).run() + + actual_output = Path(result.output_path) + assert requested_output.read_text(encoding="utf-8") == "stale genome presence\n" + assert actual_output != requested_output + assert actual_output.is_file() + assert result.outputs["genome_presence"]["path"] == str(actual_output) + assert result.genome_selection["source"] == str(actual_output) + + def test_config_rejects_incompatible_api_version(tmp_path): config_path = tmp_path / "annotation.json" config_path.write_text( diff --git a/tests/test_annotation_subprocess_cleanup.py b/tests/test_annotation_subprocess_cleanup.py new file mode 100644 index 00000000..7ad85207 --- /dev/null +++ b/tests/test_annotation_subprocess_cleanup.py @@ -0,0 +1,53 @@ +from __future__ import annotations + +import os + +import pytest + +import metax.peptide_annotator.subprocess_utils as subprocess_utils + + +class _InterruptingStdout: + def __init__(self) -> None: + self.closed = False + + def __iter__(self): + return self + + def __next__(self): + raise KeyboardInterrupt + + def close(self) -> None: + self.closed = True + + +class _FakeProcess: + pid = 12345 + + def __init__(self) -> None: + self.stdout = _InterruptingStdout() + + +def test_streaming_subprocess_cleans_up_on_keyboard_interrupt(monkeypatch): + process = _FakeProcess() + cleanup_calls = [] + monkeypatch.setattr( + subprocess_utils.subprocess, + "Popen", + lambda *_args, **_kwargs: process, + ) + monkeypatch.setattr( + subprocess_utils, + "terminate_process", + lambda active_process, **kwargs: cleanup_calls.append( + (active_process, kwargs) + ), + ) + + with pytest.raises(KeyboardInterrupt): + subprocess_utils.run_streaming_subprocess(["fake", "command"]) + + assert cleanup_calls == [ + (process, {"process_group": os.name != "nt"}) + ] + assert process.stdout.closed is True From 76e7b38f1733944bf9dacb4fc84e3183691bf9b7 Mon Sep 17 00:00:00 2001 From: Qing Date: Wed, 15 Jul 2026 11:57:21 -0400 Subject: [PATCH 2/2] fix: terminate orphaned process groups --- metax/peptide_annotator/subprocess_utils.py | 72 +++++++++++++++++++-- tests/test_annotation_subprocess_cleanup.py | 61 +++++++++++++++++ 2 files changed, 127 insertions(+), 6 deletions(-) diff --git a/metax/peptide_annotator/subprocess_utils.py b/metax/peptide_annotator/subprocess_utils.py index 166ffedc..35f33b5e 100644 --- a/metax/peptide_annotator/subprocess_utils.py +++ b/metax/peptide_annotator/subprocess_utils.py @@ -3,11 +3,67 @@ import os import signal import subprocess +import time from collections.abc import Callable, Mapping, Sequence from os import PathLike from typing import Any +def _process_group_exists(process_group_id: int) -> bool: + try: + os.killpg(process_group_id, 0) + except ProcessLookupError: + return False + except PermissionError: + return True + return True + + +def _wait_for_process_group_exit( + process_group_id: int, + timeout: float, +) -> bool: + deadline = time.monotonic() + timeout + while _process_group_exists(process_group_id): + remaining = deadline - time.monotonic() + if remaining <= 0: + return False + time.sleep(min(0.05, remaining)) + return True + + +def _terminate_posix_process_group( + process: subprocess.Popen[Any], + timeout: float, +) -> None: + process_group_id = process.pid + try: + os.killpg(process_group_id, signal.SIGTERM) + except ProcessLookupError: + return + + deadline = time.monotonic() + timeout + if process.poll() is None: + try: + process.wait(timeout=max(0.0, deadline - time.monotonic())) + except subprocess.TimeoutExpired: + pass + + if _wait_for_process_group_exit( + process_group_id, + max(0.0, deadline - time.monotonic()), + ): + return + + try: + os.killpg(process_group_id, signal.SIGKILL) + except ProcessLookupError: + return + if process.poll() is None: + process.wait() + _wait_for_process_group_exit(process_group_id, timeout) + + def terminate_process( process: subprocess.Popen[Any], *, @@ -15,6 +71,15 @@ def terminate_process( process_group: bool = False, ) -> None: """Best-effort termination for an active child and its process group/tree.""" + if process_group: + try: + _terminate_posix_process_group(process, timeout) + except OSError: + if process.poll() is None: + process.kill() + process.wait() + return + if process.poll() is not None: return try: @@ -27,16 +92,11 @@ def terminate_process( check=False, creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0), ) - elif process_group: - os.killpg(process.pid, signal.SIGTERM) else: process.terminate() process.wait(timeout=timeout) except subprocess.TimeoutExpired: - if os.name != "nt" and process_group: - os.killpg(process.pid, signal.SIGKILL) - else: - process.kill() + process.kill() process.wait() except OSError: if process.poll() is None: diff --git a/tests/test_annotation_subprocess_cleanup.py b/tests/test_annotation_subprocess_cleanup.py index 7ad85207..b9b28900 100644 --- a/tests/test_annotation_subprocess_cleanup.py +++ b/tests/test_annotation_subprocess_cleanup.py @@ -28,6 +28,16 @@ def __init__(self) -> None: self.stdout = _InterruptingStdout() +class _ExitedProcess: + pid = 12345 + + def poll(self) -> int: + return 0 + + def wait(self, timeout=None): + raise AssertionError("An exited process-group leader must not be waited on") + + def test_streaming_subprocess_cleans_up_on_keyboard_interrupt(monkeypatch): process = _FakeProcess() cleanup_calls = [] @@ -51,3 +61,54 @@ def test_streaming_subprocess_cleans_up_on_keyboard_interrupt(monkeypatch): (process, {"process_group": os.name != "nt"}) ] assert process.stdout.closed is True + + +def test_terminate_process_signals_group_when_leader_has_exited(monkeypatch): + process = _ExitedProcess() + signals = [] + monkeypatch.setattr( + subprocess_utils.os, + "killpg", + lambda process_group_id, sig: signals.append((process_group_id, sig)), + raising=False, + ) + monkeypatch.setattr( + subprocess_utils, + "_wait_for_process_group_exit", + lambda _process_group_id, _timeout: True, + ) + + subprocess_utils.terminate_process(process, process_group=True) + + assert signals == [(process.pid, subprocess_utils.signal.SIGTERM)] + + +def test_terminate_process_escalates_when_group_ignores_sigterm(monkeypatch): + process = _ExitedProcess() + signals = [] + group_exit_checks = iter((False, True)) + sigkill = getattr(subprocess_utils.signal, "SIGKILL", 9) + monkeypatch.setattr( + subprocess_utils.signal, + "SIGKILL", + sigkill, + raising=False, + ) + monkeypatch.setattr( + subprocess_utils.os, + "killpg", + lambda process_group_id, sig: signals.append((process_group_id, sig)), + raising=False, + ) + monkeypatch.setattr( + subprocess_utils, + "_wait_for_process_group_exit", + lambda _process_group_id, _timeout: next(group_exit_checks), + ) + + subprocess_utils.terminate_process(process, process_group=True) + + assert signals == [ + (process.pid, subprocess_utils.signal.SIGTERM), + (process.pid, sigkill), + ]