Update verify_* functions to take a *RootParamGroup object (#7)

* Update verify_* functions to take a RootParamGroup object instead of a runconfig dict

* Apply suggestions from code review

Co-authored-by: Geoffrey Gunter <[email protected]>

* update log message for log file

* update typing from str to path-like

* add _typing.py module with RunConfigDict type

* RootParamGroup is ABC. Update type annotations

* bugfix in _typing.py

* infer product_type in insar and offsets verify_ functions

* bugfix 2 in _typing.py

* move build_root_params() into class function of RootParamsGroup

* move build_root_params_from_runconfig() to be a classmethod in RootParamGroup

* docstring cleanup

* Docstring cleanup from code review

Co-authored-by: Geoffrey Gunter <[email protected]>

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* bugfix: build_root_params function names

* harmonize order of arguments for build_root_params

* making typing.py module public

* add RootParamGroupT type

* Apply suggestions from code review

Co-authored-by: Geoffrey Gunter <[email protected]>

---------

Co-authored-by: Samantha C. Niemoeller <[email protected]>
Co-authored-by: Geoffrey Gunter <[email protected]>
Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>

src/nisarqa/__init__.py

@@ -147,5 +147,6 @@ def _my_private_foo():
 from .utils.summary_csv import *
 from .utils.metrics_writer import *
 from .utils.sanity_checks import *
+from .utils.typing import *
 
 # isort: on

src/nisarqa/__main__.py

@@ -1,14 +1,16 @@
 #!/usr/bin/env python3
+from __future__ import annotations
+
+import os
+
+import matplotlib
+
 # Switch backend to one that doesn't require DISPLAY to be set since we're
 # just plotting to file anyway. (Some compute notes do not allow X connections)
 # This needs to be set prior to opening any matplotlib objects.
-import matplotlib
-
 matplotlib.use("Agg")
 import argparse
 
-from ruamel.yaml import YAML
-
 import nisarqa
 
 
@@ -152,27 +154,6 @@ def dumpconfig(product_type, indent=4):
         )
 
 
-def load_user_runconfig(runconfig_yaml):
-    """
-    Load a QA Runconfig yaml file into a dict format.
-
-    Parameters
-    ----------
-    runconfig_yaml : str
-        Filename (with path) to a QA runconfig yaml file.
-
-    Returns
-    -------
-    user_rncfg : dict
-        `runconfig_yaml` loaded into a dict format
-    """
-    # parse runconfig into a dict structure
-    parser = YAML(typ="safe")
-    with open(runconfig_yaml, "r") as f:
-        user_rncfg = parser.load(f)
-    return user_rncfg
-
-
 def run():
     # parse the args
     args = parse_cli_args()
@@ -184,43 +165,39 @@ def run():
         dumpconfig(product_type=args.product_type, indent=args.indent)
         return
 
-    # parse runconfig into a dict structure
     log = nisarqa.get_logger()
+
+    # Generate the *RootParamGroup object from the runconfig
+    product_type = subcommand.replace("_qa", "")
+    try:
+        root_params = nisarqa.RootParamGroup.from_runconfig_file(
+            args.runconfig_yaml, product_type
+        )
+    except nisarqa.ExitEarly:
+        # No workflows were requested. Exit early.
+        log.info(
+            "All `workflows` set to `False` in the runconfig, "
+            "so no QA outputs will be generated. This is not an error."
+        )
+        return
+
     log.info(
-        f"Begin loading user runconfig yaml to dict: {args.runconfig_yaml}"
-    )
-    user_rncfg = load_user_runconfig(args.runconfig_yaml)
-    log.info(
-        "Loading of user runconfig complete. Beginning QA for"
-        f" {subcommand.replace('_qa', '').upper()} input product."
+        "Parsing of runconfig complete. Beginning QA for"
+        f" {product_type.upper()} input product."
     )
 
+    # Run QA SAS
+    verbose = args.verbose
     if subcommand == "rslc_qa":
-        nisarqa.rslc.verify_rslc(user_rncfg=user_rncfg, verbose=args.verbose)
+        nisarqa.rslc.verify_rslc(root_params=root_params, verbose=verbose)
     elif subcommand == "gslc_qa":
