diff --git a/.gitignore b/.gitignore index 227f773a..15251294 100644 --- a/.gitignore +++ b/.gitignore @@ -13,3 +13,4 @@ gdpr_consent/node_modules/ *~ .streamlit/secrets.toml docs/superpowers/ +.venv/ \ No newline at end of file diff --git a/src/workflow/CommandExecutor.py b/src/workflow/CommandExecutor.py index 86f265b2..b15a193e 100644 --- a/src/workflow/CommandExecutor.py +++ b/src/workflow/CommandExecutor.py @@ -5,7 +5,7 @@ import threading from pathlib import Path from .Logger import Logger -from .ParameterManager import ParameterManager +from .ParameterManager import ParameterManager, bool_param_paths_from_param_xml_ini import sys import importlib.util import json @@ -216,7 +216,7 @@ def read_stderr(): stdout_thread.join() stderr_thread.join() - def run_topp(self, tool: str, input_output: dict, custom_params: dict = {}) -> bool: + def run_topp(self, tool: str, input_output: dict, custom_params: dict = {}, tool_instance_name: str = None) -> bool: """ Constructs and executes commands for the specified tool OpenMS TOPP tool based on the given input and output configurations. Ensures that all input/output file lists @@ -234,6 +234,10 @@ def run_topp(self, tool: str, input_output: dict, custom_params: dict = {}) -> b tool (str): The executable name or path of the tool. input_output (dict): A dictionary specifying the input/output parameter names (as key) and their corresponding file paths (as value). custom_params (dict): A dictionary of custom parameters to pass to the tool. + tool_instance_name (str, optional): Key for ``params.json`` when it differs + from ``tool`` (e.g. multiple instances). Defaults to ``tool``. + Custom parameters whose keys appear in the tool's ParamXML ``type="bool"`` + entries are passed as valueless CLI flags (``-name`` only when enabled). Returns: bool: True if all commands succeeded, False if any failed. @@ -261,8 +265,15 @@ def run_topp(self, tool: str, input_output: dict, custom_params: dict = {}) -> b commands = [] - # Load parameters for non-defaults params = self.parameter_manager.get_parameters_from_json() + + topp_tool_ini_path = Path(self.parameter_manager.ini_dir, f"{tool}.ini") + # Keys of type="bool" in the .ini: TOPP treats these as on/off flags (omit value when off) + topp_bool_flag_param_keys = ( + bool_param_paths_from_param_xml_ini(topp_tool_ini_path, tool) + if topp_tool_ini_path.exists() + else set() + ) # Construct commands for each process for i in range(n_processes): command = [tool] @@ -284,6 +295,16 @@ def run_topp(self, tool: str, input_output: dict, custom_params: dict = {}) -> b # Add non-default TOPP tool parameters if tool in params.keys(): for k, v in params[tool].items(): + + if k in topp_bool_flag_param_keys and v != "": + # CLI flag: include "-k" only when enabled + if isinstance(v, str): + is_enabled = v.lower() == "true" + else: + is_enabled = bool(v) + if is_enabled: + command += [f"-{k}"] + continue command += [f"-{k}"] # Skip only empty strings (pass flag with no value) # Note: 0 and 0.0 are valid values, so use explicit check @@ -295,6 +316,7 @@ def run_topp(self, tool: str, input_output: dict, custom_params: dict = {}) -> b # Add custom parameters for k, v in custom_params.items(): command += [f"-{k}"] + # Skip only empty strings (pass flag with no value) # Note: 0 and 0.0 are valid values, so use explicit check if v != "" and v is not None: diff --git a/src/workflow/ParameterManager.py b/src/workflow/ParameterManager.py index b0c36263..b8f490ee 100644 --- a/src/workflow/ParameterManager.py +++ b/src/workflow/ParameterManager.py @@ -3,8 +3,52 @@ import shutil import subprocess import streamlit as st +import xml.etree.ElementTree as ET from pathlib import Path + +def bool_param_paths_from_param_xml_ini(ini_path: Path, tool_stem: str) -> set[str]: + """ + Return short parameter paths for every ```` in a ParamXML .ini file. + + Paths match the suffix after ``Tool:1:`` in pyOpenMS (e.g. ``algorithm:epd:masstrace_snr_filtering``). + """ + try: + root = ET.parse(ini_path).getroot() + except (ET.ParseError, OSError): + return set() + + def local_tag(el: ET.Element) -> str: + t = el.tag + return t.rsplit("}", 1)[-1] if isinstance(t, str) and "}" in t else str(t) + + out: set[str] = set() + + def walk(el: ET.Element, parts: tuple[str, ...]) -> None: + for ch in el: + lt = local_tag(ch) + if lt == "NODE": + nm = ch.get("name") or "" + walk(ch, parts + (nm,)) + elif lt == "ITEM" and (ch.get("type") or "").lower() == "bool": + nm = ch.get("name") or "" + segs = [p for p in parts if p] + if nm: + segs.append(nm) + if not segs: + continue + # Strip tool root NODE name and instance NODE "1" (not part of pyOpenMS short keys) + while segs and segs[0] in (tool_stem, "1"): + segs.pop(0) + if segs: + out.add(":".join(segs)) + + for ch in root: + if local_tag(ch) == "NODE": + walk(ch, ()) + return out + + class ParameterManager: """ Manages the parameters for a workflow, including saving parameters to a JSON file, @@ -29,6 +73,29 @@ def __init__(self, workflow_dir: Path, workflow_name: str = None): # Store workflow name for preset loading; default to directory stem if not provided self.workflow_name = workflow_name or workflow_dir.stem + def bool_pairs_session_key(self) -> str: + """Session state key holding a set of (tool name, param path) for bool TOPP params.""" + return f"{self.ini_dir.parent.stem}-topp-bool-pairs" + + def get_bool_param_pairs(self) -> set: + """Return the cached set of (tool, param path) bool params; empty set if none.""" + return st.session_state.get(self.bool_pairs_session_key(), set()) + + def _merge_bool_params_from_ini(self, tool: str) -> None: + """Load tool.ini (XML) and merge type=bool parameter paths into session_state.""" + ini_path = Path(self.ini_dir, f"{tool}.ini") + if not ini_path.exists(): + return + try: + sk = self.bool_pairs_session_key() + if sk not in st.session_state: + st.session_state[sk] = set() + for short in bool_param_paths_from_param_xml_ini(ini_path, tool): + st.session_state[sk].add((tool, short)) + except RuntimeError: + # No Streamlit session (e.g. plain `python` import) + pass + def create_ini(self, tool: str) -> bool: """ Create an ini file for a TOPP tool if it doesn't exist. @@ -41,11 +108,14 @@ def create_ini(self, tool: str) -> bool: """ ini_path = Path(self.ini_dir, tool + ".ini") if ini_path.exists(): + self._merge_bool_params_from_ini(tool) return True try: subprocess.call([tool, "-write_ini", str(ini_path)]) except FileNotFoundError: return False + if ini_path.exists(): + self._merge_bool_params_from_ini(tool) return ini_path.exists() def save_parameters(self) -> None: