Skip to content

Fix tool_instance_name resolution in save_parameters and support multiple TOPP tool instances #354

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
Copilot wants to merge 2 commits into
base: main
Choose a base branch
from
Draft

Fix tool_instance_name resolution in save_parameters and support multiple TOPP tool instances #354

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
5949251
Fix tool instance name handling and move custom defaults to params.json
t0mdavid-m Apr 7, 2026
e3c3665
Remove redundant -ini flag from run_topp
t0mdavid-m Apr 7, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions .streamlit/config.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,8 @@ developmentMode = false
address = "0.0.0.0"
maxUploadSize = 200 #MB
port = 8501 # should be same as configured in deployment repo
enableCORS = false
enableXsrfProtection = false


[theme]
Expand Down
37 changes: 18 additions & 19 deletions src/workflow/CommandExecutor.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -234,6 +234,9 @@ 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): A unique instance name for this tool
invocation, used for parameter lookup when multiple instances of the
same tool exist. If not provided, defaults to the tool name.

Returns:
bool: True if all commands succeeded, False if any failed.
Expand All @@ -242,6 +245,8 @@ def run_topp(self, tool: str, input_output: dict, custom_params: dict = {}) -> b
ValueError: If the lengths of input/output file lists are inconsistent,
except for single string inputs.
"""
# Use tool_instance_name for parameter lookup, fall back to tool name
params_key = tool_instance_name if tool_instance_name else tool
# check input: any input lists must be same length, other items can be a single string
# e.g. input_mzML : [list of n mzML files], output_featureXML : [list of n featureXML files], input_database : database.tsv
io_lengths = [len(v) for v in input_output.values() if len(v) > 1]
Expand All @@ -261,8 +266,8 @@ 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()
# Load merged parameters (_defaults + user overrides) for this tool instance
merged_params = self.parameter_manager.get_merged_params(params_key)
# Construct commands for each process
for i in range(n_processes):
command = [tool]
Expand All @@ -281,17 +286,16 @@ def run_topp(self, tool: str, input_output: dict, custom_params: dict = {}) -> b
# standard case, files was a list of strings, take the file name at index
else:
command += [value[i]]
# Add non-default TOPP tool parameters
if tool in params.keys():
for k, v in params[tool].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:
if isinstance(v, str) and "\n" in v:
command += v.split("\n")
else:
command += [str(v)]
# Add merged TOPP tool parameters (_defaults + user overrides)
for k, v in merged_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:
if isinstance(v, str) and "\n" in v:
command += v.split("\n")
else:
command += [str(v)]
# Add custom parameters
for k, v in custom_params.items():
command += [f"-{k}"]
Expand All @@ -306,11 +310,6 @@ def run_topp(self, tool: str, input_output: dict, custom_params: dict = {}) -> b
command += ["-threads", str(threads_per_command)]
commands.append(command)

# check if a ini file has been written, if yes use it (contains custom defaults)
ini_path = Path(self.parameter_manager.ini_dir, tool + ".ini")
if ini_path.exists():
command += ["-ini", str(ini_path)]

# Run command(s)
if len(commands) == 1:
return self.run_command(commands[0])
Expand Down
72 changes: 53 additions & 19 deletions src/workflow/ParameterManager.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -65,7 +65,7 @@ def save_parameters(self) -> None:
# Advanced parameters are only in session state if the view is active
json_params = self.get_parameters_from_json() | json_params

# get a list of TOPP tools which are in session state
# get a list of TOPP tools (or tool instance names) which are in session state
current_topp_tools = list(
set(
[
Expand All @@ -75,12 +75,16 @@ def save_parameters(self) -> None:
]
)
)
# for each TOPP tool, open the ini file
# Retrieve the instance-name → real-tool-name mapping (set by input_TOPP)
tool_instance_map = st.session_state.get("_topp_tool_instance_map", {})
# for each TOPP tool (or instance name), open the ini file
for tool in current_topp_tools:
if not self.create_ini(tool):
# Resolve instance name to real tool name for create_ini / ini loading
real_tool = tool_instance_map.get(tool, tool)
if not self.create_ini(real_tool):
# Could not create ini file - skip this tool
continue
ini_path = Path(self.ini_dir, f"{tool}.ini")
ini_path = Path(self.ini_dir, f"{real_tool}.ini")
if tool not in json_params:
json_params[tool] = {}
# load the param object
Expand All @@ -92,19 +96,26 @@ def save_parameters(self) -> None:
# Skip display keys used by multiselect widgets
if key.endswith("_display"):
continue
# get ini_key
ini_key = key.replace(self.topp_param_prefix, "").encode()
# get ini_key – map instance name back to real tool name
ini_key = key.replace(self.topp_param_prefix, "")
if tool != real_tool:
ini_key = ini_key.replace(f"{tool}:1:", f"{real_tool}:1:", 1)
ini_key = ini_key.encode()
# get ini (default) value by ini_key
ini_value = param.getValue(ini_key)
is_list_param = isinstance(ini_value, list)
# check if value is different from default OR is an empty list parameter
# Effective default: _defaults value if present, else ini value
short_key = key.split(":1:")[1]
defaults = json_params.get("_defaults", {}).get(tool, {})
default_value = defaults.get(short_key, ini_value)
# check if value is different from effective default OR is an empty list parameter
if (
(ini_value != value)
or (key.split(":1:")[1] in json_params[tool])
(default_value != value)
or (short_key in json_params[tool])
or (is_list_param and not value) # Always save empty list params
):
# store non-default value
json_params[tool][key.split(":1:")[1]] = value
json_params[tool][short_key] = value
# Save to json file
with open(self.params_file, "w", encoding="utf-8") as f:
json.dump(json_params, f, indent=4)
Expand All @@ -130,17 +141,44 @@ def get_parameters_from_json(self) -> dict:
st.error("**ERROR**: Attempting to load an invalid JSON parameter file. Reset to defaults.")
return {}

def get_topp_parameters(self, tool: str) -> dict:
def get_merged_params(self, tool_instance_name: str, ini_params: dict = None) -> dict:
"""
Three-layer parameter merge: ini defaults < _defaults < user overrides.

