update: upload_widget

src/Workflow.py

@@ -22,7 +22,7 @@ def upload(self) -> None:
             self.ui.upload_widget(
                 key="mzML-files",
                 name="MS data",
-                file_type="mzML",
+                file_types="mzML",
                 fallback=[str(f) for f in Path("example-data", "mzML").glob("*.mzML")],
             )
 

src/common.py

@@ -15,6 +15,9 @@
 APP_NAME = "OpenMS Streamlit App"
 REPOSITORY_NAME = "streamlit-template"
 
+# Detect system platform
+OS_PLATFORM = sys.platform
+
 
 def load_params(default: bool = False) -> dict[str, Any]:
     """
@@ -111,6 +114,8 @@ def page_setup(page: str = "") -> dict[str, Any]:
         # Check location
         if "local" in sys.argv:
             st.session_state.location = "local"
+            st.session_state["previous_dir"] = os.getcwd()
+            st.session_state["local_dir"] = ""
         else:
             st.session_state.location = "online"
         # if we run the packaged windows version, we start within the Python directory -> need to change working directory to ..\streamlit-template

src/workflow/StreamlitUI.py

@@ -5,13 +5,22 @@
 import subprocess
 from typing import Any, Union, List
 import json
+import os
 import sys
 import importlib.util
 import time
 from io import BytesIO
 import zipfile
 from datetime import datetime
 
+try:
+    from tkinter import Tk, filedialog
+    TK_AVAILABLE = True
+except ImportError:
+    TK_AVAILABLE = False
+
+from ..common import OS_PLATFORM
+
 class StreamlitUI:
     """
     Provides an interface for Streamlit applications to handle file uploads,
@@ -28,12 +37,32 @@ def __init__(self, workflow_dir, logger, executor, paramter_manager):
         self.parameter_manager = paramter_manager
         self.params = self.parameter_manager.get_parameters_from_json()
 
+    def tk_directory_dialog(self, title: str = "Select Directory", parent_dir: str = os.getcwd()):
+        """
+        Creates a Tkinter directory dialog for selecting a directory.
+
+        Args:
+            title (str): The title of the directory dialog.
+            parent_dir (str): The path to the parent directory of the directory dialog.
+
+        Returns:
+            str: The path to the selected directory.
+
+        Warning:
+            This function is not avaliable in a streamlit cloud context.
+        """
+        root = Tk()
+        root.withdraw()
+        file_path = filedialog.askdirectory(title=title, initialdir=parent_dir)
+        root.destroy()
+        return file_path
+
     def upload_widget(
         self,
         key: str,
-        file_type: str,
+        file_types: Union[str, List[str]],
         name: str = "",
-        fallback: Union[List, str] = None,
+        fallback: Union[List, str] = None
     ) -> None:
         """
         Handles file uploads through the Streamlit interface, supporting both direct
@@ -42,17 +71,19 @@ def upload_widget(
 
         Args:
             key (str): A unique identifier for the upload component.
-            file_type (str): Expected file type for the uploaded files.
+            file_types (Union[str, List[str]]): Expected file type(s) for the uploaded files.
             name (str, optional): Display name for the upload component. Defaults to the key if not provided.
             fallback (Union[List, str], optional): Default files to use if no files are uploaded.
         """
         files_dir = Path(self.workflow_dir, "input-files", key)
-        
+
         # create the files dir
         files_dir.mkdir(exist_ok=True, parents=True)
 
         # check if only fallback files are in files_dir, if yes, reset the directory before adding new files
-        if [Path(f).name for f in Path(files_dir).iterdir()] == [Path(f).name for f in fallback]:
+        if [Path(f).name for f in Path(files_dir).iterdir()] == [
+            Path(f).name for f in fallback
+        ]:
             shutil.rmtree(files_dir)
             files_dir.mkdir()
 
