configbuilder: add pipeline config screens, fix validation code

nf_core/configs/create/__init__.py

@@ -20,6 +20,9 @@
 from nf_core.configs.create.hpcquestion import ChooseHpc
 from nf_core.configs.create.nfcorequestion import ChooseNfcoreConfig
 from nf_core.configs.create.welcome import WelcomeScreen
+from nf_core.configs.create.pipelineconfigquestion import PipelineConfigQuestion
+from nf_core.configs.create.defaultprocessres import DefaultProcess
+from nf_core.configs.create.multiprocessres import MultiNamedProcessConfig, MultiLabelledProcessConfig
 
 ## General utilities
 from nf_core.utils import LoggingConsole
@@ -59,6 +62,10 @@ class ConfigsCreateApp(App[utils.ConfigsCreateConfig]):
         "final": FinalScreen,
         "hpc_question": ChooseHpc,
         "hpc_customisation": HpcCustomisation,
+        "pipeline_config_question": PipelineConfigQuestion,
+        "default_process_resources": DefaultProcess,
+        "multi_named_process_config": MultiNamedProcessConfig,
+        "multi_labelled_process_config": MultiLabelledProcessConfig,
         "final_infra_details": FinalInfraDetails,
     }
 
@@ -68,6 +75,9 @@ class ConfigsCreateApp(App[utils.ConfigsCreateConfig]):
     # Tracking variables
     CONFIG_TYPE = None
     NFCORE_CONFIG = True
+    INFRA_ISHPC = False
+    PIPE_CONF_NAMED = False
+    PIPE_CONF_LABELLED = False
 
     # Log handler
     LOG_HANDLER = rich_log_handler
@@ -99,18 +109,32 @@ def on_button_pressed(self, event: Button.Pressed) -> None:
             utils.NFCORE_CONFIG_GLOBAL = False
             self.push_screen("basic_details")
         elif event.button.id == "type_hpc":
+            self.INFRA_ISHPC = True
+            utils.INFRA_ISHPC_GLOBAL = True
             self.push_screen("hpc_customisation")
+        elif event.button.id == "type_local":
+            self.INFRA_ISHPC = False
+            utils.INFRA_ISHPC_GLOBAL = False
+            self.push_screen("final_infra_details")
         elif event.button.id == "toconfiguration":
             self.push_screen("final_infra_details")
         elif event.button.id == "finish":
             self.push_screen("final")
         ## General options
-        if event.button.id == "close_app":
-            self.exit(return_code=0)
         if event.button.id == "back":
             self.pop_screen()
 
+    def close_app(self):
+        self.exit(return_code=0)
+
     ## User theme options
     def action_toggle_dark(self) -> None:
         """An action to toggle dark mode."""
         self.theme: str = "textual-dark" if self.theme == "textual-light" else "textual-light"
+
+    def get_context(self):
+        return {
+            "is_nfcore": self.NFCORE_CONFIG,
+            "is_infrastructure": self.CONFIG_TYPE == "infrastructure",
+            "is_hpc": self.INFRA_ISHPC
+        }

nf_core/configs/create/basicdetails.py

@@ -12,6 +12,7 @@
 from nf_core.configs.create.utils import (
     ConfigsCreateConfig,
     TextInput,
+    init_context
 )  ## TODO Move somewhere common?
 from nf_core.utils import add_hide_class, remove_hide_class
 
@@ -101,12 +102,13 @@ def on_button_pressed(self, event: Button.Pressed) -> None:
             else:
                 text_input.query_one(".validation_msg").update("")
         try:
-            self.parent.TEMPLATE_CONFIG = ConfigsCreateConfig(**config)
+            with init_context(self.parent.get_context()):
+                self.parent.TEMPLATE_CONFIG = ConfigsCreateConfig(**config)
             if event.button.id == "next":
                 if self.parent.CONFIG_TYPE == "infrastructure":
                     self.parent.push_screen("hpc_question")
                 elif self.parent.CONFIG_TYPE == "pipeline":
-                    self.parent.push_screen("final")
+                    self.parent.push_screen("pipeline_config_question")
         except ValueError:
             pass
 

nf_core/configs/create/create.py

@@ -3,51 +3,172 @@
 """
 
 from nf_core.configs.create.utils import ConfigsCreateConfig, generate_config_entry
+from re import sub
+from pathlib import Path
 
 
 class ConfigCreate:
-    def __init__(self, template_config: ConfigsCreateConfig):
+    def __init__(self, template_config: ConfigsCreateConfig, config_type: str, config_dir: Path | str = Path('.')):
         self.template_config = template_config
+        self.config_type = config_type
+        config_dir_path = config_dir if isinstance(config_dir, Path) else Path(config_dir)
+        assert not config_dir_path.is_file(), f'Error: the path "{str(config_dir_path)}" is a file.'
+        # Create directory if it doesn't already exist
+        config_dir_path.mkdir(parents=True, exist_ok=True)
+        self.config_dir = config_dir_path
 
-    def construct_params(self, contact, handle, description, url):
+    def construct_info_params(self):
         final_params = {}
+        contact = self.template_config.config_profile_contact
+        handle = self.template_config.config_profile_handle
+        description = self.template_config.config_profile_description
+        url = self.template_config.config_profile_url
 
-        if contact is not None:
-            if handle is not None:
+        if contact:
+            if handle:
                 config_contact = contact + " (" + handle + ")"
             else:
                 config_contact = contact
             final_params["config_profile_contact"] = config_contact
-        elif handle is not None:
+        elif handle:
             final_params["config_profile_contact"] = handle
 
-        if description is not None:
+        if description:
             final_params["config_profile_description"] = description
 
-        if url is not None:
+        if url:
             final_params["config_profile_url"] = url
 
         return final_params
 
+    def construct_params_str(self):
+        info_params = self.construct_info_params()
+
+        info_params_str_list = [
+            f'  {key} = "{value}"'
+            for key, value in info_params.items()
+            if value
+        ]
+
+        params_section = [
+            'params {',
+            '\n',
+            *info_params_str_list,
+            '\n',
+            '}',
+        ]
+
+        params_section_str = '\n'.join(params_section) + '\n\n'
+        return sub(r'\n\n\n+', '\n\n', params_section_str)
+
+    def get_resource_strings(self, cpus, memory, hours, prefix=''):
+        cpus_str = ''
+        if cpus:
+            cpus_int = int(cpus)
+            cpus_str = f'cpus = {cpus_int}'
+
+        memory_str = ''
+        if memory:
+            memory_int = int(memory)
+            memory_str = f'memory = {memory_int}.GB'
+
+        time_str = ''
+        if hours:
+            time_h = float(hours)
+            if time_h.is_integer():
+                time_h = int(time_h)
+                time_str = f"time = {time_h}.h"
+            else:
+                time_m = int(time_h * 60)
+                time_str = f"time = {time_m}.m"
+
+        resources = [cpus_str, memory_str, time_str]
+        return [
+            f'{prefix}{res}'
+            for res in resources
+            if res
+        ]
+
+    def construct_process_config_str(self):
+        # Construct default resources
+        default_resources = self.get_resource_strings(
+            cpus=self.template_config.default_process_ncpus,
+            memory=self.template_config.default_process_memgb,
+            hours=self.template_config.default_process_hours,
+            prefix='  '
+        )
+
+        # Construct named process resources
+        named_resources = []
+        if self.template_config.named_process_resources:
+            for process_name, process_resources in self.template_config.named_process_resources.items():
+                named_resource_string = self.get_resource_strings(
+                    cpus=process_resources['custom_process_ncpus'],
+                    memory=process_resources['custom_process_memgb'],
+                    hours=process_resources['custom_process_hours'],
+                    prefix='    '
+                )
+                if not named_resource_string:
+                    continue
+                named_resources.append(
+                    f"  withName: '{process_name}'" + " {"
+                )
+                named_resources.extend(named_resource_string)
+                named_resources.append('  }')
+                named_resources.append('\n')
+
+        # Construct labelled process resources
+        labelled_resources = []
+        if self.template_config.labelled_process_resources:
+            for process_label, process_resources in self.template_config.labelled_process_resources.items():
+                labelled_resource_string = self.get_resource_strings(
+                    cpus=process_resources['custom_process_ncpus'],
+                    memory=process_resources['custom_process_memgb'],
+                    hours=process_resources['custom_process_hours'],
+                    prefix='    '
+                )
+                if not labelled_resource_string:
+                    continue
+                labelled_resources.append(
+                    f"  withLabel: '{process_label}'" + " {"
+                )
+                labelled_resources.extend(labelled_resource_string)
+                labelled_resources.append('  }')
+                labelled_resources.append('\n')
+
+        process_section = [
+            'process {',
+            '\n',
+            *default_resources,
+            '\n',
+            *named_resources,
+            '\n',
+            *labelled_resources,
+            '\n',
+            '}',
+        ]
+
+        process_section_str = '\n'.join(process_section) + '\n'
+
+        return sub(r'\n\n\n+', '\n\n', process_section_str)
+
     def write_to_file(self):
         ## File name option
-        filename = "_".join(self.template_config.general_config_name) + ".conf"
+        config_name = str(self.template_config.general_config_name).strip()
+        config_name_clean = sub(r'\W+', '_', config_name)
+        config_name_clean = sub(r'_+$', '', config_name_clean)
+        filename = f'{config_name_clean}.conf'
+        filename = self.config_dir / filename
 
         ## Collect all config entries per scope, for later checking scope needs to be written
-        validparams = self.construct_params(
-            self.template_config.config_profile_contact,
-            self.template_config.config_profile_handle,
-            self.template_config.config_profile_description,
-            self.template_config.config_profile_url,
-        )
+        params_section_str = self.construct_params_str()
+
+        if self.config_type == 'pipeline':
+            process_section_str = self.construct_process_config_str()
+        else:
+            process_section_str = ''
 
         with open(filename, "w+") as file:
             ## Write params
-            if any(validparams):
-                file.write("params {\n")
-                for entry_key, entry_value in validparams.items():
-                    if entry_value != "":
-                        file.write(generate_config_entry(self, entry_key, entry_value))
-                    else:
-                        continue
-                file.write("}\n")
+            file.write(params_section_str)
+            file.write(process_section_str)

nf_core/configs/create/defaultprocessres.py

@@ -0,0 +1,97 @@
+"""Get information about which process/label the user wants to configure."""
+
+from textwrap import dedent
+
+from textual import on
+from textual.app import ComposeResult
+from textual.containers import Center, Horizontal
+from textual.screen import Screen
+from textual.widgets import Button, Footer, Header, Input, Markdown
+
+from nf_core.configs.create.utils import (
+    ConfigsCreateConfig,
+    TextInput,
+    init_context
+)  ## TODO Move somewhere common?
+from nf_core.utils import add_hide_class, remove_hide_class
+
+
+class DefaultProcess(Screen):
+    """Get default process resource requirements."""
+
+    def compose(self) -> ComposeResult:
+        yield Header()
+        yield Footer()
+        yield Markdown(
+            dedent(
+                """
+                # Default process resources
+                """
+            )
+        )
+        yield TextInput(
+            "default_process_ncpus",
+            "2",
+            "Number of CPUs to use by default for all processes.",
+            "2",
+            classes="column",
+        )
+        yield TextInput(
+            "default_process_memgb",
+            "8",
+            "Amount of memory in GB to use by default for all processes.",
+            "8",
+            classes="column",
+        )
+        yield TextInput(
+            "default_process_hours",
+            "1",
+            "The default number of hours of walltime required for processes:",
+            "1",
+            classes="column",
+        )
+        yield Center(
+            Button("Back", id="back", variant="default"),
+            Button("Skip", id="skip", variant="default"),
+            Button("Next", id="next", variant="success"),
+            classes="cta",
+        )
+
+    @on(Button.Pressed, "#skip")
+    def skip_to_next_screen(self) -> None:
+        # Skip to the next screen without saving
+        if self.parent.PIPE_CONF_NAMED:
+            self.parent.push_screen("multi_named_process_config")
+        elif self.parent.PIPE_CONF_LABELLED:
+            self.parent.push_screen("multi_labelled_process_config")
+        else:
+            self.parent.push_screen("final")
+
+    # Updates the __init__ initialised TEMPLATE_CONFIG object (which is built from the ConfigsCreateConfig class) with the values from the text inputs
+    @on(Button.Pressed, "#next")
+    def on_button_pressed(self, event: Button.Pressed) -> None:
+        """Save fields to the config."""
+        new_config = {}
+        for text_input in self.query("TextInput"):
+            this_input = text_input.query_one(Input)
+            validation_result = this_input.validate(this_input.value)
+            new_config[text_input.field_id] = this_input.value
+            if not validation_result.is_valid:
+                text_input.query_one(".validation_msg").update("\n".join(validation_result.failure_descriptions))
+            else:
+                text_input.query_one(".validation_msg").update("")
+        try:
+            with init_context(self.parent.get_context()):
+                # First, validate the new config data
+                ConfigsCreateConfig(**new_config)
+                # If that passes validation, update the existing config
+                self.parent.TEMPLATE_CONFIG = self.parent.TEMPLATE_CONFIG.model_copy(update=new_config)
+            # Push the next screen
+            if self.parent.PIPE_CONF_NAMED:
+                self.parent.push_screen("multi_named_process_config")
+            elif self.parent.PIPE_CONF_LABELLED:
+                self.parent.push_screen("multi_labelled_process_config")
+            else:
+                self.parent.push_screen("final")
+        except ValueError:
+            pass