Tidy set-command logging etc.

Author: hjoliver

cylc/flow/id.py

@@ -529,7 +529,11 @@ def duplicate(
 )
 
 
-def quick_relative_id(cycle: Union[str, int, 'PointBase'], task: str) -> str:
+def quick_relative_id(
+    cycle: Union[str, int, 'PointBase'],
+    task: str,
+    output: Optional[str] = None
+) -> str:
     """Generate a relative ID for a task.
 
     This is a more efficient solution to `Tokens` for cases where
@@ -541,7 +545,10 @@ def quick_relative_id(cycle: Union[str, int, 'PointBase'], task: str) -> str:
         True
 
     """
-    return f'{cycle}/{task}'
+    if output is None:
+        return f'{cycle}/{task}'
+    else:
+        return f'{cycle}/{task}:{output}'
 
 
 def _dict_strip(dictionary):

cylc/flow/prerequisite.py

@@ -26,7 +26,6 @@
     KeysView,
     NamedTuple,
     Optional,
-    Set,
     Tuple,
     Union,
 )
@@ -57,9 +56,12 @@ class PrereqTuple(NamedTuple):
     task: str
     output: str
 
-    def get_id(self) -> str:
+    def get_id(self, show_output=False) -> str:
         """Get the relative ID of the task in this prereq output."""
-        return quick_relative_id(self.point, self.task)
+        if show_output:
+            return quick_relative_id(self.point, self.task, self.output)
+        else:
+            return quick_relative_id(self.point, self.task)
 
     @staticmethod
     def coerce(tuple_: AnyPrereqTuple) -> 'PrereqTuple':
@@ -265,32 +267,27 @@ def satisfy_me(
         outputs: Iterable['Tokens'],
         mode: Optional[RunMode] = None,
         forced: bool = False,
-    ) -> 'Set[Tokens]':
+    ) -> None:
         """Set the given outputs as satisfied (if they are not already).
 
-        Return outputs that match.
-
         Args:
             outputs: List of outputs to satisfy.
             mode: Task run mode.
             forced: If True, records that this should not be undone by
                 `cylc remove`.
         """
-        valid = set()
         for output in outputs:
             output_tuple = PrereqTuple(
                 output['cycle'], output['task'], output['task_sel']
             )
             if output_tuple not in self._satisfied:
                 continue
-            valid.add(output)
             if not self._satisfied[output_tuple]:
                 self[output_tuple] = (
                     'force satisfied' if forced
                     else 'satisfied by skip mode' if mode == RunMode.SKIP
                     else 'satisfied naturally'
                 )
-        return valid
 
     def api_dump(self) -> Optional[PbPrerequisite]:
         """Return list of populated Protobuf data objects."""

cylc/flow/subprocctx.py

@@ -199,13 +199,20 @@ def update_command(self, workflow_run_dir):
                     json.dumps(self.func_kwargs),
                     workflow_run_dir]
 
-    def get_signature(self):
-        """Return the function call signature (as a string)."""
+    def get_signature(self) -> str:
+        """Return the function call signature."""
         skeys = sorted(self.func_kwargs.keys())
         args = self.func_args + [
             "%s=%s" % (i, self.func_kwargs[i]) for i in skeys]
         return "%s(%s)" % (self.func_name, ", ".join([str(a) for a in args]))
 
+    def get_description(self, show_intvl: bool = False) -> str:
+        """Return a complete description of the function."""
+        descr = f"{self.label} = {self.get_signature()}"
+        if show_intvl:
+            descr = f"call interval {self.intvl}s: {descr}"
+        return descr
+
     def dump(self) -> str:
         """Output for logging."""
         return SubProcContext.__str__(self)

cylc/flow/task_pool.py

@@ -60,6 +60,7 @@
 from cylc.flow.id_cli import contains_fnmatch
 from cylc.flow.id_match import filter_ids
 from cylc.flow.platforms import get_platform
