swell.py is refactored to commands/ create.py task.py etc.

src/swell/commands/clone.py

@@ -0,0 +1,36 @@
+import click
+from swell.commands.help_strings import input_method_help, platform_help, advanced_help
+
+
+@click.command()
+@click.argument('configuration')
+@click.argument('experiment_id')
+@click.option('-m', '--input_method', 'input_method', default='defaults',
+              type=click.Choice(['defaults', 'cli']), help=input_method_help)
+@click.option('-p', '--platform', 'platform', default=None, help=platform_help())
+@click.option('-a', '--advanced', 'advanced', default=False, help=advanced_help)
+def clone(
+    configuration: str,
+    experiment_id: str,
+    input_method: str,
+    platform: str,
+    advanced: bool
+) -> None:
+    """
+    Clone an existing experiment
+
+    This command creates an experiment directory based on the provided experiment configuration.
+
+    Arguments: \n
+        configuration (str): Path to a YAML containing the experiment configuration you wish to
+        clone from. \n
+
+    """
+
+    from swell.deployment.create_experiment import clone_config, create_experiment_directory
+    # Create experiment configuration by cloning from existing experiment
+    experiment_dict_str = clone_config(configuration, experiment_id, input_method, platform,
+                                       advanced)
+
+    # Create the experiment directory
+    create_experiment_directory(experiment_dict_str)

src/swell/commands/create.py

@@ -0,0 +1,54 @@
+import os
+import click
+from typing import Union
+from swell.swell_path import get_swell_path
+from swell.deployment.platforms.platforms import get_platforms
+from swell.commands.help_strings import (input_method_help, platform_help,
+                                         override_help, advanced_help, slurm_help, skip_r2d2_help)
+
+
+class LazySuiteChoice(click.ParamType):
+    name = "suite"
+
+    def convert(self, value, param, ctx):
+        suites_dir = os.path.join(get_swell_path(), 'suites')
+        suite_names = [
+            name for name in os.listdir(suites_dir)
+            if os.path.isdir(os.path.join(suites_dir, name))
+            and not name.startswith('__')
+        ]
+        if value not in suite_names:
+            self.fail(f"{value} is not a valid suite", param, ctx)
+        return value
+
+
+@click.command()
+@click.argument('suite', type=LazySuiteChoice())
+@click.option('-m', '--input_method', 'input_method', default='defaults',
+              type=click.Choice(['defaults', 'cli']), help=input_method_help)
+@click.option('-p', '--platform', 'platform', default='nccs_discover_sles15',
+              type=click.Choice(get_platforms()), help=platform_help())
+@click.option('-o', '--override', 'override', default=None, help=override_help)
+@click.option('-a', '--advanced', 'advanced', default=False, help=advanced_help)
+@click.option('-s', '--slurm', 'slurm', default=None, help=slurm_help)
+@click.option('-k', '--skip-r2d2', 'skip_r2d2', is_flag=True, default=False, help=skip_r2d2_help)
+def create(
+    suite: str,
+    input_method: str,
+    platform: str,
+    override: Union[dict, str, None],
+    advanced: bool,
+    slurm: str,
+    skip_r2d2: bool
+) -> None:
+    """
+    Create a new experiment
+
+    This command creates an experiment directory based on the provided suite name and options.
+
+    Arguments: \n
+        suite (str): Name of the suite you wish to run. \n
+
+    """
+    from swell.deployment.create_experiment import create_experiment_directory
+    create_experiment_directory(suite, input_method, platform, override, advanced, slurm, skip_r2d2)

src/swell/commands/help_strings.py

@@ -0,0 +1,53 @@
+# Help strings for optional arguments
+
+input_method_help = 'Method by which to create the YAML configuration file. If choosing ' + \
+                    'defaults the setting for the default suite test will be used. If using ' + \
+                    'CLI you will be led through the questions to configure the experiment.'
+
+
+def platform_help():
+    from swell.deployment.platforms.platforms import get_platforms
+    return (
+        "If using defaults for input_method, this option is used to determine which "
+        "platform to use for platform specific defaults. Options are "
+        + str(get_platforms())
+    )
+
+
+override_help = 'After generating the config file, parameters inside can be overridden ' + \
+                'using values from the override config file.'
+
+
+advanced_help = 'Show configuration questions which are otherwise not shown to the user.'
+
+
+no_detach_help = 'Tells the workflow manager not to detach. That is to say run the entire ' + \
+                 'run the entire workflow in the foreground and pass back a return code.'
+
+
+def log_path_help():
+    return (
+        'Directory to receive workflow manager logging output (instead of '
+        '$HOME/cylc-run/<suite_name>'
+        )
+
+
+datetime_help = 'Datetime to use for task execution. Format is yyyy-mm-ddThh:mm:ss. Note that ' + \
+                'non-numeric characters will be stripped from the string. Minutes and seconds ' + \
+                'are optional.'
+
+
+model_help = 'Data assimilation system. I.e. the model being initialized by data assimilation.'
+
+
+ensemble_help = 'When handling ensemble workflows using a parallel strategy, ' + \
+                'specify which packet of ensemble members to consider.'
+
+
+slurm_help = """
+Customize SLURM directives, globally (e.g., account name), for specific tasks,
+or for task-model combinations.
+"""
+
+
+skip_r2d2_help = """Skip registering this experiment and storing products in R2D2."""

src/swell/commands/launch.py

@@ -0,0 +1,24 @@
+import click
+from swell.commands.help_strings import no_detach_help, log_path_help
+
+
+@click.command()
+@click.argument('suite_path')
+@click.option('-b', '--no-detach', 'no_detach', is_flag=True, default=False, help=no_detach_help)
+@click.option('-l', '--log_path', 'log_path', default=None, help=log_path_help())
+def launch(
+    suite_path: str,
+    no_detach: bool,
+    log_path: str
+) -> None:
+    """
+    Launch an experiment with the cylc workflow manager
+
+    This command launches an experiment using the provided suite path and options.
+
+    Arguments: \n
+        suite_path (str): Path to where the flow.cylc and associated suite files are located. \n
+
+    """
+    from swell.deployment.launch_experiment import launch_experiment
+    launch_experiment(suite_path, no_detach, log_path)

src/swell/commands/t1test.py

@@ -0,0 +1,23 @@
+import click
+from swell.deployment.platforms.platforms import get_platforms
+from swell.commands.help_strings import platform_help
+from typing import Optional, Literal
+
+
+@click.command()
+@click.option('-p', '--platform', 'platform', type=click.Choice(get_platforms()),
+              default="nccs_discover_sles15", help=platform_help())
+@click.argument('suite', type=click.Choice(("hofx", "3dvar_marine", "3dvar_atmos",
+                                            "localensembleda", "3dvar_cycle")))
+def t1test(
+        suite: Literal["hofx", "3dvar_marine", "3dvar_atmos", "localensembleda", "3dvar_cycle"],
+        platform: Optional[str] = "nccs_discover_sles15"
+) -> None:
+    """
+    Run a particular swell suite from the tier 1 tests.
+
+    Arguments:
+        suite (str): Name of the suite to run (e.g., 3dvar_marine, 3dvar_atmos, localensembleda)
+    """
+    from swell.test.suite_tests.suite_tests import run_suite, TestSuite
+    run_suite(suite, platform, TestSuite.TIER1)

src/swell/commands/t2test.py

@@ -0,0 +1,24 @@
+import click
+from swell.deployment.platforms.platforms import get_platforms
+from swell.commands.help_strings import platform_help
+from typing import Optional, Literal
+
+
+@click.command()
+@click.option('-p', '--platform', 'platform', type=click.Choice(get_platforms()),
+              default="nccs_discover_sles15", help=platform_help())
+@click.argument('suite', type=click.Choice(("hofx", "3dvar_marine", "ufo_testing",
+                                            "convert_ncdiags", "3dfgat_atmos", "build_jedi")))
+def t2test(
+    suite: Literal["hofx", "3dvar_marine", "ufo_testing",
+                   "convert_ncdiags", "3dfgat_atmos", "build_jedi"],
+        platform: Optional[str] = "nccs_discover_sles15"
+) -> None:
+    """
+    Run a particular swell suite from the tier 2 tests.
+
+    Arguments:
+        suite (str): Name of the suite to run (e.g., hofx, 3dvar_marine, ufo_testing)
+    """
+    from swell.test.suite_tests.suite_tests import run_suite, TestSuite
+    run_suite(suite, platform, TestSuite.TIER2)

src/swell/commands/task.py

@@ -0,0 +1,62 @@
+import os
+import glob
+import click
+from typing import Optional
+from swell.swell_path import get_swell_path
+from swell.commands.help_strings import datetime_help, model_help, ensemble_help
+
+
+def snake_case_to_camel_case(name: str) -> str:
+    return ''.join(word.capitalize() for word in name.split('_'))
+
+
+class TaskChoice(click.ParamType):
+    """Click parameter type for dynamically listing tasks."""
+
+    name = "task"
+
+    def convert(self, value, param, ctx):
+        tasks = self.get_tasks()  # lazy evaluation
+        if value not in tasks:
+            self.fail(f"{value} is not a valid task", param, ctx)
+        return value
+
+    @staticmethod
+    def get_tasks() -> list[str]:
+        """Return list of task names in CamelCase."""
+        tasks_dir = os.path.join(get_swell_path(), "tasks", "*.py")
+        return sorted(
+            snake_case_to_camel_case(os.path.splitext(os.path.basename(f))[0])
+            for f in glob.glob(tasks_dir)
+            if "__" not in os.path.basename(f)
+        )
+
+
+TASK = TaskChoice()
+
+
+@click.command()
+@click.argument('task', type=TASK)
+@click.argument('config')
+@click.option('-d', '--datetime', 'datetime', default=None, help=datetime_help)
+@click.option('-m', '--model', 'model', default=None, help=model_help)
+@click.option('-p', '--ensemblePacket', 'ensemblePacket', default=None, help=ensemble_help)
+def task(
+        task: str,
+        config: str,
+        datetime: Optional[str],
+        model: Optional[str],
+        ensemblePacket: Optional[str]
+) -> None:
+    """
+    Run a workflow task
+
+    This command executes a task using the provided task name, configuration file and options.
+
+    Arguments:\n
+        task (str): Name of the task to execute.\n
+        config (str): Path to the configuration file for the task.\n
+
+    """
+    from swell.tasks.base.task_base import task_wrapper
+    task_wrapper(task, config, datetime, model, ensemblePacket)

src/swell/commands/test.py

@@ -0,0 +1,18 @@
+import click
+from swell.test.test_driver import valid_tests
+
+
+@click.command()
+@click.argument('test', type=click.Choice(valid_tests))
+def test(test: str) -> None:
+    """
+    Run one of the test suites
+
+    This command performs the test specified by the test argument.
+
+    Arguments:\n
+        test (str): Name of the test to execute.
+
+    """
+    from swell.test.test_driver import test_wrapper
+    test_wrapper(test)

src/swell/commands/utility.py

@@ -0,0 +1,18 @@
+import click
+from swell.utilities.scripts.utility_driver import get_utilities
+
+
+@click.command()
+@click.argument('utility', type=click.Choice(get_utilities()))
+def utility(utility: str) -> None:
+    """
+    Run a utility script
+
+    This command performs a utility operation specified by the utility argument.
+
+    Arguments:\n
+        utility (str): Name of the utility operation to perform.\n
+
+    """
+    from swell.utilities.scripts.utility_driver import utility_wrapper
+    utility_wrapper(utility)