TaskTrigger Refactor

lib/cylc/conditional_simplifier.py

@@ -19,22 +19,50 @@
 import re
 import ast
 import copy
-import sys
 
 
 class ConditionalSimplifier(object):
     """A class to simplify logical expressions"""
+    RE_CONDITIONALS = re.compile("(&|\||\(|\))")
 
     def __init__(self, expr, clean):
         self.raw_expression = expr
         self.clean_list = clean
         self.nested_expr = self.format_expr(self.raw_expression)
 
-    def listify(self, message):
-        """Convert a string containing a logical expression to a list"""
+    def get_cleaned(self):
+        """Return the simplified logical expression"""
+        cleaned = self.nested_expr
+        for item in self.clean_list:
+            cleaned = self.clean_expr(cleaned, item)
+        cleaned = self.flatten_nested_expr(cleaned)
+        return cleaned
+
+    @classmethod
+    def listify(cls, message):
+        """Convert a string containing a logical expression to a list
+
+        Examples:
+            >>> ConditionalSimplifier.listify('(foo)')
+            [['foo']]
+
+            >>> ConditionalSimplifier.listify('foo & (bar | baz)')
+            ['foo', '&', ['bar', '|', 'baz']]
+
+            >>> ConditionalSimplifier.listify('(a&b)|(c|d)&(e|f)')
+            [['a', '&', 'b'], '|', ['c', '|', 'd'], '&', ['e', '|', 'f']]
+
+            >>> ConditionalSimplifier.listify('a & (b, c,)')
+            ['a', '&', ['b, c,']]
+
+            >>> ConditionalSimplifier.listify('((foo)')
+            Traceback (most recent call last):
+            SyntaxError: unexpected EOF while parsing
+
+        """
         message = message.replace("'", "\"")
-        RE_CONDITIONALS = "(&|\||\(|\))"
-        tokenised = re.split("(&|\||\(|\))", message)
+
+        tokenised = cls.RE_CONDITIONALS.split(message)
         listified = ["["]
         for item in tokenised:
             if item.strip() != "" and item.strip() not in ["(", ")"]:
@@ -52,7 +80,8 @@ def listify(self, message):
         listified = ast.literal_eval(listified)
         return listified
 
-    def get_bracketed(self, nest_me):
+    @classmethod
+    def get_bracketed(cls, nest_me):
         """Nest a list according to any brackets in it"""
         start = 0
         finish = len(nest_me)
@@ -69,24 +98,17 @@ def get_bracketed(self, nest_me):
                 finish = i
                 break
         bracket_nested = nest_me[0:start + 1]
-        bracket_nested.append(self.get_bracketed(nest_me[start + 1:finish]))
+        bracket_nested.append(cls.get_bracketed(nest_me[start + 1:finish]))
         bracket_nested.extend(nest_me[finish:len(nest_me)])
         return bracket_nested
 
-    def get_cleaned(self):
-        """Return the simplified logical expression"""
-        cleaned = self.nested_expr
-        for item in self.clean_list:
-            cleaned = self.clean_expr(cleaned, item)
-        cleaned = self.flatten_nested_expr(cleaned)
-        return cleaned
-
-    def nest_by_oper(self, nest_me, oper):
+    @classmethod
+    def nest_by_oper(cls, nest_me, oper):
         """Nest a list based on a specified logical operation"""
         found = False
         for i in range(len(nest_me)):
             if isinstance(nest_me[i], list):
-                nest_me[i] = self.nest_by_oper(nest_me[i], oper)
+                nest_me[i] = cls.nest_by_oper(nest_me[i], oper)
             if nest_me[i] == oper:
                 found = i
                 break
@@ -97,11 +119,12 @@ def nest_by_oper(self, nest_me, oper):
             nested += [nest_me[found - 1:found + 2]]
             if (found + 2) < len(nest_me):
                 nested += nest_me[found + 2:]
-            return self.nest_by_oper(nested, oper)
+            return cls.nest_by_oper(nested, oper)
         else:
             return nest_me
 
-    def clean_expr(self, nested_list, criterion):
+    @classmethod
+    def clean_expr(cls, nested_list, criterion):
         """Return a list with entries specified by 'criterion' removed"""
         cleaned = copy.deepcopy(nested_list)
 
@@ -123,7 +146,7 @@ def clean_expr(self, nested_list, criterion):
         found = None
         for i in range(0, len(cleaned)):
             if isinstance(cleaned[i], list):
-                cleaned[i] = self.clean_expr(cleaned[i], criterion)
+                cleaned[i] = cls.clean_expr(cleaned[i], criterion)
             if cleaned[i] in [criterion, '']:
                 found = i
                 break
@@ -134,25 +157,27 @@ def clean_expr(self, nested_list, criterion):
                 cleaned = cleaned[2:]
             else:
                 del cleaned[found - 1:found + 1]
-            return self.clean_expr(cleaned, criterion)
+            return cls.clean_expr(cleaned, criterion)
         else:
             return cleaned
 
-    def format_expr(self, expr):
+    @classmethod
+    def format_expr(cls, expr):
         """Carry out list conversion and nesting of a logical expression in
         the correct order."""
-        listified = self.listify(expr)
-        bracketed = self.get_bracketed(listified)
-        nested_by_and = self.nest_by_oper(bracketed, "&")
-        nested_by_or = self.nest_by_oper(nested_by_and, "|")
+        listified = cls.listify(expr)
+        bracketed = cls.get_bracketed(listified)
+        nested_by_and = cls.nest_by_oper(bracketed, "&")
+        nested_by_or = cls.nest_by_oper(nested_by_and, "|")
         return nested_by_or
 
-    def flatten_nested_expr(self, expr):
+    @classmethod
+    def flatten_nested_expr(cls, expr):
         """Convert a logical expression in a nested list back to a string"""
         flattened = copy.deepcopy(expr)
         for i in range(len(flattened)):
             if isinstance(flattened[i], list):
-                flattened[i] = self.flatten_nested_expr(flattened[i])
+                flattened[i] = cls.flatten_nested_expr(flattened[i])
         if isinstance(flattened, list):
             flattened = (" ").join(flattened)
         flattened = "(" + flattened

lib/cylc/config.py

@@ -29,6 +29,7 @@
 import traceback
 
 from cylc.c3mro import C3
+from cylc.conditional_simplifier import ConditionalSimplifier
 from cylc.exceptions import CylcError
 from cylc.graph_parser import GraphParser
 from cylc.param_expand import NameExpander
@@ -44,30 +45,23 @@
 from cylc.print_tree import print_tree
 from cylc.taskdef import TaskDef, TaskDefError
 from cylc.task_id import TaskID
-from cylc.task_trigger import TaskTrigger
+from cylc.task_trigger import TaskTrigger, Dependency
 from cylc.wallclock import get_current_time_string
 from isodatetime.data import Calendar
 from isodatetime.parsers import DurationParser
 from parsec.OrderedDict import OrderedDictWithDefaults
 from parsec.util import replicate
 from cylc.suite_logging import OUT, ERR
-
+from cylc.task_outputs import TASK_OUTPUT_SUCCEEDED
 
 RE_SUITE_NAME_VAR = re.compile('\${?CYLC_SUITE_(REG_)?NAME}?')
 RE_TASK_NAME_VAR = re.compile('\${?CYLC_TASK_NAME}?')
 CLOCK_OFFSET_RE = re.compile(r'(' + TaskID.NAME_RE + r')(?:\(\s*(.+)\s*\))?')
 EXT_TRIGGER_RE = re.compile('(.*)\s*\(\s*(.+)\s*\)\s*')
 NUM_RUNAHEAD_SEQ_POINTS = 5  # Number of cycle points to look at per sequence.
 