-        nisarqa.gslc.verify_gslc(user_rncfg=user_rncfg, verbose=args.verbose)
+        nisarqa.gslc.verify_gslc(root_params=root_params, verbose=verbose)
     elif subcommand == "gcov_qa":
-        nisarqa.gcov.verify_gcov(user_rncfg=user_rncfg, verbose=args.verbose)
-    elif subcommand == "rifg_qa":
-        nisarqa.igram.verify_igram(
-            user_rncfg=user_rncfg, product_type="rifg", verbose=args.verbose
-        )
-    elif subcommand == "runw_qa":
-        nisarqa.igram.verify_igram(
-            user_rncfg=user_rncfg, product_type="runw", verbose=args.verbose
-        )
-    elif subcommand == "gunw_qa":
-        nisarqa.igram.verify_igram(
-            user_rncfg=user_rncfg, product_type="gunw", verbose=args.verbose
-        )
-    elif subcommand == "roff_qa":
-        nisarqa.offsets.verify_offset(
-            user_rncfg=user_rncfg, product_type="roff", verbose=args.verbose
-        )
-    elif subcommand == "goff_qa":
-        nisarqa.offsets.verify_offset(
-            user_rncfg=user_rncfg, product_type="goff", verbose=args.verbose
-        )
+        nisarqa.gcov.verify_gcov(root_params=root_params, verbose=verbose)
+    elif subcommand in ("rifg_qa", "runw_qa", "gunw_qa"):
+        nisarqa.igram.verify_igram(root_params=root_params, verbose=verbose)
+    elif subcommand in ("roff_qa", "goff_qa"):
+        nisarqa.offsets.verify_offset(root_params=root_params, verbose=verbose)
     else:
         raise ValueError(f"Unknown subcommand: {subcommand}")
 

src/nisarqa/parameters/nisar_params.py

@@ -9,7 +9,7 @@
 from collections.abc import Iterable, Sequence
 from dataclasses import dataclass, field, fields
 from pathlib import Path
-from typing import Any, ClassVar, Optional, Union
+from typing import Any, ClassVar, Optional, Type, Union
 
 import h5py
 from ruamel.yaml import YAML, CommentedMap, CommentedSeq
@@ -1530,5 +1530,245 @@ def get_log_filename(self) -> Path:
 
         return Path("LOG.txt")
 
