diff --git a/nf_core/configs/create/__init__.py b/nf_core/configs/create/__init__.py index ef8bd8b92f..371364e120 100644 --- a/nf_core/configs/create/__init__.py +++ b/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 + } diff --git a/nf_core/configs/create/basicdetails.py b/nf_core/configs/create/basicdetails.py index d9fb416c72..7214e397b1 100644 --- a/nf_core/configs/create/basicdetails.py +++ b/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 diff --git a/nf_core/configs/create/create.py b/nf_core/configs/create/create.py index 0d58551089..6c37e42c10 100644 --- a/nf_core/configs/create/create.py +++ b/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) diff --git a/nf_core/configs/create/defaultprocessres.py b/nf_core/configs/create/defaultprocessres.py new file mode 100644 index 0000000000..bf149154af --- /dev/null +++ b/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 diff --git a/nf_core/configs/create/final.py b/nf_core/configs/create/final.py index 457293dd1c..9ba08e56f6 100644 --- a/nf_core/configs/create/final.py +++ b/nf_core/configs/create/final.py @@ -2,12 +2,12 @@ from textual.app import ComposeResult from textual.containers import Center from textual.screen import Screen -from textual.widgets import Button, Footer, Header, Markdown +from textual.widgets import Button, Footer, Header, Markdown, Input from nf_core.configs.create.create import ( ConfigCreate, ) -from nf_core.configs.create.utils import TextInput +from nf_core.configs.create.utils import TextInput, ConfigsCreateConfig, init_context class FinalScreen(Screen): @@ -28,14 +28,28 @@ def compose(self) -> ComposeResult: ".", classes="row", ) - yield Center(Button("Save and close!", id="close_app", variant="success"), classes="cta") + yield Center( + Button("Back", id="back", variant="default"), + Button("Save and close!", id="close_app", variant="success"), + classes="cta" + ) - def _create_config(self) -> None: + def _create_config(self, config_dir='.') -> None: """Create the config.""" - create_obj = ConfigCreate(template_config=self.parent.TEMPLATE_CONFIG) + create_obj = ConfigCreate(template_config=self.parent.TEMPLATE_CONFIG, config_type=self.parent.CONFIG_TYPE, config_dir=config_dir) create_obj.write_to_file() @on(Button.Pressed, "#close_app") def on_button_pressed(self, event: Button.Pressed) -> None: """Save fields to the config.""" - self._create_config() + # Validate the save location + save_location = self.query_one("TextInput") + save_location_text = save_location.query_one(Input) + try: + with init_context(self.parent.get_context()): + ConfigsCreateConfig(savelocation=save_location_text.value) + # If validation passes, create the config + self._create_config(config_dir=save_location_text.value) + self.parent.close_app() + except ValueError: + pass diff --git a/nf_core/configs/create/multiprocessres.py b/nf_core/configs/create/multiprocessres.py new file mode 100644 index 0000000000..0e38c81b59 --- /dev/null +++ b/nf_core/configs/create/multiprocessres.py @@ -0,0 +1,174 @@ +"""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, HorizontalGroup, VerticalScroll +from textual.screen import Screen +from textual.widgets import Button, Footer, Header, Input, Markdown, Switch, Label +from textual.events import Mount, ScreenResume +from nf_core.utils import add_hide_class, remove_hide_class + +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 ProcessConfig(HorizontalGroup): + """Get resource requirements for a single process.""" + + def __init__(self, selector: str) -> None: + super().__init__() + assert selector in ['name', 'label'] + self.selector = selector + + def compose(self) -> ComposeResult: + yield TextInput( + "custom_process_id", + "", + f"Process {self.selector}:", + "", + classes="column", + ) + yield TextInput( + "custom_process_ncpus", + "2", + "# CPUs:", + "2", + classes="column", + ) + yield TextInput( + "custom_process_memgb", + "8", + "Memory (GB):", + "8", + classes="column", + ) + yield TextInput( + "custom_process_hours", + "1", + "Walltime (hours):", + "1", + classes="column", + ) + yield Button( + "-", + id="remove", + variant="error" + ) + + @on(Button.Pressed, "#remove") + def remove_widget(self) -> None: + self.remove() + + +class MultiProcessConfig(Screen): + """Get resource requirements for multiple processes.""" + + def __init__(self, selector_type: str, config_key: str, title: str) -> None: + super().__init__() + assert isinstance(title, str) and title + self.title = title + assert isinstance(selector_type, str) and selector_type + self.selector_type = selector_type + assert isinstance(config_key, str) and config_key + self.config_key = config_key + + def _set_next_screen(self, next_screen: str) -> None: + assert isinstance(next_screen, str) + self.next_screen = next_screen + + def compose(self) -> ComposeResult: + yield Header() + yield Footer() + yield Markdown(f'# {self.title}') + yield VerticalScroll( + ProcessConfig(selector=self.selector_type), + ProcessConfig(selector=self.selector_type), + ProcessConfig(selector=self.selector_type), + id="configs" + ) + yield Center( + Button("Add another process", id="another", variant="success") + ) + 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, "#another") + def add_config(self) -> None: + new_config = ProcessConfig(selector='name') + self.query_one("#configs").mount(new_config) + + @on(Button.Pressed, "#next") + def save_and_load_next_screen(self) -> None: + try: + config_list = [] + for config_widget in self.query("ProcessConfig"): + tmp_config = {} + for text_input in config_widget.query("TextInput"): + this_input = text_input.query_one(Input) + validation_result = this_input.validate(this_input.value) + tmp_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("") + # Validate the config + with init_context(self.parent.get_context()): + ConfigsCreateConfig(**tmp_config) + # Add to the config list + config_list.append(tmp_config) + # Add to the final config + key = self.config_key + new_config = {self.config_key: {}} + for tmp_config in config_list: + process_id = tmp_config.get('custom_process_id') + new_config[self.config_key][process_id] = tmp_config + self.parent.TEMPLATE_CONFIG = self.parent.TEMPLATE_CONFIG.model_copy(update=new_config) + # Push the next screen + self.parent.push_screen(self.next_screen) + except ValueError: + pass + + @on(Button.Pressed, "#skip") + def skip_to_next_screen(self) -> None: + self.parent.push_screen(self.next_screen) + + +class MultiNamedProcessConfig(MultiProcessConfig): + def __init__(self) -> None: + super().__init__( + title='Configure processes by name', + selector_type='name', + config_key='named_process_resources' + ) + + @on(Mount) + @on(ScreenResume) + def set_next_screen(self) -> None: + next_screen = "final" + if self.parent.PIPE_CONF_LABELLED: + next_screen = "multi_labelled_process_config" + self._set_next_screen(next_screen) + + +class MultiLabelledProcessConfig(MultiProcessConfig): + def __init__(self) -> None: + super().__init__( + title='Configure processes by label', + selector_type='label', + config_key='labelled_process_resources' + ) + + @on(Mount) + @on(ScreenResume) + def set_next_screen(self) -> None: + self._set_next_screen('final') \ No newline at end of file diff --git a/nf_core/configs/create/pipelineconfigquestion.py b/nf_core/configs/create/pipelineconfigquestion.py new file mode 100644 index 0000000000..68e1dab0a2 --- /dev/null +++ b/nf_core/configs/create/pipelineconfigquestion.py @@ -0,0 +1,121 @@ +"""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, Switch, Label +from enum import Enum +from nf_core.utils import add_hide_class, remove_hide_class + +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 PipelineConfigQuestion(Screen): + """Determine whether the user wants to configure the default resources and/or specific process names/labels.""" + + def __init__(self) -> None: + super().__init__() + self.config_defaults = False + self.config_named_processes = False + self.config_labels = False + + def compose(self) -> ComposeResult: + yield Header() + yield Footer() + yield Markdown( + dedent( + """ + # What would you like to configure? + """ + ) + ) + with Horizontal(): + yield Label( + "Configure default process resources?", + id="toggle_configure_defaults_label" + ) + yield Switch( + id="toggle_configure_defaults", + value=self.config_defaults, + ) + yield Label( + "Yes" if self.config_defaults else "No", + id="toggle_configure_defaults_state_label" + ) + with Horizontal(): + yield Label( + "Configure specific named processes?", + id="toggle_configure_names_label" + ) + yield Switch( + id="toggle_configure_names", + value=self.config_defaults, + ) + yield Label( + "Yes" if self.config_defaults else "No", + id="toggle_configure_names_state_label" + ) + with Horizontal(): + yield Label( + "Configure labels?", + id="toggle_configure_labels_label" + ) + yield Switch( + id="toggle_configure_labels", + value=self.config_defaults, + ) + yield Label( + "Yes" if self.config_defaults else "No", + id="toggle_configure_labels_state_label" + ) + yield Center( + Button("Back", id="back", variant="default"), + Button("Next", id="next", variant="success"), + classes="cta", + ) + + @on(Switch.Changed) + def on_toggle_switch(self, event: Switch.Changed) -> None: + """ Handle toggling the switches that determine which pipeline resources to configure """ + valid_toggles = { + 'toggle_configure_defaults': 'config_defaults', + 'toggle_configure_names': 'config_named_processes', + 'toggle_configure_labels': 'config_labels', + } + + if event.switch.id not in valid_toggles: + return + + attr = valid_toggles[event.switch.id] + self.__setattr__(attr, event.value) + + # Update the switch label + for label in self.query(Label): + if label.id == f'{event.switch.id}_state_label': + label.update("Yes" if event.value else "No") + + # 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_next_button_pressed(self, event: Button.Pressed) -> None: + """Save configuration options and then move to the next screen.""" + # Update app tracking variables for whether to configure named and/or labelled processes + self.parent.PIPE_CONF_NAMED = self.config_named_processes + self.parent.PIPE_CONF_LABELLED = self.config_labels + + # Proceed to next screen depending on what choices the user has made + if self.config_defaults: + self.parent.push_screen("default_process_resources") + elif self.config_named_processes: + self.parent.push_screen("multi_named_process_config") + elif self.config_labels: + self.parent.push_screen("multi_labelled_process_config") + else: + self.parent.push_screen("final") diff --git a/nf_core/configs/create/utils.py b/nf_core/configs/create/utils.py index 3f7dd54e9d..4494944cd2 100644 --- a/nf_core/configs/create/utils.py +++ b/nf_core/configs/create/utils.py @@ -31,6 +31,7 @@ def init_context(value: dict[str, Any]) -> Iterator[None]: # Define a global variable to store the config type CONFIG_ISINFRASTRUCTURE_GLOBAL: bool = True NFCORE_CONFIG_GLOBAL: bool = True +INFRA_ISHPC_GLOBAL: bool = False class ConfigsCreateConfig(BaseModel): @@ -52,8 +53,52 @@ class ConfigsCreateConfig(BaseModel): """ Config description """ config_profile_url: Optional[str] = None """ Config institution URL """ + default_process_ncpus: Optional[str] = None + """ Default number of CPUs """ + default_process_memgb: Optional[str] = None + """ Default amount of memory """ + default_process_hours: Optional[str] = None + """ Default walltime - hours """ + custom_process_id: Optional[str] = None + """" Name or label of a process to configure """ + custom_process_ncpus: Optional[str] = None + """ Number of CPUs for process """ + custom_process_memgb: Optional[str] = None + """ Amount of memory for process """ + custom_process_hours: Optional[str] = None + """ Walltime for process - hours """ + named_process_resources: Optional[dict] = None + """ Dictionary containing custom resource requirements for named processes """ + labelled_process_resources: Optional[dict] = None + """ Dictionary containing custom resource requirements for labelled processes """ is_nfcore: Optional[bool] = None """ Whether the config is part of the nf-core organisation """ + savelocation: Optional[str] = None + """ Final location of the configuration file """ + scheduler: Optional[str] = None + """ The scheduler that the HPC uses """ + queue: Optional[str] = None + """ The default queue that the HPC uses """ + module_system: Optional[str] = None + """ Modules to load when running processes """ + container_system: Optional[str] = None + """ The container system the HPC uses """ + memory: Optional[str] = None + """ The maximum memory available to processes """ + cpus: Optional[str] = None + """ The maximum number of CPUs available to processes """ + time: Optional[str] = None + """ The maximum walltime available to processes """ + envvar: Optional[str] = None + """ An environment variable to hold a custom Nextflow container cachedir """ + cachedir: Optional[str] = None + """ An environment variable to hold a custom Nextflow container cachedir """ + igenomes_cachedir: Optional[str] = None + """ A cachedir for iGenomes """ + scratch_dir: Optional[str] = None + """ A scratch directory to use """ + retries: Optional[str] = None + """ Number of retries for failed jobs """ model_config = ConfigDict(extra="allow") @@ -78,19 +123,47 @@ def notempty(cls, v: str) -> str: def path_valid(cls, v: str, info: ValidationInfo) -> str: """Check that a path is valid.""" context = info.context - if context and not context["is_infrastructure"]: + if context and (not context["is_infrastructure"] and not context["is_nfcore"]): if v.strip() == "": raise ValueError("Cannot be left empty.") if not Path(v).is_dir(): raise ValueError("Must be a valid path.") return v - @field_validator("config_profile_contact", "config_profile_description", "config_pipeline_name") + @field_validator("savelocation") + @classmethod + def final_path_valid(cls, v: str, info: ValidationInfo) -> str: + """Check that the final save directory is valid.""" + if v.strip() == "": + raise ValueError("Cannot be left empty.") + if not Path(v).is_dir(): + raise ValueError("Must be a valid path to a directory.") + return v + + @field_validator("config_pipeline_name") + @classmethod + def nfcore_name_valid(cls, v: str, info: ValidationInfo) -> str: + """Check that an nf-core pipeline name is valid.""" + context = info.context + if context and (not context["is_infrastructure"] and context["is_nfcore"]): + if v.strip() == "": + raise ValueError("Cannot be left empty.") + return v + + @field_validator("config_profile_description") @classmethod - def notempty_nfcore(cls, v: str, info: ValidationInfo) -> str: - """Check that string values are not empty when the config is nf-core.""" + def notempty_description(cls, v: str) -> str: + """Check that description is not empty when.""" + if v.strip() == "": + raise ValueError("Cannot be left empty.") + return v + + @field_validator("config_profile_contact") + @classmethod + def notempty_contact(cls, v: str, info: ValidationInfo) -> str: + """Check that contact values are not empty when the config is infrastructure.""" context = info.context - if context and context["is_nfcore"]: + if context and context["is_infrastructure"]: if v.strip() == "": raise ValueError("Cannot be left empty.") return v @@ -103,7 +176,7 @@ def handle_prefix(cls, v: str, info: ValidationInfo) -> str: """Check that GitHub handles start with '@'. Make providing a handle mandatory for nf-core configs""" context = info.context - if context and context["is_nfcore"]: + if context and context["is_infrastructure"]: if v.strip() == "": raise ValueError("Cannot be left empty.") elif not re.match( @@ -122,7 +195,7 @@ def handle_prefix(cls, v: str, info: ValidationInfo) -> str: def url_prefix(cls, v: str, info: ValidationInfo) -> str: """Check that institutional web links start with valid URL prefix.""" context = info.context - if context and context["is_nfcore"]: + if context and context["is_infrastructure"]: if v.strip() == "": raise ValueError("Cannot be left empty.") elif not re.match( @@ -142,6 +215,47 @@ def url_prefix(cls, v: str, info: ValidationInfo) -> str: ) return v + @field_validator("custom_process_id") + @classmethod + def notempty_process_name(cls, v: str, info: ValidationInfo) -> str: + """Check that the custom process name or label isn't empty.""" + context = info.context + if context and not context["is_infrastructure"]: + if v.strip() == "": + raise ValueError("Cannot be left empty.") + return v + + @field_validator("default_process_ncpus", "default_process_memgb", "custom_process_ncpus", "custom_process_memgb") + @classmethod + def pos_integer_valid(cls, v: str, info: ValidationInfo) -> str: + """Check that integer values are either empty or positive.""" + context = info.context + if context and not context["is_infrastructure"]: + if v.strip() == "": + return v + try: + v_int = int(v.strip()) + except ValueError: + raise ValueError("Must be an integer.") + if not v_int > 0: + raise ValueError("Must be a positive integer.") + return v + + @field_validator("default_process_hours", "custom_process_hours") + @classmethod + def non_neg_float_valid(cls, v: str, info: ValidationInfo) -> str: + """Check that numeric values are either empty or non-negative.""" + context = info.context + if context and not context["is_infrastructure"]: + if v.strip() == "": + return v + try: + vf = float(v.strip()) + except ValueError: + raise ValueError("Must be a number.") + if not vf >= 0: + raise ValueError("Must be a non-negative number.") + return v ## TODO Duplicated from pipelines utils - move to common location if possible (validation seems to be context specific so possibly not) class TextInput(Static): @@ -210,7 +324,7 @@ def validate(self, value: str) -> ValidationResult: If it fails, return the error messages.""" try: - with init_context({"is_nfcore": NFCORE_CONFIG_GLOBAL, "is_infrastructure": CONFIG_ISINFRASTRUCTURE_GLOBAL}): + with init_context({"is_nfcore": NFCORE_CONFIG_GLOBAL, "is_infrastructure": CONFIG_ISINFRASTRUCTURE_GLOBAL, "is_hpc": INFRA_ISHPC_GLOBAL}): ConfigsCreateConfig(**{f"{self.key}": value}) return self.success() except ValidationError as e: