omnibus-2024-03-19 (ufs-community#431)

src/uwtools/api/chgres_cube.py

@@ -1,13 +1,13 @@
 """
-API access to the uwtools chgres_cube driver.
+API access to the uwtools ``chgres_cube`` driver.
 """
 import datetime as dt
 from pathlib import Path
 from typing import Dict, Optional
 
 import iotaa as _iotaa
 
-from uwtools.drivers import support
+import uwtools.drivers.support as _support
 from uwtools.drivers.chgres_cube import ChgresCube as _ChgresCube
 
 
@@ -20,7 +20,7 @@ def execute(
     graph_file: Optional[Path] = None,
 ) -> bool:
     """
-    Execute a chgres_cube task.
+    Execute a ``chgres_cube`` task.
 
     If ``batch`` is specified, a runscript will be written and submitted to the batch system.
     Otherwise, the executable will be run directly on the current system.
@@ -31,7 +31,7 @@ def execute(
     :param batch: Submit run to the batch system
     :param dry_run: Do not run the executable, just report what would have been done
     :param graph_file: Write Graphviz DOT output here
-    :return: True if task completes without raising an exception
+    :return: ``True`` if task completes without raising an exception
     """
     obj = _ChgresCube(config_file=config_file, cycle=cycle, batch=batch, dry_run=dry_run)
     getattr(obj, task)()
@@ -52,4 +52,4 @@ def tasks() -> Dict[str, str]:
     """
     Returns a mapping from task names to their one-line descriptions.
     """
-    return support.tasks(_ChgresCube)
+    return _support.tasks(_ChgresCube)

src/uwtools/api/config.py

@@ -1,5 +1,5 @@
 """
-API access to uwtools configuration management tools.
+API access to ``uwtools`` configuration management tools.
 """
 import os
 from pathlib import Path
@@ -123,7 +123,7 @@ def realize_to_dict(  # pylint: disable=unused-argument
     dry_run: bool = False,
 ) -> dict:
     """
-    Realize an output config to a dict, based on an input config and optional supplemental configs.
+    Realize a config to a ``dict``, based on an input config and optional supplemental configs.
 
     See ``realize()`` for details on arguments, etc.
     """
@@ -153,9 +153,9 @@ def _ensure_config_arg_type(
     config: Union[dict, _Config, Optional[Path]]
 ) -> Union[_Config, Optional[Path]]:
     """
-    Encapsulate a dict in a Config; return a Config or path argument as-is.
+    Encapsulate a ``dict`` in a ``Config``; return a ``Config`` or path argument as-is.
 
-    :param config: A config as a dict, Config, or path
+    :param config: A config as a ``dict``, ``Config``, or ``Path``
     """
     if isinstance(config, dict):
         return _YAMLConfig(config=config)
@@ -184,7 +184,7 @@ def _ensure_config_arg_type(
 
 
 realize.__doc__ = """
-Realize an output config based on an input config and optional supplemental configs.
+Realize a config based on an input config and optional supplemental configs.
 
 If no input is specified, ``stdin`` is read. A ``dict`` or ``Config`` object may also be provided as
 input.  If no output is specified, ``stdout`` is written to. When an input or output filename is

src/uwtools/api/file.py

@@ -1,5 +1,5 @@
 """
-API access to uwtools file-management tools.
+API access to ``uwtools`` file-management tools.
 """
 from pathlib import Path
 from typing import List, Optional, Union
@@ -18,10 +18,10 @@ def copy(
     Copy files.
 
     :param target_dir: Path to target directory
-    :param config: YAML-file path, or dict (read stdin if missing or None).
+    :param config: YAML-file path, or ``dict`` (read ``stdin`` if missing or ``None``).
     :param keys: YAML keys leading to file dst/src block
     :param dry_run: Do not copy files
-    :return: True if no exception is raised
+    :return: ``True`` if no exception is raised
     """
     _FileCopier(target_dir=target_dir, config=config, keys=keys, dry_run=dry_run).go()
     return True
@@ -37,10 +37,10 @@ def link(
     Link files.
 
     :param target_dir: Path to target directory
-    :param config: YAML-file path, or dict (read stdin if missing or None).
+    :param config: YAML-file path, or ``dict`` (read ``stdin`` if missing or ``None``).
     :param keys: YAML keys leading to file dst/src block
     :param dry_run: Do not link files
-    :return: True if no exception is raised
+    :return: ``True`` if no exception is raised
     """
     _FileLinker(target_dir=target_dir, config=config, keys=keys, dry_run=dry_run).go()
     return True

src/uwtools/api/fv3.py

@@ -1,11 +1,11 @@
 """
-API access to the uwtools FV3 driver.
+API access to the ``uwtools`` FV3 driver.
 """
 import datetime as dt
 from pathlib import Path
 from typing import Dict, Optional
 
-from uwtools.drivers import support
+import uwtools.drivers.support as _support
 from uwtools.drivers.fv3 import FV3 as _FV3
 
 
@@ -43,11 +43,11 @@ def graph() -> str:
     """
     Returns Graphviz DOT code for the most recently executed task.
     """
-    return support.graph()
+    return _support.graph()
 
 
 def tasks() -> Dict[str, str]:
     """
     Returns a mapping from task names to their one-line descriptions.
     """
-    return support.tasks(_FV3)
+    return _support.tasks(_FV3)

src/uwtools/api/logging.py

@@ -1,5 +1,5 @@
 """
-API access to uwtools logging logic.
+API access to ``uwtools`` logging logic.
 """
 import logging
 

src/uwtools/api/rocoto.py

@@ -1,15 +1,15 @@
 """
-API access to uwtools Rocoto support.
+API access to ``uwtools`` Rocoto support.
 """
 from pathlib import Path
 from typing import Optional, Union
 
-from uwtools.config.formats.yaml import YAMLConfig
+from uwtools.config.formats.yaml import YAMLConfig as _YAMLConfig
 from uwtools.rocoto import realize_rocoto_xml as _realize
 from uwtools.rocoto import validate_rocoto_xml_file as _validate
 
 
-def realize(config: Union[YAMLConfig, Optional[Path]], output_file: Optional[Path] = None) -> bool:
+def realize(config: Union[_YAMLConfig, Optional[Path]], output_file: Optional[Path] = None) -> bool:
     """
     Realize the Rocoto workflow defined in the given YAML as XML.
 

src/uwtools/api/sfc_climo_gen.py

@@ -1,10 +1,10 @@
 """
-API access to the uwtools sfc_climo_gen driver.
+API access to the ``uwtools`` ``sfc_climo_gen`` driver.
 """
 from pathlib import Path
 from typing import Dict, Optional
 
-from uwtools.drivers import support
+import uwtools.drivers.support as _support
 from uwtools.drivers.sfc_climo_gen import SfcClimoGen as _SfcClimoGen
 
 
@@ -16,7 +16,7 @@ def execute(
     graph_file: Optional[Path] = None,
 ) -> bool:
     """
-    Execute an sfc_climo_gen task.
+    Execute an ``sfc_climo_gen`` task.
 
     If ``batch`` is specified, a runscript will be written and submitted to the batch system.
     Otherwise, the forecast will be run directly on the current system.
@@ -40,11 +40,11 @@ def graph() -> str:
     """
     Returns Graphviz DOT code for the most recently executed task.
     """
-    return support.graph()
+    return _support.graph()
 
 
 def tasks() -> Dict[str, str]:
     """
     Returns a mapping from task names to their one-line descriptions.
     """
-    return support.tasks(_SfcClimoGen)
+    return _support.tasks(_SfcClimoGen)

src/uwtools/api/template.py

@@ -1,5 +1,5 @@
 """
-API access to uwtools templating logic.
+API access to ``uwtools`` templating logic.
 """
 import os
 from pathlib import Path

src/uwtools/drivers/chgres_cube.py

@@ -2,8 +2,6 @@
 A driver for chgres_cube.
 """
 
-import os
-import stat
 from datetime import datetime
 from pathlib import Path
 from typing import Any, Dict
@@ -21,8 +19,6 @@ class ChgresCube(Driver):
     A driver for chgres_cube.
     """
 
-    _driver_name = STR.chgrescube
-
     def __init__(
         self, config_file: Path, cycle: datetime, dry_run: bool = False, batch: bool = False
     ):
@@ -86,25 +82,22 @@ def runscript(self):
         yield self._taskname(path.name)
         yield asset(path, path.is_file)
         yield None
-
         envvars = {
             "KMP_AFFINITY": "scatter",
             "OMP_NUM_THREADS": self._driver_config.get("execution", {}).get("threads", 1),
             "OMP_STACKSIZE": "1024m",
         }
-        envcmds = self._driver_config.get("execution", {}).get("envcmds", [])
-        execution = [self._runcmd, "test $? -eq 0 && touch %s/done" % self._rundir]
-        scheduler = self._scheduler if self._batch else None
-        path.parent.mkdir(parents=True, exist_ok=True)
-        rs = self._runscript(
-            envcmds=envcmds, envvars=envvars, execution=execution, scheduler=scheduler
-        )
-        with open(path, "w", encoding="utf-8") as f:
-            print(rs, file=f)
-        os.chmod(path, os.stat(path).st_mode | stat.S_IEXEC)
+        self._write_runscript(path=path, envvars=envvars)
 
     # Private helper methods
 
+    @property
+    def _driver_name(self) -> str:
+        """
+        Returns the name of this driver.
+        """
+        return STR.chgrescube
+
     @property
     def _resources(self) -> Dict[str, Any]:
         """

src/uwtools/drivers/driver.py

@@ -1,7 +1,9 @@
 """
 An abstract class for component drivers.
 """
+import os
 import re
+import stat
 from abc import ABC, abstractmethod
 from pathlib import Path
 from textwrap import dedent
@@ -207,3 +209,18 @@ def _validate(self) -> None:
         """
         for schema_name in (self._driver_name.replace("_", "-"), "platform"):
             validate_internal(schema_name=schema_name, config=self._config)
+
+    def _write_runscript(self, path: Path, envvars: Dict[str, str]) -> None:
+        """
+        Write the runscript.
+        """
+        path.parent.mkdir(parents=True, exist_ok=True)
+        rs = self._runscript(
+            envcmds=self._driver_config.get("execution", {}).get("envcmds", []),
+            envvars=envvars,
+            execution=[self._runcmd, "test $? -eq 0 && touch %s/done" % self._rundir],
+            scheduler=self._scheduler if self._batch else None,
+        )
+        with open(path, "w", encoding="utf-8") as f:
+            print(rs, file=f)
+        os.chmod(path, os.stat(path).st_mode | stat.S_IEXEC)

src/uwtools/drivers/fv3.py

@@ -2,8 +2,6 @@
 A driver for the FV3 model.
 """
 
-import os
-import stat
 from datetime import datetime
 from pathlib import Path
 from shutil import copy
@@ -25,8 +23,6 @@ class FV3(Driver):
     A driver for the FV3 model.
     """
 
-    _driver_name = STR.fv3
-
     def __init__(
         self, config_file: Path, cycle: datetime, dry_run: bool = False, batch: bool = False
     ):
@@ -197,19 +193,17 @@ def runscript(self):
             "OMP_NUM_THREADS": self._driver_config.get("execution", {}).get("threads", 1),
             "OMP_STACKSIZE": "512m",
         }
-        envcmds = self._driver_config.get("execution", {}).get("envcmds", [])
-        execution = [self._runcmd, "test $? -eq 0 && touch %s/done" % self._rundir]
-        scheduler = self._scheduler if self._batch else None
-        path.parent.mkdir(parents=True, exist_ok=True)
-        rs = self._runscript(
-            envcmds=envcmds, envvars=envvars, execution=execution, scheduler=scheduler
-        )
-        with open(path, "w", encoding="utf-8") as f:
-            print(rs, file=f)
-        os.chmod(path, os.stat(path).st_mode | stat.S_IEXEC)
+        self._write_runscript(path=path, envvars=envvars)
 
     # Private helper methods
 
+    @property
+    def _driver_name(self) -> str:
+        """
+        Returns the name of this driver.
+        """
+        return STR.fv3
+
     def _taskname(self, suffix: str) -> str:
         """
         Returns a common tag for graph-task log messages.

src/uwtools/drivers/sfc_climo_gen.py

@@ -2,8 +2,6 @@
 A driver for sfc_climo_gen.
 """
 
-import os
-import stat
 from pathlib import Path
 from typing import Any, Dict
 
@@ -20,8 +18,6 @@ class SfcClimoGen(Driver):
     A driver for sfc_climo_gen.
     """
 
-    _driver_name = STR.sfcclimogen
-
     def __init__(self, config_file: Path, dry_run: bool = False, batch: bool = False):
         """
         The driver.
@@ -76,17 +72,17 @@ def runscript(self):
         yield self._taskname(path.name)
         yield asset(path, path.is_file)
         yield None
-        envcmds = self._driver_config.get("execution", {}).get("envcmds", [])
-        execution = [self._runcmd, "test $? -eq 0 && touch %s/done" % self._rundir]
-        scheduler = self._scheduler if self._batch else None
-        path.parent.mkdir(parents=True, exist_ok=True)
-        rs = self._runscript(envcmds=envcmds, execution=execution, scheduler=scheduler)
-        with open(path, "w", encoding="utf-8") as f:
-            print(rs, file=f)
-        os.chmod(path, os.stat(path).st_mode | stat.S_IEXEC)
+        self._write_runscript(path=path, envvars={})
 
     # Private helper methods
 
+    @property
+    def _driver_name(self) -> str:
+        """
+        Returns the name of this driver.
+        """
+        return STR.sfcclimogen
+
     @property
     def _resources(self) -> Dict[str, Any]:
         """

src/uwtools/tests/api/test_chgres_cube.py

@@ -31,6 +31,6 @@ def test_graph():
 
 
 def test_tasks():
-    with patch.object(chgres_cube.support, "tasks") as _tasks:
+    with patch.object(chgres_cube._support, "tasks") as _tasks:
         chgres_cube.tasks()
     _tasks.assert_called_once_with(chgres_cube._ChgresCube)