Expose iotaa Graphviz output

recipe/meta.json

@@ -8,7 +8,7 @@
       "coverage =7.3.*",
       "docformatter =1.7.*",
       "f90nml =1.4.*",
-      "iotaa =0.7.2.*",
+      "iotaa =0.7.3.*",
       "isort =5.13.*",
       "jinja2 =3.1.*",
       "jq =1.7.*",
@@ -24,7 +24,7 @@
     ],
     "run": [
       "f90nml =1.4.*",
-      "iotaa =0.7.2.*",
+      "iotaa =0.7.3.*",
       "jinja2 =3.1.*",
       "jsonschema =4.20.*",
       "lxml =4.9.*",

recipe/meta.yaml

@@ -12,7 +12,7 @@ requirements:
     - pip
   run:
     - f90nml 1.4.*
-    - iotaa 0.7.2.*
+    - iotaa 0.7.3.*
     - jinja2 3.1.*
     - jsonschema 4.20.*
     - lxml 4.9.*

src/uwtools/api/fv3.py

@@ -3,9 +3,9 @@
 """
 import datetime as dt
 from pathlib import Path
-from typing import Dict
+from typing import Dict, Optional
 
-import iotaa
+import iotaa as _iotaa
 
 from uwtools.drivers.fv3 import FV3
 
@@ -16,6 +16,7 @@ def execute(
     cycle: dt.datetime,
     batch: bool = False,
     dry_run: bool = False,
+    graph_file: Optional[Path] = None,
 ) -> bool:
     """
     Execute an FV3 task.
@@ -28,17 +29,28 @@ def execute(
     :param cycle: The cycle to run
     :param batch: Submit run to the batch system
     :param dry_run: Do not run forecast, just report what would have been done
+    :param graph_file: Write Graphviz DOT output here
     :return: True if task completes without raising an exception
     """
     obj = FV3(config_file=config_file, cycle=cycle, batch=batch, dry_run=dry_run)
     getattr(obj, task)()
+    if graph_file:
+        with open(graph_file, "w", encoding="utf-8") as f:
+            print(graph(), file=f)
     return True
 
 
+def graph() -> str:
+    """
+    Returns Graphviz DOT code for the most recently executed task.
+    """
+    return _iotaa.graph()
+
+
 def tasks() -> Dict[str, str]:
     """
     Returns a mapping from task names to their one-line descriptions.
     """
     return {
-        task: getattr(FV3, task).__doc__.strip().split("\n")[0] for task in iotaa.tasknames(FV3)
+        task: getattr(FV3, task).__doc__.strip().split("\n")[0] for task in _iotaa.tasknames(FV3)
     }

src/uwtools/api/sfc_climo_gen.py

@@ -2,9 +2,9 @@
 API access to the uwtools sfc_climo_gen driver.
 """
 from pathlib import Path
-from typing import Dict
+from typing import Dict, Optional
 
-import iotaa
+import iotaa as _iotaa
 
 from uwtools.drivers.sfc_climo_gen import SfcClimoGen
 
@@ -14,6 +14,7 @@ def execute(
     config_file: Path,
     batch: bool = False,
     dry_run: bool = False,
+    graph_file: Optional[Path] = None,
 ) -> bool:
     """
     Execute an sfc_climo_gen task.
@@ -25,18 +26,29 @@ def execute(
     :param config_file: Path to YAML config file
     :param batch: Submit run to the batch system
     :param dry_run: Do not run forecast, just report what would have been done
+    :param graph_file: Write Graphviz DOT output here
     :return: True if task completes without raising an exception
     """
     obj = SfcClimoGen(config_file=config_file, batch=batch, dry_run=dry_run)
     getattr(obj, task)()
+    if graph_file:
+        with open(graph_file, "w", encoding="utf-8") as f:
+            print(graph(), file=f)
     return True
 
 
+def graph() -> str:
+    """
+    Returns Graphviz DOT code for the most recently executed task.
+    """
+    return _iotaa.graph()
+
+
 def tasks() -> Dict[str, str]:
     """
     Returns a mapping from task names to their one-line descriptions.
     """
     return {
         task: getattr(SfcClimoGen, task).__doc__.strip().split("\n")[0]
-        for task in iotaa.tasknames(SfcClimoGen)
+        for task in _iotaa.tasknames(SfcClimoGen)
     }

src/uwtools/cli.py

@@ -233,6 +233,7 @@ def _add_subparser_fv3_task(subparsers: Subparsers, task: str, helpmsg: str) ->
     optional = _basic_setup(parser)
     _add_arg_batch(optional)
     _add_arg_dry_run(optional)
+    _add_arg_graph_file(optional)
     checks = _add_args_verbosity(optional)
     return checks
 
@@ -249,6 +250,7 @@ def _dispatch_fv3(args: Args) -> bool:
         cycle=args[STR.cycle],
         batch=args[STR.batch],
         dry_run=args[STR.dryrun],
+        graph_file=args[STR.graphfile],
     )
 
 
@@ -363,6 +365,7 @@ def _add_subparser_sfc_climo_gen_task(
     optional = _basic_setup(parser)
     _add_arg_batch(optional)
     _add_arg_dry_run(optional)
+    _add_arg_graph_file(optional)
     checks = _add_args_verbosity(optional)
     return checks
 
@@ -378,6 +381,7 @@ def _dispatch_sfc_climo_gen(args: Args) -> bool:
         config_file=args[STR.cfgfile],
         batch=args[STR.batch],
         dry_run=args[STR.dryrun],
+        graph_file=args[STR.graphfile],
     )
 
 
@@ -549,6 +553,15 @@ def _add_arg_file_path(group: Group, switch: str, helpmsg: str, required: bool =
     )
 
 
+def _add_arg_graph_file(group: Group) -> None:
+    group.add_argument(
+        _switch(STR.graphfile),
+        help="Path to Graphviz DOT output [experimental]",
+        metavar="PATH",
+        type=str,
+    )
+
+
 def _add_arg_input_file(group: Group, required: bool = False) -> None:
     group.add_argument(
         _switch(STR.infile),
@@ -832,6 +845,7 @@ class STR:
     file2fmt: str = "file_2_format"
     file2path: str = "file_2_path"
     fv3: str = "fv3"
+    graphfile: str = "graph_file"
     help: str = "help"
     infile: str = "input_file"
     infmt: str = "input_format"

src/uwtools/tests/api/test_fv3.py

@@ -3,24 +3,37 @@
 import datetime as dt
 from unittest.mock import patch
 
-from iotaa import external, task, tasks
+from iotaa import asset, external, task, tasks
 
 from uwtools.api import fv3
 
 
-def test_execute():
+def test_execute(tmp_path):
+    dot = tmp_path / "graph.dot"
     args: dict = {
         "config_file": "config.yaml",
         "cycle": dt.datetime.utcnow(),
         "batch": False,
         "dry_run": True,
+        "graph_file": dot,
     }
     with patch.object(fv3, "FV3") as FV3:
         assert fv3.execute(**args, task="foo") is True
+    del args["graph_file"]
     FV3.assert_called_once_with(**args)
     FV3().foo.assert_called_once_with()
 
 
+def test_graph():
+    @external
+    def ready():
+        yield "ready"
+        yield asset("ready", lambda: True)
+
+    ready()
+    assert fv3.graph().startswith("digraph")
+
+
 def test_tasks():
     @external
     def t1():

src/uwtools/tests/api/test_sfc_climo_gen.py

@@ -2,23 +2,36 @@
 
 from unittest.mock import patch
 
-from iotaa import external, task, tasks
+from iotaa import asset, external, task, tasks
 
 from uwtools.api import sfc_climo_gen
 
 
-def test_execute():
+def test_execute(tmp_path):
+    dot = tmp_path / "graph.dot"
     args: dict = {
         "config_file": "config.yaml",
         "batch": False,
         "dry_run": True,
+        "graph_file": dot,
     }
     with patch.object(sfc_climo_gen, "SfcClimoGen") as SfcClimoGen:
         assert sfc_climo_gen.execute(**args, task="foo") is True
+    del args["graph_file"]
     SfcClimoGen.assert_called_once_with(**args)
     SfcClimoGen().foo.assert_called_once_with()
 
 
+def test_graph():
+    @external
+    def ready():
+        yield "ready"
+        yield asset("ready", lambda: True)
+
+    ready()
+    assert sfc_climo_gen.graph().startswith("digraph")
+
+
 def test_tasks():
     @external
     def t1():

src/uwtools/tests/test_cli.py

@@ -296,6 +296,7 @@ def test__dispatch_fv3():
         "config_file": "config.yaml",
         "cycle": dt.datetime.now(),
         "dry_run": False,
+        "graph_file": None,
     }
     with patch.object(uwtools.api.fv3, "execute") as execute:
         cli._dispatch_fv3({**args, "action": "foo"})
@@ -356,6 +357,7 @@ def test__dispatch_sfc_climo_gen():
         "batch": True,
         "config_file": "config.yaml",
         "dry_run": False,
+        "graph_file": None,
     }
     with patch.object(uwtools.api.sfc_climo_gen, "execute") as execute:
         cli._dispatch_sfc_climo_gen({**args, "action": "foo"})