-# Replace \W characters in conditional graph expressions.
-CONDITIONAL_REGEX_REPLACEMENTS = [
-    ("\[", "_leftsquarebracket_"),
-    ("\]", "_rightsquarebracket_"),
-    ("-", "_minus_"),
-    ("\^", "_caret_"),
-    (":", "_colon_"),
-    ("\+", "_plus_"),
-]
+# Message trigger offset regex.
+BCOMPAT_MSG_RE_C6 = re.compile(r'^(.*)\[\s*(([+-])?\s*(.*))?\s*\](.*)$')
 
 try:
     import cylc.graphing
@@ -1504,55 +1498,105 @@ def generate_taskdefs(self, orig_expr, left_nodes, right, section, seq,
 
             # Record custom message outputs.
             for item in self.cfg['runtime'][name]['outputs'].items():
-                if (item, base_interval) not in self.taskdefs[name].outputs:
-                    self.taskdefs[name].outputs.append((item, base_interval))
+                if item not in self.taskdefs[name].outputs:
+                    self.taskdefs[name].outputs.append(item)
+
+    def generate_triggers(self, lexpression, left_nodes, right, seq,
+                          suicide, base_interval, task_triggers):
+        """Create Dependency and TaskTrigger objects.
+
+        Register dependency with the relevant TaskDef object.
 
-    def generate_triggers(self, lexpression, left_nodes,
-                          right, seq, suicide, base_interval):
+        """
         if not right or not left_nodes:
             # Lone nodes have no triggers.
             return
 
-        ctrig = {}
-        cname = {}
+        # Convert expression to a (nested) list.
+        try:
+            expr_list = ConditionalSimplifier.listify(lexpression)
+        except SyntaxError:
+            raise SuiteConfigError('Error in expression "%s"' % lexpression)
+
+        def recursive_replace(lst, obj, replacement):
+            """Replace occurances of an object in a list with a standin.
+
+            Works recursively to replace instances in nested lists.
+
+            """
+            ret = []
+            for item in lst:
+                if type(item) is list:
+                    ret.append(recursive_replace(item, obj, replacement))
+                elif item == obj:
+                    ret.append(replacement)
+                else:
+                    ret.append(item)
+            return ret
+
+        trigers = set([])
         for left in left_nodes:
             # (GraphNodeError checked above)
-            cycle_point = None
             lnode = graphnode(left, base_interval=base_interval)
             ltaskdef = self.taskdefs[lnode.name]
 
+            # Determine intercycle offsets.
+            abs_cycle_point = None
+            cycle_point_offset = None
             if lnode.offset_is_from_ict:
                 first_point = get_point_relative(
                     lnode.offset_string, self.initial_point)
                 last_point = seq.get_stop_point()
+                abs_cycle_point = first_point
                 if last_point is None:
                     # This dependency persists for the whole suite run.
                     ltaskdef.intercycle_offsets.add(
                         (None, seq))
                 else:
                     ltaskdef.intercycle_offsets.add(
                         (str(-(last_point - first_point)), seq))
-                cycle_point = first_point
+            elif lnode.is_absolute:
+                abs_cycle_point = lnode.offset_string
             elif lnode.intercycle:
                 if lnode.offset_is_irregular:
                     offset_tuple = (lnode.offset_string, seq)
                 else:
                     offset_tuple = (lnode.offset_string, None)
                 ltaskdef.intercycle_offsets.add(offset_tuple)
+                cycle_point_offset = lnode.offset_string
 
-            trig = TaskTrigger(
-                lnode.name, lnode.output, lnode.offset_string, cycle_point,
-                suicide, self.cfg['runtime'][lnode.name]['outputs'],
-                base_interval)
+            # Check for message trigger offsets.
+            outputs = self.cfg['runtime'][lnode.name]['outputs']
+            for message in outputs.values():
+                if BCOMPAT_MSG_RE_C6.match(message):
+                    raise SuiteConfigError(
+                        'ERROR: Message trigger offsets are obsolete.')
 
-            # Use fully qualified name for trigger expression label
-            # (task name is not unique, e.g.: "F | F:fail => G").
-            label = self.get_conditional_label(left)
-            ctrig[label] = trig
-            cname[label] = lnode.name
+            # Qualifier.
+            if outputs and lnode.output in outputs:
+                # Task message.
+                qualifier = outputs[lnode.output]
+            else:
+                # Built-in qualifier.
+                if lnode.output:
+                    qualifier = TaskTrigger.get_trigger_name(lnode.output)
+                else:
+                    qualifier = TASK_OUTPUT_SUCCEEDED
 
-        expr = self.get_conditional_label(lexpression)
-        self.taskdefs[right].add_trigger(ctrig, expr, seq)
+            # Generate TaskTrigger if not already done.
+            key = (lnode.name, abs_cycle_point, cycle_point_offset, qualifier)
+            try:
+                task_trigger = task_triggers[key]
+            except KeyError:
+                task_trigger = TaskTrigger(*key)
+                task_triggers[key] = task_trigger
+
+            trigers.add(task_trigger)
+
+            expr_list = recursive_replace(expr_list, left, task_trigger)
+
+        dependency = Dependency(expr_list, trigers, suicide)
+        self.taskdefs[right].add_dependency(dependency, seq)
 
     def get_actual_first_point(self, start_point):
         # Get actual first cycle point for the suite (get all
@@ -1574,30 +1618,6 @@ def get_actual_first_point(self, start_point):
             self.actual_first_point = start_point
         return self.actual_first_point
 
-    def get_conditional_label(self, expression):
-        """Return a label to ID the expression.
-
-        Special characters such as [, or ^ are replaced with
-        nice \w+ text for use in regular expressions and trigger
-        task matching. We don't back-transform the label, so
-        all it needs to is provide locally unique IDs for the
-        bits of the trigger.
-
-        For example, "foo[^] | bar" is represented in text as
-        "foo_leftsquarebracket__caret__rightsquarebracket_ | bar".
-        As long as no one uses that exact "foo_leftsquare...." text as
-        a task name as part of a conditional trigger for the *same*
-        task, we're OK.
-
-        Should we use unicodedata.name to convert the character names,
-        and support much more characters in the task names?
-
-        """
-        label = expression
-        for regex, replacement in CONDITIONAL_REGEX_REPLACEMENTS:
-            label = re.sub(regex, replacement, label)
-        return label
-
     def get_graph_raw(self, start_point_string, stop_point_string,
                       group_nodes=None, ungroup_nodes=None,
                       ungroup_recursive=False, group_all=False,
@@ -1896,6 +1916,7 @@ def load_graph(self):
                 sections.append((section, sec_map['graph']))
 
         # Parse and process each graph section.
+        task_triggers = {}
         for section, graph in sections:
             try:
                 seq = get_sequence(section, icp, fcp)
@@ -1914,9 +1935,10 @@ def load_graph(self):
             gp = GraphParser(family_map, self.parameters)
             gp.parse_graph(graph)
             self.suite_polling_tasks.update(gp.suite_state_polling_tasks)
-            self._proc_triggers(gp.triggers, gp.original, section, seq)
+            self._proc_triggers(gp.triggers, gp.original, section, seq,
+                                task_triggers)
 
-    def _proc_triggers(self, triggers, original, section, seq):
+    def _proc_triggers(self, triggers, original, section, seq, task_triggers):
         """Define graph edges, taskdefs, and triggers, from graph sections."""
         base_interval = seq.get_interval()
         for right, val in triggers.items():
@@ -1929,7 +1951,8 @@ def _proc_triggers(self, triggers, original, section, seq):
                 self.generate_taskdefs(
                     orig_expr, lefts, right, section, seq, base_interval)
                 self.generate_triggers(
-                    expr, lefts, right, seq, suicide, base_interval)
+                    expr, lefts, right, seq, suicide, base_interval,
+                    task_triggers)
 
     def find_taskdefs(self, name):
         """Find TaskDef objects in family "name" or matching "name".