Args:
tool_instance_name: Instance name (or tool name) to look up in params.json.
ini_params: Base parameters from the .ini file. Optional — callers that
don't need the ini layer (e.g., run_topp, which passes -ini separately)
can omit this.

Returns:
Merged dict with the effective value for each parameter.
"""
params = self.get_parameters_from_json()
defaults = params.get("_defaults", {}).get(tool_instance_name, {})
user = params.get(tool_instance_name, {})

merged = {}
if ini_params:
merged.update(ini_params)
merged.update(defaults)
merged.update(user)
return merged

def get_topp_parameters(self, tool: str, tool_instance_name: str = None) -> dict:
"""
Get all parameters for a TOPP tool, merging defaults with user values.

Args:
tool: Name of the TOPP tool (e.g., "CometAdapter")
tool: Name of the TOPP tool executable (e.g., "CometAdapter")
tool_instance_name: Optional instance name used for parameter storage
(e.g., "IDFilter_step1"). If not provided, defaults to tool name.

Returns:
Dict with parameter names as keys (without tool prefix) and their values.
Returns empty dict if ini file doesn't exist.
"""
instance_name = tool_instance_name or tool
ini_path = Path(self.ini_dir, f"{tool}.ini")
if not ini_path.exists():
return {}
Expand All @@ -151,18 +189,14 @@ def get_topp_parameters(self, tool: str) -> dict:

# Build dict from ini (extract short key names)
prefix = f"{tool}:1:"
full_params = {}
ini_params = {}
for key in param.keys():
key_str = key.decode() if isinstance(key, bytes) else str(key)
if prefix in key_str:
short_key = key_str.split(prefix, 1)[1]
full_params[short_key] = param.getValue(key)

# Override with user-modified values from JSON
user_params = self.get_parameters_from_json().get(tool, {})
full_params.update(user_params)
ini_params[short_key] = param.getValue(key)

return full_params
return self.get_merged_params(instance_name, ini_params=ini_params)

def reset_to_default_parameters(self) -> None:
"""
Expand Down
78 changes: 52 additions & 26 deletions src/workflow/StreamlitUI.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -616,6 +616,7 @@ def input_TOPP(
display_subsections: bool = True,
display_subsection_tabs: bool = False,
custom_defaults: dict = {},
tool_instance_name: str = None,
) -> None:
"""
Generates input widgets for TOPP tool parameters dynamically based on the tool's
Expand All @@ -631,29 +632,43 @@ def input_TOPP(
display_subsections (bool, optional): Whether to split parameters into subsections based on the prefix. Defaults to True.
display_subsection_tabs (bool, optional): Whether to display main subsections in separate tabs (if more than one main section). Defaults to False.
custom_defaults (dict, optional): Dictionary of custom defaults to use. Defaults to an empty dict.
tool_instance_name (str, optional): A unique instance name for this tool
invocation. Allows multiple instances of the same TOPP tool with
independent parameters (e.g., two IDFilter calls). If not provided,
defaults to topp_tool_name. The instance name is used for session
state keys and parameter storage, while topp_tool_name is used for
the actual tool executable and ini file creation.
"""
# Default instance name to the tool name when not provided
if tool_instance_name is None:
tool_instance_name = topp_tool_name

