WIP: workflow progress bar.

Author: jmchilton

planemo/commands/cmd_workflow_test_on_invocation.py

@@ -15,9 +15,7 @@
 
 @click.command("workflow_test_on_invocation")
 @options.optional_tools_arg(multiple=False, allow_uris=False, metavar="TEST.YML")
-@options.required_invocation_id_arg()
-@options.galaxy_url_option(required=True)
-@options.galaxy_user_key_option(required=True)
+@options.invocation_target_options()
 @options.test_index_option()
 @options.test_output_options()
 @command_function

planemo/commands/cmd_workflow_track.py

@@ -0,0 +1,49 @@
+"""Module describing the planemo ``workflow_track`` command."""
+
+import click
+
+from planemo import options
+from planemo.cli import command_function
+from planemo.galaxy.workflow_progress import WorkflowProgress
+from planemo.engine.factory import engine_context
+
+
+@click.command("workflow_track")
+@options.invocation_target_options()
+@command_function
+def cli(ctx, invocation_id, **kwds):
+    """Run defined tests against existing workflow invocation."""
+    with WorkflowProgress() as workflow_progress:
+        workflow_progress.add_bars()
+        import time
+        time.sleep(1)
+        new_step = {"state": "new"}
+        scheduled_step = {"state": "scheduled"}
+        new_steps = [new_step, new_step, new_step]
+        one_scheduled_steps = [scheduled_step, new_step, new_step]
+        two_scheduled_steps = [scheduled_step, scheduled_step, new_step]
+        all_scheduled_steps = [scheduled_step, scheduled_step, scheduled_step]
+        state_pairs = [
+            ({"state": "new"}, {}),
+            ({"state": "ready", "steps": new_steps}, {}),
+            ({"state": "ready", "steps": one_scheduled_steps}, {"states": {"new": 1}}),
+            ({"state": "ready", "steps": two_scheduled_steps}, {"states": {"new": 2}}),
+            ({"state": "ready", "steps": two_scheduled_steps}, {"states": {"new": 1, "running": 1}}),
+            ({"state": "ready", "steps": two_scheduled_steps}, {"states": {"new": 1, "ok": 1}}),
+            ({"state": "ready", "steps": two_scheduled_steps}, {"states": {"ok": 2}}),
+            ({"state": "scheduled", "steps": all_scheduled_steps}, {"states": {"ok": 2, "new": 3}}),
+            ({"state": "scheduled", "steps": all_scheduled_steps}, {"states": {"ok": 2, "running": 1, "new": 2}}),
+            ({"state": "scheduled", "steps": all_scheduled_steps}, {"states": {"ok": 3, "running": 1, "new": 1}}),
+            ({"state": "scheduled", "steps": all_scheduled_steps}, {"states": {"ok": 4, "running": 1}}),
+            ({"state": "scheduled", "steps": all_scheduled_steps}, {"states": {"ok": 5}}),
+        ]
+        for (invocation, job_states_summary) in state_pairs:
+            workflow_progress.handle_invocation(invocation, job_states_summary)
+            time.sleep(1)
+
+    with engine_context(ctx, engine="external_galaxy", **kwds) as engine, engine.ensure_runnables_served([]) as config:
+        user_gi = config.user_gi
+        invocation = user_gi.invocations.show_invocation(invocation_id)
+        # https://stackoverflow.com/questions/23113494/double-progress-bar-in-python
+
+    ctx.exit(0)

planemo/galaxy/workflow_progress.py