+from cylc.flow.prerequisite import PrereqTuple
 from cylc.flow.run_modes import RunMode
 from cylc.flow.run_modes.skip import process_outputs as get_skip_mode_outputs
 from cylc.flow.task_action_timer import (
@@ -1916,7 +1917,9 @@ def _get_task_proxy_db_outputs(
         self._load_historical_outputs(itask)
         return itask
 
-    def _standardise_prereqs(self, prereqs: 'Iterable[str]') -> 'Set[Tokens]':
+    def _standardise_prereqs(
+            self, prereqs: 'Iterable[str]'
+    ) -> 'Set[PrereqTuple]':
         """Extract task prerequistes from user input and standardise.
 
         Weed out any xtrigger prerequisites.
@@ -1953,7 +1956,7 @@ def _standardise_prereqs(self, prereqs: 'Iterable[str]') -> 'Set[Tokens]':
                 LOG.warning(
                     f'Invalid prerequisite cycle point:\n{exc.args[0]}')
             else:
-                _prereqs.add(pre.duplicate(task_sel=msg, cycle=cycle))
+                _prereqs.add(PrereqTuple(str(cycle), str(pre['task']), msg))
         return _prereqs
 
     def _standardise_outputs(
@@ -1975,7 +1978,7 @@ def _standardise_outputs(
 
     def _get_prereq_params(
         self, prereqs: 'Iterable[str]', tdef: 'TaskDef', point: 'PointBase'
-    ) -> 'Tuple[bool, Set[Tokens], Dict[str, bool]]':
+    ) -> 'Tuple[bool, Set[PrereqTuple], Dict[str, bool]]':
         """Convert input prerequisites to Tokens of just the valid ones.
 
         And convert the (mutually exclusive) "['all']" shortcut to a bool.
@@ -2104,7 +2107,7 @@ def set_prereqs_and_outputs(
 
     def _get_valid_prereqs(
             self, prereqs: Iterable[str], tdef: 'TaskDef', point: 'PointBase'
-    ) -> 'Set[Tokens]':
+    ) -> 'Set[PrereqTuple]':
         """Validate CLI prerequisites and return associated task messages.
 
         To set prerequisites, the user gives trigger names, but we need the
@@ -2120,24 +2123,23 @@ def _get_valid_prereqs(
         """
         # Valid prerequisites as tokens (outputs as task messages).
         valid_pre = {
-            Tokens(f"{key.point}/{key.task}:{key.output}", relative=True)
+            PrereqTuple(key.point, key.task, key.output)
             for pre in tdef.get_prereqs(point)
             for key in pre.keys()
         }
 
-        # standardise, tokenise, and weed out xtrigger prerequisites
+        # standardise and weed out xtrigger prerequisites
         req_pre = self._standardise_prereqs(prereqs)
 
         for prereq in req_pre - valid_pre:
             # But log bad ones with triggers, not messages.
             trg = self.config.get_taskdef(
-                str(prereq["task"])
-            ).get_output(prereq["task_sel"])
+                str(prereq.task)
+            ).get_output(prereq.output)
             LOG.warning(
                 f'{point}/{tdef.name} does not depend on '
-                f'"{prereq["cycle"]}/{prereq["task"]}:{trg}"'
+                f'"{prereq.point}/{prereq.task}:{trg}"'
             )
-
         return valid_pre & req_pre
 
     def _get_valid_xtrigs(
@@ -2228,7 +2230,7 @@ def _set_outputs_itask(
     def _set_prereqs_itask(
         self,
         itask: 'TaskProxy',
-        prereqs: 'Iterable[Tokens]',
+        prereqs: 'Iterable[PrereqTuple]',
         xtrigs: 'Dict[str, bool]',
         set_all: bool
     ) -> None:
@@ -2237,9 +2239,12 @@ def _set_prereqs_itask(
         Designated flows should already be merged to the task proxy.
         """
         if set_all:
-            itask.state.set_prerequisites_all_satisfied()
+            # (task prerequisites, not xtriggers)
+            itask.force_satisfy_all()
         else:
-            itask.satisfy_me(prereqs, forced=True)
+            # task prerequisites
+            itask.force_satisfy(prereqs)
+            # xtriggers, including "all"
             self.xtrigger_mgr.force_satisfy(itask, xtrigs)
 
         if (
@@ -2253,7 +2258,7 @@ def _set_prereqs_tdef(
         self,
         point: 'PointBase',
         taskdef: 'TaskDef',
-        prereqs: 'Iterable[Tokens]',
+        prereqs: 'Iterable[PrereqTuple]',
         xtrigs: 'Dict[str, bool]',
         flow_nums: 'FlowNums',
         flow_wait: bool,

cylc/flow/task_proxy.py

@@ -569,20 +569,63 @@ def state_reset(
 
     def satisfy_me(
         self,
-        task_messages: 'Iterable[Tokens]',
+        outputs: 'Iterable[Tokens]',
         mode: Optional[RunMode] = RunMode.LIVE,
-        forced: bool = False,
-    ) -> 'Set[Tokens]':
-        """Try to satisfy my prerequisites with given output messages.
+        forced: bool = False,  # TODO forced no longer needed here
+    ) -> None:
+        """Try to satisfy my prerequisites with given task output messages.
+
+        Output format: "cycle/task:message"
 
-        The task output messages are of the form "cycle/task:message"
-        Log a warning for messages that I don't depend on.
+        """
+        for prereq in (
+            *self.state.prerequisites, *self.state.suicide_prerequisites
+        ):
+            prereq.satisfy_me(outputs, mode=mode, forced=forced)
 
-        Return a set of unmatched task messages.
+    def force_satisfy(self, prereqs: 'Iterable[PrereqTuple]') -> None:
+        """Force satisfy given task prerequisites.
+
+        Only called via "cylc set" command so no need to record run mode.
+
+        """
+        for prereq in self.state.prerequisites:
+            for pre, state in prereq.items():
+                # (PrereqTuple, False or "satisfied naturally" etc.)
+                if pre not in prereqs:
+                    continue
+                if not state:
+                    prereq[pre] = "force satisfied"
+                    LOG.info(
+                        f"[{self}] prerequisite satisfied (forced):"
+                        f" {pre.get_id(True)}"
+                    )
+                else:
+                    LOG.info(
+                        f"[{self}] prerequisite already satisfied:"
+                        f" {pre.get_id(True)}"
+                    )
+
+    def force_satisfy_all(self):
+        """Force satisfy all task prerequisites.
+
+        Only called via "cylc set" command so no need to record run mode.
 
         """
-        used = self.state.satisfy_me(task_messages, mode=mode, forced=forced)
-        return set(task_messages) - used
+        for prereq in self.state.prerequisites:
+            for pre, state in prereq.items():
+                # (PrereqTuple, False or "satisfied naturally" etc.)
+                if not state:
+                    prereq[pre] = "force satisfied"
+                    LOG.info(
+                        f"[{self}] prerequisite satisfied (forced):"
+                        f" {pre.get_id(True)}"
+                    )
+                else:
+                    LOG.info(
+                        f"[{self}] prerequisite already satisfied:"
+                        f" {pre.get_id(True)}"
+                    )
 
     def clock_expire(self) -> bool:
         """Return True if clock expire time is up, else False."""

cylc/flow/task_state.py

@@ -19,10 +19,8 @@
 from typing import (
     TYPE_CHECKING,
     Dict,
-    Iterable,
     List,
     Optional,
-    Set,
 )
 
 from cylc.flow.prerequisite import Prerequisite
@@ -40,9 +38,7 @@
 
 if TYPE_CHECKING:
     from cylc.flow.cycling import PointBase
-    from cylc.flow.id import Tokens
     from cylc.flow.prerequisite import PrereqTuple
-    from cylc.flow.run_modes import RunMode
     from cylc.flow.taskdef import TaskDef
 
 
@@ -321,23 +317,6 @@ def __call__(
             )
         )
 
-    def satisfy_me(
-        self,
-        outputs: Iterable['Tokens'],
-        mode: 'Optional[RunMode]',
-        forced: bool = False,
-    ) -> Set['Tokens']:
-        """Try to satisfy my prerequisites with given outputs.
-
-        Return which outputs I actually depend on.
-        """
-        valid: Set[Tokens] = set()
-        for prereq in (*self.prerequisites, *self.suicide_prerequisites):
-            valid.update(
-                prereq.satisfy_me(outputs, mode=mode, forced=forced)
-            )
-        return valid
-
     def xtriggers_all_satisfied(self):
         """Return True if all xtriggers are satisfied."""
         return all(self.xtriggers.values())