@@ -62,10 +93,13 @@ def upload_widget(
         c1, c2 = st.columns(2)
         c1.markdown("**Upload file(s)**")
         with c1.form(f"{key}-upload", clear_on_submit=True):
-            if any(c.isupper() for c in file_type) and (c.islower() for c in file_type):
-                file_type_for_uploader = None
-            else:
-                file_type_for_uploader = [file_type]
+            # Convert file_types to a list if it's a string
+            if isinstance(file_types, str):
+                file_types = [file_types]
+
+            # Streamlit file uploader accepts file types as a list or None
+            file_type_for_uploader = file_types if file_types else None
+
             files = st.file_uploader(
                 f"{name}",
                 accept_multiple_files=(st.session_state.location == "local"),
@@ -80,9 +114,10 @@ def upload_widget(
                     if not isinstance(files, list):
                         files = [files]
                     for f in files:
-                        if f.name not in [
-                            f.name for f in files_dir.iterdir()
-                        ] and f.name.endswith(file_type):
+                        # Check if file type is in the list of accepted file types
+                        if f.name not in [f.name for f in files_dir.iterdir()] and any(
+                            f.name.endswith(ft) for ft in file_types
+                        ):
                             with open(Path(files_dir, f.name), "wb") as fh:
                                 fh.write(f.getbuffer())
                     st.success("Successfully added uploaded files!")
@@ -91,27 +126,67 @@ def upload_widget(
 
         # Local file upload option: via directory path
         if st.session_state.location == "local":
-            c2.markdown("**OR copy files from local folder**")
-            with c2.form(f"{key}-local-file-upload"):
-                local_dir = st.text_input(f"path to folder with **{name}** files")
-                if st.form_submit_button(
-                    f"Copy **{name}** files from local folder", use_container_width=True
-                ):
-                    # raw string for file paths
-                    if not any(Path(local_dir).glob(f"*.{file_type}")):
+            c2_text, c2_checkbox = c2.columns([1.5, 1], gap="large")
+            c2_text.markdown("**OR copy files from local folder**")
+            use_copy = c2_checkbox.checkbox("Make a copy of files", key=f"{key}-copy_files", value=True, help="Create a copy of files in workspace.")
+            with c2.container(border=True):
+                st_cols = st.columns([0.05, 0.55], gap="small")
+                with st_cols[0]:
+                    st.write("\n")
+                    st.write("\n")
+                    dialog_button = st.button("📁", key='local_browse', help="Browse for your local directory with MS data.", disabled=not TK_AVAILABLE)
+                    if dialog_button:
+                        st.session_state["local_dir"] = self.tk_directory_dialog("Select directory with your MS data", st.session_state["previous_dir"])
+                        st.session_state["previous_dir"] = st.session_state["local_dir"]
+
+                with st_cols[1]:
+                    local_dir = st.text_input(f"path to folder with **{name}** files", value=st.session_state["local_dir"])
+
+                if c2.button(f"Copy **{name}** files from local folder", use_container_width=True):
+                    files = []
+                    local_dir = Path(
+                        local_dir
+                    ).expanduser()  # Expand ~ to full home directory path
+
+                    for ft in file_types:
+                        # Search for both files and directories with the specified extension
+                        for path in local_dir.iterdir():
+                            if path.is_file() and path.name.endswith(f".{ft}"):
+                                files.append(path)
+                            elif path.is_dir() and path.name.endswith(f".{ft}"):
+                                files.append(path)
+
+                    if not files:
                         st.warning(
-                            f"No files with type **{file_type}** found in specified folder."
+                            f"No files with type **{', '.join(file_types)}** found in specified folder."
                         )
                     else:
-                        # Copy all mzML files to workspace mzML directory, add to selected files
-                        files = list(Path(local_dir).glob("*.mzML"))
                         my_bar = st.progress(0)
                         for i, f in enumerate(files):
                             my_bar.progress((i + 1) / len(files))
-                            shutil.copy(f, Path(files_dir, f.name))
+                            if use_copy:
+                                if os.path.isfile(f):
+                                    shutil.copy(f, Path(files_dir, f.name))
+                                elif os.path.isdir(f):
+                                    shutil.copytree(f, Path(files_dir, f.name), dirs_exist_ok=True)
+                            else:
+                                # Do we need to change the reference of Path(st.session_state.workspace, "mzML-files") to point to local dir? 
+                                pass
                         my_bar.empty()
                         st.success("Successfully copied files!")
-
+
+            if not TK_AVAILABLE:
+                c2.warning("**Warning**: Failed to import tkinter, either it is not installed, or this is being called from a cloud context. " "This function is not available in a Streamlit Cloud context. "
+                "You will have to manually enter the path to the folder with the MS files."
+                           )
+
+            if not use_copy:
+                c2.warning(
+        "**Warning**: You have deselected the `Make a copy of files` option. "
+        "This **_assumes you know what you are doing_**. "
+        "This means that the original files will be used instead. "
+    )
+
         if fallback and not any(Path(files_dir).iterdir()):
             if isinstance(fallback, str):
                 fallback = [fallback]
@@ -144,12 +219,14 @@ def upload_widget(
             ):
                 shutil.rmtree(files_dir)
                 del self.params[key]
-                with open(self.parameter_manager.params_file, "w", encoding="utf-8") as f:
+                with open(
+                    self.parameter_manager.params_file, "w", encoding="utf-8"
+                ) as f:
                     json.dump(self.params, f, indent=4)
                 st.rerun()
         elif not fallback:
             st.warning(f"No **{name}** files!")
-
+            
     def select_input_file(
         self,
         key: str,
@@ -398,7 +475,6 @@ def input_TOPP(
                     if encoded_key in param.keys():
                         param.setValue(encoded_key, value)
                 poms.ParamXMLFile().store(str(ini_file_path), param)
-
 
         # read into Param object
         param = poms.Param()
@@ -431,7 +507,7 @@ def input_TOPP(
                 "section_description": param.getSectionDescription(':'.join(key.decode().split(':')[:-1]))
             }
             params_decoded.append(tmp)
-                    
+
         # 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
@@ -449,19 +525,19 @@ def input_TOPP(
         section_description = None
         cols = st.columns(num_cols)
         i = 0
-        
+
         for p in params_decoded:
             # skip avdanced parameters if not selected
             if not st.session_state["advanced"] and p["advanced"]:
                 continue
 
             key = f"{self.parameter_manager.topp_param_prefix}{p['key'].decode()}"
             if display_subsections:
-                 name = p["name"]
-                 if section_description is None:
+                name = p["name"]
+                if section_description is None:
                     section_description = p['section_description']
-                    
-                 elif section_description != p['section_description']:
+
+                elif section_description != p['section_description']:
                     section_description = p['section_description']
                     st.markdown(f"**{section_description}**")
                     cols = st.columns(num_cols)
@@ -638,7 +714,7 @@ def zip_and_download_files(self, directory: str):
         Args:
             directory (str): The directory whose files are to be zipped.
         """
-       # Ensure directory is a Path object and check if directory is empty
+        # Ensure directory is a Path object and check if directory is empty
         directory = Path(directory)
         if not any(directory.iterdir()):
             st.error("No files to compress.")
@@ -675,8 +751,7 @@ def zip_and_download_files(self, directory: str):
             mime="application/zip",
             use_container_width=True
         )
-
-
+
     def file_upload_section(self, custom_upload_function) -> None:
         custom_upload_function()
         c1, _ = st.columns(2)