Merge pull request #195 from NOAA-GFDL/194.bug-fix

Various bug fixes for AM5 analysis scripts

Author: ceblanton

Jinja2Filters/form_remap_dep.py

@@ -1,27 +1,27 @@
-""" 
+"""
 form_remap_dep:
 - parses the remap_pp_components rose-app.conf
 - uses input from rose-suite.conf in the form of env variables
 - returns the pp component and source name dependencies for
-  remap_pp_components task execution. 
+  remap_pp_components task execution.
 
 For instance, for an atmos PP component that requires the regridded
 atmos_month and regridded atmos_daily history files, this JinjaFilter
 when called within flow.cylc helps identify this dependency to
-complete the corresponding task graph. 
+complete the corresponding task graph.
 
 This JinjaFilter ensures a remap-pp-component only waits for the
 dependent make-timeseries tasks such that the succeeded components
-output are made available in the final destination. 
+output are made available in the final destination.
 
 See form_remap_dep invocations from flow.cylc
 """
 # Function parameter type hints, PEP 484
 
 import os
 from pathlib import Path
-import yaml
 import logging
+import yaml
 
 # set up logging
 logging.basicConfig()
@@ -35,22 +35,22 @@ def form_remap_dep(grid_type: str,
                    output_type: str,
                    yamlfile: str,
                    history_segment: str=None) -> str:
-    """ 
+    """
     Form the task parameter list based on the grid type,
     the temporal type, and the desired pp component(s)
 
     Arguments:
        @param grid_type (str): One of: native or regrid-xy
        @param temporal_type (str): One of: temporal or static
-       @param chunk (str): e.g P5Y for 5-year time series 
+       @param chunk (str): e.g P5Y for 5-year time series
        @param pp_component (str): all, or a space-separated list
        @param output_type (str): ts or av
        @param yamlfile (str): yaml configuration file passed through workflow
        @param history_segment (str): if given, handle special case where history
                                      segment equals pp-chunk-a
 
        @return remap_dep (multiline str) with Jinja formatting listing source-pp
-               dependencies  
+               dependencies
     """
     pp_components = pp_components_str.split(' ')
     if grid_type == "regrid-xy":

Jinja2Filters/form_task_parameters.py

@@ -1,5 +1,5 @@
 """
-Form the task parameter list based on the grid type, 
+Form the task parameter list based on the grid type,
 the temporal type,and the desired pp component(s)
 """
 

Jinja2Filters/get_analysis_info.py

@@ -1,12 +1,12 @@
+import logging
 from pathlib import Path
 
 from metomi.isodatetime.parsers import DurationParser, TimePointParser
 from yaml import safe_load
 
-from legacy_date_conversions import *
+from legacy_date_conversions import convert_iso_duration_to_bronx_chunk
 
 # set up logging
-import logging
 logging.basicConfig()
 logger = logging.getLogger(__name__)
 logger.setLevel(logging.INFO)
@@ -17,9 +17,9 @@
 time_parser = TimePointParser(assumed_time_zone=(0, 0))
 
 
-class AnalysisScript(object):
+class AnalysisScript:
     def __init__(self, name, config, experiment_components, experiment_starting_date,
-                 experiment_stopping_date, pp_chunks):
+                 experiment_stopping_date, pp_chunks, yaml):
         """Initialize the analysis script object.
 
         Args:
@@ -29,13 +29,14 @@ def __init__(self, name, config, experiment_components, experiment_starting_date
             experiment_starting_date: Starting date for the experiment.
             experiment_stopping_date: Stopping date for the experiment.
             pp_chunks: List of ISO8601 durations used by the workflow.
+            yaml: Resolved postprocessing yaml
         """
         self.name = name
         logger.debug(f"{name}: initializing AnalysisScript instance")
 
         # Skip if configuration wants to skip it
         self.switch = config["workflow"]["analysis_on"]
-        if self.switch == False:
+        if self.switch is False:
             return
 
         # Skip if the components are not available