@@ -0,0 +1,133 @@
+from typing import (
+    Dict,
+    Optional,
+)
+
+from rich.console import Console
+from rich.progress import (
+    BarColumn,
+    Progress,
+    TextColumn,
+    TaskProgressColumn,
+)
+from rich.theme import Theme
+
+
+class WorkflowProgress(Progress):
+    invocation_state: str = "new"
+    step_count: Optional[int] = None
+    job_count: int = 0
+    steps_color: str = "cyan"
+    jobs_color: str = "cyan"
+    step_states: Dict = {}
+
+    def __init__(self):
+        custom_theme = Theme({
+            "bar.complete": "green",
+        })
+        console = Console(theme=custom_theme)
+        super().__init__(
+            TextColumn("[progress.description]{task.description}"),
+            BarColumn(),
+            TaskProgressColumn(),
+            TextColumn(text_format="{task.fields[status]}"),
+            console=console,
+        )
+
+    def handle_invocation(self, invocation: Dict, job_state_summary: Dict):
+        self.invocation_state = invocation.get("state") or "new"
+        self.step_count = len(invocation.get("steps") or []) or None
+        self.step_states = step_states(invocation)
+
+        steps_completed = None
+
+        steps_status = ""
+        if self.step_count is None:
+            steps_status = "Loading steps."
+            self.steps_color = "cyan"
+        elif self.invocation_state == "cancelled":
+            steps_status = "Invocation cancelled"
+            self.steps_color = "red"
+        elif self.invocation_state == "failed":
+            steps_status = "Invocation failed"
+            self.steps_color = "red"
+        else:
+            num_scheduled = self.step_states.get("scheduled") or 0
+            if num_scheduled > 0:
+                self.steps_color = "green"
+            else:
+                self.steps_color = "cyan"
+            steps_completed = num_scheduled
+            steps_status = f"{num_scheduled}/{self.step_count} scheduled"
+
+        jobs_status = ""
+        self.job_count = job_count(job_state_summary)
+        num_errors = error_count(job_state_summary)
+        num_ok = ok_count(job_state_summary)
+        jobs_completed = num_ok + num_errors
+        jobs_total = self.job_count
+        if num_errors > 0:
+            self.jobs_color = "red"
+        elif self.job_count > 0:
+            self.jobs_color = "green"
+        else:
+            self.jobs_color = "cyan"
+            jobs_completed = None
+            jobs_total = None
+        if self.job_count > 0:
+            jobs_status = f"{jobs_completed}/{self.job_count} terminal"
+        self.update(self._steps_task, total=self.step_count, completed=steps_completed, description=f"[{self.steps_color}]Steps...", status=steps_status)
+        self.update(self._jobs_task, total=jobs_total, completed=jobs_completed, description=f"[{self.jobs_color}]Jobs...", status=jobs_status)
+
+    def add_bars(self):
+        self._steps_task = self.add_task("[green]Steps...", status="")
+        self._jobs_task = self.add_task("[green]Jobs...", status="")
+
+
+# converted from Galaxy TypeScript (see util.ts next to WorkflowInvocationState.vue)
+def count_states(job_summary: Optional[Dict], query_states: list[str]) -> int:
+    count = 0
+    states = job_summary.get("states") if job_summary else None
+    if states:
+        for state in query_states:
+            count += states.get(state, 0)
+    return count
+
+
+def job_count(job_summary: Optional[Dict]) -> int:
+    states = job_summary.get("states") if job_summary else None
+    count = 0
+    if states:
+        for state_count in states.values():
+            if state_count:
+                count += state_count
+    return count
+
+
+def step_states(invocation: Dict):
+    step_states = {}
+    steps = invocation.get("steps") or []
+    for step in steps:
+        if not step:
+            continue
+        step_state = step.get("state") or "unknown"
+        if step_state not in step_states:
+            step_states[step_state] = 0
+        step_states[step_state] += 1
+
+    return step_states
+
+
+def ok_count(job_summary: Dict) -> int:
+    return count_states(job_summary, ["ok", "skipped"])
+
+
+ERROR_STATES = ["error", "deleted"]
+
+
+def error_count(job_summary: Dict) -> int:
+    return count_states(job_summary, ERROR_STATES)
+
+
+def running_count(job_summary: Dict) -> int:
+    return count_states(job_summary, ["running"])

planemo/options.py

@@ -2103,6 +2103,14 @@ def mulled_action_option():
     )
 
 
+def invocation_target_options():
+    return _compose(
+        required_invocation_id_arg(),
+        galaxy_url_option(required=True),
+        galaxy_user_key_option(required=True),
+    )
+
+
 def mulled_options():
     return _compose(
         mulled_conda_option(),