+    @classmethod
+    def from_runconfig_dict(
+        cls: type[nisarqa.RootParamGroupT],
+        user_rncfg: nisarqa.RunConfigDict,
+        product_type: str,
+    ) -> nisarqa.RootParamGroupT:
+        """
+        Build a *RootParamGroup for `product_type` from a QA runconfig dict.
+
+        Parameters
+        ----------
+        user_rncfg : nisarqa.RunConfigDict
+            A dictionary whose structure matches `product_type`'s QA runconfig
+            YAML file and that contains the parameters needed to run its QA SAS.
+        product_type : str
+            One of: 'rslc', 'gslc', 'gcov', 'rifg', 'runw', 'gunw', 'roff',
+            or 'goff'.
+
+        Returns
+        -------
+        root_params : nisarqa.RootParamGroup
+            *RootParamGroup object for the specified product type. This will be
+            populated with runconfig values where provided,
+            and default values for missing runconfig parameters.
+
+        Raises
+        ------
+        nisarqa.ExitEarly
+            If all `workflows` were set to False in the runconfig.
+
+        See Also
+        --------
+        RootParamGroup.from_runconfig_file
+        """
+        if product_type not in nisarqa.LIST_OF_NISAR_PRODUCTS:
+            raise ValueError(
+                f"{product_type=}, must be one of:"
+                f" {nisarqa.LIST_OF_NISAR_PRODUCTS}"
+            )
+
+        if product_type == "rslc":
+            workflows_param_cls_obj = nisarqa.RSLCWorkflowsParamGroup
+            root_param_class_obj = nisarqa.RSLCRootParamGroup
+        elif product_type == "gslc":
+            workflows_param_cls_obj = nisarqa.SLCWorkflowsParamGroup
+            root_param_class_obj = nisarqa.GSLCRootParamGroup
+        elif product_type == "gcov":
+            workflows_param_cls_obj = WorkflowsParamGroup
+            root_param_class_obj = nisarqa.GCOVRootParamGroup
+        elif product_type == "rifg":
+            workflows_param_cls_obj = nisarqa.RIFGWorkflowsParamGroup
+            root_param_class_obj = nisarqa.RIFGRootParamGroup
+        elif product_type == "runw":
+            workflows_param_cls_obj = nisarqa.RUNWWorkflowsParamGroup
+            root_param_class_obj = nisarqa.RUNWRootParamGroup
+        elif product_type == "gunw":
+            workflows_param_cls_obj = nisarqa.GUNWWorkflowsParamGroup
+            root_param_class_obj = nisarqa.GUNWRootParamGroup
+        elif product_type == "roff":
+            workflows_param_cls_obj = nisarqa.ROFFWorkflowsParamGroup
+            root_param_class_obj = nisarqa.ROFFRootParamGroup
+        elif product_type == "goff":
+            workflows_param_cls_obj = nisarqa.GOFFWorkflowsParamGroup
+            root_param_class_obj = nisarqa.GOFFRootParamGroup
+        else:
+            raise NotImplementedError(f"{product_type} code not implemented.")
+
+        # Dictionary to hold the *ParamGroup objects. Will be used as
+        # kwargs for the *RootParamGroup instance.
+        root_inputs = {}
+
+        # Construct *WorkflowsParamGroup dataclass (necessary for all workflows)
+        try:
+            root_inputs["workflows"] = (
+                cls._get_param_group_instance_from_runcfg(
+                    param_grp_cls_obj=workflows_param_cls_obj,
+                    user_rncfg=user_rncfg,
+                )
+            )
+
+        except KeyError as e:
+            raise KeyError(
+                "`workflows` group is a required runconfig group"
+            ) from e
+        # If all functionality is off (i.e. all workflows are set to false),
+        # then exit early. We will not need any of the other runconfig groups.
+        if not root_inputs["workflows"].at_least_one_wkflw_requested():
+            raise nisarqa.ExitEarly("All `workflows` were set to False.")
+
+        workflows = root_inputs["workflows"]
+
+        wkflws2params_mapping = (
+            root_param_class_obj.get_mapping_of_workflows2param_grps(
+                workflows=workflows
+            )
+        )
+
+        for param_grp in wkflws2params_mapping:
+            if param_grp.flag_param_grp_req:
+                populated_rncfg_group = (
+                    cls._get_param_group_instance_from_runcfg(
+                        param_grp_cls_obj=param_grp.param_grp_cls_obj,
+                        user_rncfg=user_rncfg,
+                    )
+                )
+
+                root_inputs[param_grp.root_param_grp_attr_name] = (
+                    populated_rncfg_group
+                )
+
+        # Construct and return *RootParamGroup
+        return root_param_class_obj(**root_inputs)
+
+    @staticmethod
+    def _get_param_group_instance_from_runcfg(
+        param_grp_cls_obj: Type[YamlParamGroup],
+        user_rncfg: Optional[dict] = None,
+    ):
+        """
+        Generate an instance of a YamlParamGroup subclass) object
+        where the values from a user runconfig take precedence.
+
+        Parameters
+        ----------
+        param_grp_cls_obj : Type[YamlParamGroup]
+            A class instance of a subclass of YamlParamGroup.
+            For example, `HistogramParamGroup`.
+        user_rncfg : nested dict, optional
+            A dict containing the user's runconfig values that (at minimum)
+            correspond to the `param_grp_cls_obj` parameters. (Other values
+            will be ignored.) For example, a QA runconfig yaml loaded directly
+            into a dict would be a perfect input for `user_rncfg`.
+            The nested structure of `user_rncfg` must match the structure
+            of the QA runconfig yaml file for this parameter group.
+            To see the expected yaml structure for e.g. RSLC, run
+            `nisarqa dumpconfig rslc` from the command line.
+            If `user_rncfg` contains entries that do not correspond to
+            attributes in `param_grp_cls_obj`, they will be ignored.
+            If `user_rncfg` is either None, an empty dict, or does not contain
+            values for `param_grp_cls_obj` in a nested structure that matches
+            the QA runconfig group that corresponds to `param_grp_cls_obj`,
+            then an instance with all default values will be returned.
+
+        Returns
+        -------
+        param_grp_instance : `param_grp_cls_obj` instance
+            An instance of `param_grp_cls_obj` that is fully instantiated
+            using default values and the arguments provided in `user_rncfg`.
+            The values in `user_rncfg` have precedence over the defaults.
+        """
+
+        if not user_rncfg:
+            # If user_rncfg is None or is an empty dict, then return the default
+            return param_grp_cls_obj()
+
+        # Get the runconfig path for this *ParamGroup
+        rncfg_path = param_grp_cls_obj.get_path_to_group_in_runconfig()
+
+        try:
+            runcfg_grp_dict = nisarqa.get_nested_element_in_dict(
+                user_rncfg, rncfg_path
+            )
+        except KeyError:
+            # Group was not found, so construct an instance using all defaults.
+            # If a dataclass has a required parameter, this will (correctly)
+            # throw another error.
+            return param_grp_cls_obj()
+        else:
+            # Get the relevant yaml runconfig parameters for this ParamGroup
+            yaml_names = param_grp_cls_obj.get_dict_of_yaml_names()
+
+            # prune extraneous fields from the runconfig group
+            # (aka keep only the runconfig fields that are relevant to QA)
+            # The "if..." logic will allow us to skip missing runconfig fields.
+            user_input_args = {
+                cls_attr_name: runcfg_grp_dict[yaml_name]
+                for cls_attr_name, yaml_name in yaml_names.items()
+                if yaml_name in runcfg_grp_dict
+            }
+
+            return param_grp_cls_obj(**user_input_args)
+
+    @classmethod
+    def from_runconfig_file(
+        cls: type[nisarqa.RootParamGroupT],
+        runconfig_yaml: str | os.PathLike,
+        product_type: str,
+    ) -> nisarqa.RootParamGroupT:
+        """
+        Get a *RootParamGroup for `product_type` from a QA Runconfig YAML file.
+
+        The input runconfig file must follow the standard QA runconfig
+        format for `product_type`.
+        For an example runconfig template with default parameters,
+        run the command line command 'nisar_qa dumpconfig <product_type>'.
+        (Ex: Use 'nisarqa dumpconfig rslc' for the RSLC runconfig template.)
+
+        Parameters
+        ----------
+        runconfig_yaml : path-like
+            Filename (with path) to a QA runconfig YAML file for `product_type`.
+        product_type : str
+            One of: 'rslc', 'gslc', 'gcov', 'rifg', 'runw', 'gunw', 'roff',
+            or 'goff'.
+
+        Returns
+        -------
+        root_params : nisarqa.RootParamGroup
+            An instance of *RootParamGroup corresponding to `product_type`.
+            For example, if `product_type` is 'gcov', the return type
+            will be `nisarqa.GCOVRootParamGroup`. This will be
+            populated with runconfig values where provided,
+            and default values for missing runconfig parameters.
+
+        Raises
+        ------
+        nisarqa.ExitEarly
+            If all `workflows` were set to False in the runconfig.
+
+        See Also
+        --------
+        RootParamGroup.from_runconfig_dict
+        """
+        # parse runconfig into a dict structure
+        log = nisarqa.get_logger()
+        log.info(f"Begin loading user runconfig yaml to dict: {runconfig_yaml}")
+        user_rncfg = nisarqa.load_user_runconfig(runconfig_yaml)
+
+        log.info("Begin parsing of runconfig for user-provided QA parameters.")
+
+        # Build the *RootParamGroup parameters per the runconfig
+        # (Raises an ExitEarly exception if all workflows in runconfig are
+        # set to False)
+        root_params = cls.from_runconfig_dict(
+            user_rncfg=user_rncfg, product_type=product_type
+        )
+        log.info("Loading of user runconfig complete.")
+
+        return root_params
+
 
 __all__ = nisarqa.get_all(__name__, objects_to_skip)