@@ -55,9 +56,30 @@ def __init__(self, name, config, experiment_components, experiment_starting_date
         self.script_type = config["workflow"]["script_type"]
         self.chunk = duration_parser.parse(config["workflow"]["chunk_size"])
 
-        if self.chunk not in pp_chunks:
-            raise ValueError(f"ERROR: Analysis script '{self.name}' requests chunk size '{self.chunk}', but " +
-                             f"this chunk size is not declared in 'pp_chunks'")
+        # Retrieve other config
+        self.data_frequency = config["required"]["data_frequency"]
+
+        # check for needed pp prerequisites
+        if self.product not in ['av', 'ts']:
+            raise ValueError("ERROR: product type must be 'ts' or 'av'")
+        if self.product == "ts":
+            if self.chunk not in pp_chunks:
+                raise ValueError(f"ERROR: Analysis script '{self.name}' requests timeseries chunk size '{self.chunk}', but " +
+                                 "this chunk size is not declared in 'pp_chunks'")
+        else:
+            # Loop through the components and look for the ones specified by the analysis script
+            # For each component to check, confirm that its climatology section contains the requested climo chunk
+            for ana_comp in config["workflow"]["components"]:
+                found_needed_inputs_for_component = False
+                for exp_comp in yaml["postprocess"]["components"]:
+                    if exp_comp["type"] == ana_comp:
+                        if 'climatology' in exp_comp:
+                            for climo_request in exp_comp["climatology"]:
+                                if climo_request["frequency"] == self.data_frequency and climo_request["interval_years"] == self.chunk.years:
+                                    found_needed_inputs_for_component = True
+                if not found_needed_inputs_for_component:
+                    raise ValueError(f"ERROR: Analysis script '{self.name}' requests climatology chunk size '{self.chunk}', but " +
+                                     f"no suitable climatology sections were found in postprocess component '{ana_comp}'")
 
         # Parse the new analysis config items
         if 'legacy' in config:
@@ -73,8 +95,6 @@ def __init__(self, name, config, experiment_components, experiment_starting_date
         else:
             self.is_legacy = False
 
-        self.data_frequency = config["required"]["data_frequency"]
-
         # if dates are years, convert to string or else ISO conversion will fail
         if isinstance(config["required"]["date_range"][0], int):
             one = "{:04d}".format(config["required"]["date_range"][0])
@@ -98,7 +118,7 @@ def graph(self, analysis_only):
         Returns:
             String cylc task graph for the analysis.
         """
-        if self.switch == False:
+        if self.switch is False:
             return ""
 
         graph = ""
@@ -191,7 +211,7 @@ def definition(self, pp_dir):
         Returns:
             Cylc task definition string for this analysis script
         """
-        if self.switch == False:
+        if self.switch is False:
             return ""
 
         definitions = ""
@@ -270,12 +290,12 @@ def definition(self, pp_dir):
         new_analysis_str = f"""
     [[analysis-{self.name}]]
         script = '''
-fre analysis run \
-    --name              freanalysis_{self.name} \
-    --catalog           $catalog \
-    --output-directory  $out_dir/{self.name} \
-    --output-yaml       $out_dir/{self.name}/output.yaml \
-    --experiment-yaml   $experiment_yaml \
+fre analysis run
+    --name              freanalysis_{self.name}
+    --catalog           $catalog
+    --output-directory  $out_dir/{self.name}
+    --output-yaml       $out_dir/{self.name}/output.yaml
+    --experiment-yaml   $experiment_yaml
     --library-directory $CYLC_WORKFLOW_SHARE_DIR/analysis-envs/freanalysis_{self.name}
         '''
         # retry 10 times (due to mysterious intake-esm issue)
@@ -293,25 +313,27 @@ def definition(self, pp_dir):
         '''
         """
 
-        if self.script_type == "independent" and self.date_range == self.experiment_date_range:
+        if self.script_type == "independent":
             # to make the task run, we will create a corresponding task graph below
             # corresponding to the interval (chunk), e.g. ANALYSIS-P1Y.
             # Then, the analysis script will inherit from that family, to enable
             # both the task triggering and the yr1 and datachunk template vars.
-            logger.info(f"{self.name}: Will run every chunk {self.chunk}")
+            logger.debug(f"{self.name}: Will run every chunk {self.chunk}")
             if self.is_legacy:
                 definitions += legacy_analysis_str
             else:
                 definitions += new_analysis_str
 
             # create the task family for all every-interval analysis scripts
+            interval_years_minus_one = self.chunk - one_year
             definitions += f"""
     [[data-catalog-{self.chunk}]]
         inherit = DATA-CATALOG
     [[ANALYSIS-{self.chunk}]]
         inherit = ANALYSIS
         [[[environment]]]
             yr1 = $(cylc cycle-point --template=CCYY)
+            yr2 = $(cylc cycle-point --template=CCYY --offset-years={interval_years_minus_one.years})
             databegyr = $yr1
             dataendyr = $yr2
             datachunk = {self.chunk.years}
@@ -346,12 +368,12 @@ def definition(self, pp_dir):
             logger.debug(f"{self.name}: Finished determining scripting")
             return definitions
 
-        if self.script_type == "cumulative" and self.date_range == self.experiment_date_range:
+        if self.script_type == "cumulative":
             # Case 2: run the analysis every chunk, but depend on all previous chunks too.
             # To make the task run, we will create a task family for
             # each chunk/interval, starting from the beginning of pp data
             # then we create an analysis script task for each of these task families.
-            logger.info(f"{self.name}: Will run each chunk {self.chunk} from beginning {self.experiment_date_range[0]}")
+            logger.debug(f"{self.name}: Will run each chunk {self.chunk} from beginning {self.experiment_date_range[0]}")
             date = self.experiment_date_range[0]
             while date <= self.experiment_date_range[1]:
                 date_str = f"{date.year:04}"
@@ -428,7 +450,7 @@ def definition(self, pp_dir):
                 d2 -= self.chunk
             d1_str = f"{d1.year:04}"
             d2_str = f"{d2.year:04}"
-            logger.info(f"{self.name}: Will run once for time period {self.date_range[0]} to {self.date_range[1]} (chunks {d1_str} to {d2_str})")
+            logger.debug(f"{self.name}: Will run once for time period {self.date_range[0]} to {self.date_range[1]} (chunks {d1_str} to {d2_str})")
             date1_str = f"{self.date_range[0].year:04}"
             date2_str = f"{self.date_range[1].year:04}"
 
@@ -494,9 +516,9 @@ def task_generator(yaml_, experiment_components, experiment_start, experiment_st
     for script_name, script_params in yaml_["analysis"].items():
         # Retrieve information about the script
         script_info = AnalysisScript(script_name, script_params, experiment_components,
-                                     experiment_start, experiment_stop, pp_chunks)
-        if script_info.switch == False:
-            logger.info(f"{script_name}: Skipping, switch set to off")
+                                     experiment_start, experiment_stop, pp_chunks, yaml_)
+        if script_info.switch is False:
+            logger.debug(f"{script_name}: Skipping, switch set to off")
             continue
         yield script_info
 
@@ -575,7 +597,7 @@ def get_analysis_info(experiment_yaml, info_type, experiment_components, pp_dir,
             logger.debug("get_analysis_info: about to return graph")
             return task_graph(yaml_, experiment_components, experiment_start,
                               experiment_stop, pp_chunks, analysis_only)
-        elif info_type == "task-definitions":
+        if info_type == "task-definitions":
             logger.debug("get_analysis_info: about to return definitions")
             return task_definitions(yaml_, experiment_components, experiment_start,
                                    experiment_stop, pp_chunks, pp_dir)

Jinja2Filters/get_climatology_info.py

@@ -1,17 +1,14 @@
-from pathlib import Path
-
+import logging
 import metomi.isodatetime.dumpers
 import metomi.isodatetime.parsers
 from yaml import safe_load
-import pprint
 
 from legacy_date_conversions import *
 
 # set up logging
-import logging
 logging.basicConfig()
 logger = logging.getLogger(__name__)
-logger.setLevel(logging.DEBUG)
+logger.setLevel(logging.INFO)
 
 # Global variables just set to reduce typing a little.
 duration_parser = metomi.isodatetime.parsers.DurationParser()
@@ -22,9 +19,9 @@
 def sort_pp_chunks(unsorted_strings):
     """Create descending list of pp chunk durations"""
     durations = []
-    for s in unsorted_strings:
-        durations.append(duration_parser.parse(s))
-    return(sorted(durations, reverse=True))
+    for string in unsorted_strings:
+        durations.append(duration_parser.parse(string))
+    return sorted(durations, reverse=True)
 
 def lookup_source_for_component(yaml_, component):
     """Return list of history files associated with a pp component"""
@@ -33,9 +30,9 @@ def lookup_source_for_component(yaml_, component):
         if item["type"] == component:
             for source in item["sources"]:
                 sources.append(source["history_file"])
-    return(sources)
+    return sources
 
-class Climatology(object):
+class Climatology:
     def __init__(self, component, frequency, interval_years, pp_chunk, sources, grid):
         """Initialize the climatology object
 
@@ -74,10 +71,10 @@ def graph(self, history_segment, clean_work):
 
         chunks_per_interval = self.interval_years / self.pp_chunk.years
         assert chunks_per_interval == int(chunks_per_interval)
-        for index, source in enumerate(self.sources):
+        for source in self.sources:
             count = 0
             while count < chunks_per_interval:
-                if index == 0:
+                if count == 0:
                     connector = ""
                 else:
                     connector = " & "
@@ -97,7 +94,10 @@ def graph(self, history_segment, clean_work):
             graph += f" => climo-{self.frequency}-P{self.interval_years}Y_{self.component}\n"
             graph += f" => remap-climo-{self.frequency}-P{self.interval_years}Y_{self.component}\n"
             graph += f" => combine-climo-{self.frequency}-P{self.interval_years}Y_{self.component}\n"
-            graph += f"\"\"\"\n"
+            if clean_work:
+                graph += f"remap-climo-{self.frequency}-P{self.interval_years}Y_{self.component} => clean-shards-av-P{self.interval_years}Y\n"
+                graph += f"combine-climo-{self.frequency}-P{self.interval_years}Y_{self.component} => clean-pp-timeavgs-P{self.interval_years}Y\n"
+        graph += f"\"\"\"\n"
 
         # then, create the cleaning tasks
         if clean_work:
@@ -140,12 +140,14 @@ def definition(self, clean_work):
 
         definitions += f"""
     [[combine-climo-{self.frequency}-P{self.interval_years}Y_{self.component}]]
-        inherit = COMBINE-TIMEAVGS
+        inherit = COMBINE-TIMEAVGS-P{self.interval_years}Y
         [[[environment]]]
             component = {self.component}
             frequency = {self.frequency}
             interval = P{self.interval_years}Y
             end = $(cylc cycle-point --print-year --offset={offset})
+    [[COMBINE-TIMEAVGS-P{self.interval_years}Y]]
+        inherit = COMBINE-TIMEAVGS
         """
 
         if clean_work:
@@ -163,8 +165,6 @@ def definition(self, clean_work):
         return definitions
 
 def task_generator(yaml_):
-    history_segment = yaml_["postprocess"]["settings"]["history_segment"]
-
     # Retrieve the pp components
     components = []
     for component in yaml_["postprocess"]["components"]:
@@ -265,7 +265,7 @@ def get_climatology_info(experiment_yaml, info_type):
         if info_type == "task-graph":
             logger.debug("about to return graph")
             return task_graphs(yaml_, history_segment, clean_work)
-        elif info_type == "task-definitions":
+        if info_type == "task-definitions":
             logger.debug("about to return definitions")
             return task_definitions(yaml_, clean_work)
 

Jinja2Filters/get_components.py

@@ -1,10 +1,7 @@
-import re
-import os
-from pathlib import Path
+import logging
 import yaml
 
 # set up logging
-import logging
 logging.basicConfig(level=logging.INFO)
 fre_logger = logging.getLogger(__name__)
 
@@ -26,4 +23,4 @@ def get_components(yamlfile):
 
     # we want to return a list, but some other scripts are expecting a space-separated string
     #return(components)
-    return(" ".join(components))
+    return " ".join(components)