return task states from run_blockwise instead of just a boolean for…

… whether or not all blocks finished.

This allows finer grained views into how many succeeded vs failed vs skipped etc.

daisy/convenience.py

@@ -56,4 +56,5 @@ def run_blockwise(tasks, multiprocessing=True):
 def _run_blockwise(tasks, stop_event):
     server = Server(stop_event=stop_event)
     cl_monitor = CLMonitor(server)  # noqa
-    return server.run_blockwise(tasks)
+    task_states = server.run_blockwise(tasks)
+    return all(task_state.is_done() for task_state in task_states.values())

daisy/serial_server.py

@@ -1,5 +1,6 @@
 from .block import BlockStatus
 from .scheduler import Scheduler
+from .task_state import TaskState
 from .server_observer import ServerObservee
 import logging
 
@@ -10,7 +11,7 @@ class SerialServer(ServerObservee):
     def __init__(self):
         super().__init__()
 
-    def run_blockwise(self, tasks, scheduler=None):
+    def run_blockwise(self, tasks, scheduler=None) -> dict[str, TaskState]:
         if scheduler is None:
             scheduler = Scheduler(tasks)
         else:
@@ -60,6 +61,5 @@ def run_blockwise(self, tasks, scheduler=None):
                     del process_funcs[block.task_id]
 
             if len(process_funcs) == 0:
-                return True
-
-        self.notify_server_exit()
+                self.notify_server_exit()
+                return scheduler.task_states

daisy/server.py

@@ -13,6 +13,7 @@
     UnexpectedMessage,
 )
 from .scheduler import Scheduler
+from .task_state import TaskState
 from .server_observer import ServerObservee
 from .task_worker_pools import TaskWorkerPools
 from .tcp import TCPServer
@@ -40,7 +41,7 @@ def __init__(self, stop_event=None):
 
         logger.debug("Started server listening at %s:%s", self.hostname, self.port)
 
-    def run_blockwise(self, tasks, scheduler=None):
+    def run_blockwise(self, tasks, scheduler=None) -> dict[str, TaskState]:
 
         if scheduler is None:
             self.scheduler = Scheduler(tasks)
@@ -68,7 +69,7 @@ def run_blockwise(self, tasks, scheduler=None):
             logger.debug("TCP streams closed.")
             self.notify_server_exit()
 
-        return True if self.all_done else False
+        return self.scheduler.task_states
 
     def _event_loop(self):
         last_time = time()