# Register instance-name → real-tool-name mapping in session state
if "_topp_tool_instance_map" not in st.session_state:
st.session_state["_topp_tool_instance_map"] = {}
st.session_state["_topp_tool_instance_map"][tool_instance_name] = topp_tool_name

if not display_subsections:
display_subsection_tabs = False
if display_subsection_tabs:
display_subsections = True

# write defaults ini files
# Create pristine ini file (never mutated with custom defaults)
ini_file_path = Path(self.parameter_manager.ini_dir, f"{topp_tool_name}.ini")
ini_existed = ini_file_path.exists()
if not self.parameter_manager.create_ini(topp_tool_name):
st.error(f"TOPP tool **'{topp_tool_name}'** not found.")
return
if not ini_existed:
# update custom defaults if necessary
if custom_defaults:
param = poms.Param()
poms.ParamXMLFile().load(str(ini_file_path), param)
for key, value in custom_defaults.items():
encoded_key = f"{topp_tool_name}:1:{key}".encode()
if encoded_key in param.keys():
param.setValue(encoded_key, value)
poms.ParamXMLFile().store(str(ini_file_path), param)

# Seed custom defaults into params.json under _defaults key
if custom_defaults:
params = self.parameter_manager.get_parameters_from_json()
if "_defaults" not in params:
params["_defaults"] = {}
params["_defaults"][tool_instance_name] = custom_defaults
with open(self.parameter_manager.params_file, "w", encoding="utf-8") as f:
json.dump(params, f, indent=4)
# Refresh self.params so widget resolution sees _defaults
self.params = self.parameter_manager.get_parameters_from_json()

# read into Param object
param = poms.Param()
Expand Down Expand Up @@ -730,18 +745,18 @@ def _matches_parameter(pattern: str, key: bytes) -> bool:
)
params.append(p)

# for each parameter in params_decoded
# if a parameter with custom default value exists, use that value
# else check if the parameter is already in self.params, if yes take the value from self.params
# Build ini_params dict for three-layer merge
ini_params = {}
for p in params:
name = p["key"].decode().split(":1:")[1]
ini_params[name] = p["value"]

# Resolve effective values: ini < _defaults < user overrides
merged = self.parameter_manager.get_merged_params(tool_instance_name, ini_params=ini_params)
for p in params:
name = p["key"].decode().split(":1:")[1]
if topp_tool_name in self.params:
if name in self.params[topp_tool_name]:
p["value"] = self.params[topp_tool_name][name]
elif name in custom_defaults:
p["value"] = custom_defaults[name]
elif name in custom_defaults:
p["value"] = custom_defaults[name]
if name in merged:
p["value"] = merged[name]
# Ensure list parameters stay as lists after loading from JSON
# (JSON may store single-item lists as strings)
if p["original_is_list"] and isinstance(p["value"], str):
Expand Down Expand Up @@ -775,7 +790,7 @@ def _matches_parameter(pattern: str, key: bytes) -> bool:

# Display tool name if required
if display_tool_name:
st.markdown(f"**{topp_tool_name}**")
st.markdown(f"**{tool_instance_name}**")

tab_names = [k for k in param_sections.keys() if ":" not in k]
tabs = None
Expand Down Expand Up @@ -803,8 +818,11 @@ def display_TOPP_params(params: dict, num_cols):
cols = st.columns(num_cols)
i = 0
for p in params:
# get key and name
key = f"{self.parameter_manager.topp_param_prefix}{p['key'].decode()}"
# get key and name – use tool_instance_name in session state key
key_str = p['key'].decode()
if tool_instance_name != topp_tool_name:
key_str = key_str.replace(f"{topp_tool_name}:1:", f"{tool_instance_name}:1:", 1)
key = f"{self.parameter_manager.topp_param_prefix}{key_str}"
name = p["name"]
try:
# sometimes strings with newline, handle as list
Expand Down Expand Up @@ -1371,7 +1389,8 @@ def remove_full_paths(d: dict) -> dict:
general = {}

for k, v in params.items():
# skip if v is a file path
if k == "_defaults":
continue
if isinstance(v, dict):
topp[k] = v
elif ".py" in k:
Expand All @@ -1382,6 +1401,13 @@ def remove_full_paths(d: dict) -> dict:
else:
general[k] = v

# Merge _defaults into topp so summary shows custom defaults + user overrides
defaults = params.get("_defaults", {})
for tool_name, default_vals in defaults.items():
if tool_name not in topp:
topp[tool_name] = {}
topp[tool_name] = {**default_vals, **topp.get(tool_name, {})}

markdown = []

def dict_to_markdown(d: dict):
Expand Down
Loading
Loading