Added Pause and Rewind

malcolm/controllers/builtin/managercontroller.py

@@ -3,7 +3,7 @@
 import numpy as np
 
 from malcolm.core import RunnableDeviceStateMachine, REQUIRED, method_returns, \
-    method_only_in, method_takes, ElementMap, Attribute
+    method_only_in, method_takes, ElementMap, Attribute, Task
 from malcolm.core.vmetas import PointGeneratorMeta, StringArrayMeta, NumberMeta
 from malcolm.controllers.builtin.defaultcontroller import DefaultController
 
@@ -23,10 +23,11 @@ class ManagerController(DefaultController):
     """RunnableDevice implementer that also exposes GUI for child parts"""
     # default attributes
     totalSteps = None
-    exposure = None
-    generator = None
+    # For storing iterator
     iterator = None
     points = None
+    # Params passed to configure()
+    configure_params = None
 
     def get_point(self, num):
         npoints = len(self.points)
@@ -76,34 +77,59 @@ def do_validate(self, params):
     def configure(self, params):
         try:
             self.transition(sm.CONFIGURING, "Configuring", create_tasks=True)
-            self.exposure = params.exposure
-            self.generator = params.generator
+            self.configure_params = params
             self.points = []
             self.iterator = params.generator.iterator()
             steps = np.prod(params.generator.index_dims)
             self.totalSteps.set_value(steps)
-            self.do_configure(params)
+            self.block["completedSteps"].set_value(0)
+            self.do_configure(self.part_tasks, params)
             self.transition(sm.READY, "Done configuring")
         except Exception as e:  # pylint:disable=broad-except
             self.log_exception("Fault occurred while Configuring")
             self.transition(sm.FAULT, str(e))
             raise
 
-    def do_configure(self, params):
+    def do_configure(self, part_tasks, params, start_index=0):
         raise NotImplementedError()
 
     @method_only_in(sm.READY)
     def run(self):
         try:
             self.transition(sm.PRERUN, "Preparing for run", create_tasks=True)
-            next_state = self.do_run()
+            next_state = self._call_do_run()
             self.transition(next_state, "Run finished")
+        except StopIteration:
+            self.log_warning("Run aborted")
+            raise
         except Exception as e:  # pylint:disable=broad-except
             self.log_exception("Fault occurred while Running")
             self.transition(sm.FAULT, str(e))
             raise
 
-    def do_run(self):
+    def _call_do_run(self):
+        try:
+            return self.do_run(self.part_tasks)
+        except StopIteration:
+            # Work out if it was an abort or pause
+            with self.lock:
+                state = self.state.value
+            self.log_debug("Do run got StopIteration from %s", state)
+            if state in (sm.REWINDING, sm.PAUSED):
+                # Wait to be restarted
+                self.log_debug("Waiting for PreRun")
+                task = Task("StateWaiter", self.process)
+                futures = task.when_matches(self.state, sm.PRERUN, [
+                    sm.DISABLING, sm.ABORTING, sm.FAULT])
+                task.wait_all(futures)
+                # Restart it
+                return self._call_do_run()
+            else:
+                # just drop out
+                self.log_debug("We were aborted")
+                raise
+
+    def do_run(self, part_tasks):
         raise NotImplementedError()
 
     @method_only_in(sm.IDLE, sm.CONFIGURING, sm.READY, sm.PRERUN, sm.RUNNING,
@@ -121,6 +147,43 @@ def abort(self):
     def do_abort(self):
         self.stop_and_wait_part_tasks()
 
+    @method_only_in(sm.PRERUN, sm.RUNNING)
+    def pause(self):
+        try:
+            self.transition(sm.REWINDING, "Rewinding")
+            current_index = self.block.completedSteps
+            self.do_abort()
+            self.part_tasks = self.create_part_tasks()
+            self.do_configure(
+                self.part_tasks, self.configure_params, current_index)
+            self.transition(sm.PAUSED, "Pause finished")
+        except Exception as e:  # pylint:disable=broad-except
+            self.log_exception("Fault occurred while Pausing")
+            self.transition(sm.FAULT, str(e))
+            raise
+
+    @method_only_in(sm.READY, sm.PAUSED)
+    @method_takes("steps", NumberMeta(
+        "uint32", "Number of steps to rewind"), REQUIRED)
+    def rewind(self, params):
+        try:
+            self.transition(sm.REWINDING, "Rewinding")
+            current_index = self.block.completedSteps
+            requested_index = current_index - params.steps
+            assert requested_index >= 0, \
+                "Cannot retrace to before the start of the scan"
+            self.block["completedSteps"].set_value(requested_index)
+            self.do_configure(
+                self.part_tasks, self.configure_params, requested_index)
+            self.transition(sm.PAUSED, "Rewind finished")
+        except Exception as e:  # pylint:disable=broad-except
+            self.log_exception("Fault occurred while Rewinding")
+            self.transition(sm.FAULT, str(e))
+            raise
+
+    @method_only_in(sm.PAUSED)
+    def resume(self):
+        self.transition(sm.PRERUN, "Resuming run")
 
 
 

malcolm/controllers/pmac/pmactrajectorycontroller.py

@@ -31,7 +31,7 @@ class PMACTrajectoryController(ManagerController):
     axis_mapping = None
     cs_port = None
     points_built = 0
-    current_steps = None
+    completed_steps = None
 
     def do_validate(self, params):
         self.get_cs_port(params.axes_to_move)
@@ -63,22 +63,22 @@ def get_cs_port(self, axes_to_move):
             "CS axis defs %s have more that one raw motor attached" % overlap
         return cs_ports.pop(), axis_mapping
 
-    def do_configure(self, params):
+    def do_configure(self, part_tasks, params, start_index=0):
         self.cs_port, self.axis_mapping = self.get_cs_port(params.axes_to_move)
-        self.build_start_profile()
-        self.run_hook(self.RunProfile, self.part_tasks)
-        self.points_built, self.current_steps = self.build_generator_profile()
+        self.build_start_profile(part_tasks, start_index)
+        self.run_hook(self.RunProfile, part_tasks)
+        self.points_built, self.completed_steps = self.build_generator_profile(
+            part_tasks, start_index)
 
-    def do_run(self):
+    def do_run(self, part_tasks):
         self.transition(sm.RUNNING, "Waiting for scan to complete")
-        self.run_hook(self.RunProfile, self.part_tasks,
-                      current_steps=self.current_steps)
+        self.run_hook(self.RunProfile, part_tasks,
+                      completed_steps=self.completed_steps)
         more_to_do = self.points_built < self.totalSteps.value - 1
         if more_to_do:
             self.transition(sm.POSTRUN, "Building next stage")
-            self.points_built, self.current_steps = \
-                self.build_generator_profile(self.points_built,
-                                             reset_current_step=False)
+            self.points_built, self.completed_steps = \
+                self.build_generator_profile(part_tasks, self.points_built)
             return self.stateMachine.READY
         else:
             self.transition(sm.POSTRUN, "Finishing run")
@@ -109,11 +109,11 @@ def calculate_acceleration_time(self):
                 acceleration_time, part.get_acceleration_time())
         return acceleration_time
 
-    def build_start_profile(self):
+    def build_start_profile(self, part_tasks, start_index):
         """Move to the run up position ready to start the scan"""
         acceleration_time = self.calculate_acceleration_time()
-        fraction = acceleration_time / self.exposure
-        first_point = self.get_point(0)
+        fraction = acceleration_time / self.configure_params.exposure
+        first_point = self.get_point(start_index)
         trajectory = {}
         move_time = 0
 
@@ -125,17 +125,16 @@ def build_start_profile(self):
 
         time_array = [move_time]
         velocity_mode = [ZERO]
-        self.build_profile(time_array, velocity_mode, trajectory)
+        self.build_profile(part_tasks, time_array, velocity_mode, trajectory)
 
-    def build_profile(self, time_array, velocity_mode, trajectory,
-                      reset_current_step=False):
-        """Build profile using self.part_tasks
+    def build_profile(self, part_tasks, time_array, velocity_mode, trajectory):
+        """Build profile using part_tasks
 
         Args:
             time_array (list): List of times in ms
             velocity_mode (list): List of velocity modes like PREV_TO_NEXT
             trajectory (dict): {axis_name: [positions in EGUs]}
-            reset_current_step (bool): Whether to reset currentStep attribute
+            part_tasks (dict): {part: task}
         """
         profile = Table(profile_table)
         profile.time = time_array
@@ -155,20 +154,19 @@ def build_profile(self, time_array, velocity_mode, trajectory,
             if cs_axis not in use:
                 profile[cs_axis] = [0] * len(time_array)
 
-        self.run_hook(self.BuildProfile, self.part_tasks, profile=profile,
-                      use=use, resolutions=resolutions, offsets=offsets,
-                      reset_current_step=reset_current_step)
+        self.run_hook(self.BuildProfile, part_tasks, profile=profile,
+                      use=use, resolutions=resolutions, offsets=offsets)
 
-    def build_generator_profile(self, start=0, reset_current_step=True):
+    def build_generator_profile(self, part_tasks, start_index=0):
         acceleration_time = self.calculate_acceleration_time()
-        fraction = acceleration_time / self.exposure
+        fraction = acceleration_time / self.configure_params.exposure
         trajectory = {}
         time_array = []
         velocity_mode = []
-        current_steps = []
+        completed_steps = []
         last_point = None
 
-        for i in range(start, self.totalSteps.value):
+        for i in range(start_index, self.totalSteps.value):
             point = self.get_point(i)
 
             # Check that none of the external motors need moving
@@ -188,14 +186,14 @@ def build_generator_profile(self, start=0, reset_current_step=True):
                     velocity_mode[-1] = PREV_TO_CURRENT
                 velocity_mode.append(CURRENT_TO_NEXT)
                 time_array.append(lower_move_time)
-                current_steps.append(i)
+                completed_steps.append(i)
 
             # Add position and upper bound
             for x in range(2):
-                time_array.append(self.exposure / 2.0)
+                time_array.append(self.configure_params.exposure / 2.0)
                 velocity_mode.append(PREV_TO_NEXT)
-            current_steps.append(i)
-            current_steps.append(i + 1)
+            completed_steps.append(i)
+            completed_steps.append(i + 1)
 
             # Add the axis positions
             for axis_name, cs_def in self.axis_mapping.items():
@@ -206,7 +204,7 @@ def build_generator_profile(self, start=0, reset_current_step=True):
                 positions.append(point.positions[axis_name])
                 positions.append(point.upper[axis_name])
             last_point = point
-
+        a = 1
         # Add a tail off position
         for axis_name, tail_off in \
                 self.run_up_positions(last_point, fraction).items():
@@ -216,13 +214,12 @@ def build_generator_profile(self, start=0, reset_current_step=True):
         time_array.append(acceleration_time)
         if i + 1 < self.totalSteps.value:
             # We broke, so don't add i + 1 to current step
-            current_steps.append(i)
+            completed_steps.append(i)
         else:
-            current_steps.append(i + 1)
+            completed_steps.append(i + 1)
 
-        self.build_profile(time_array, velocity_mode, trajectory,
-                           reset_current_step=reset_current_step)
-        return i, current_steps
+        self.build_profile(part_tasks, time_array, velocity_mode, trajectory)
+        return i, completed_steps
 
     def need_lower_move_time(self, last_point, point):
         # First point needs to insert lower bound point
@@ -236,7 +233,7 @@ def need_lower_move_time(self, last_point, point):
         return lower_move_time
 
     def external_axis_has_moved(self, last_point, point):
-        for axis_name in self.generator.position_units:
+        for axis_name in self.configure_params.generator.position_units:
             if axis_name not in self.axis_mapping:
                 # Check it hasn't needed to move
                 if point.positions[axis_name] != \

malcolm/core/controller.py

@@ -199,12 +199,12 @@ def _do_transition(self, state, message):
             if isinstance(child, MethodMeta):
                 method = child
                 writeable = self.methods_writeable[state][method]
-                self.log_debug("Setting %s %s to writeable %s", name, method, writeable)
                 self.add_change(changes, method, "writeable", writeable)
                 for ename in method.takes.elements:
                     meta = method.takes.elements[ename]
                     self.add_change(changes, meta, "writeable", writeable)
 
+        self.log_debug("Transitioning to %s", state)
         self.block.apply_changes(*changes)
 
     def register_method_writeable(self, method, states):
@@ -263,6 +263,7 @@ def _gather_task_return_value(func, task):
 
         for func, task in func_tasks.items():
             task.define_spawn_function(_gather_task_return_value, func, task)
+            self.log_debug("Starting task %r", task)
             task.start()
 
         # Create the reverse dictionary so we know where to store the results
@@ -279,16 +280,22 @@ def wait_hook(self, hook_queue, func_tasks, task_part_names):
         while func_tasks:
             func, ret = hook_queue.get()
             task = func_tasks.pop(func)
-            # Need to wait on it to clear spawned
-            task.wait()
             part_name = task_part_names[task]
             return_dict[part_name] = ret
+
             if isinstance(ret, Exception):
                 # Stop all other tasks
                 for task in func_tasks.values():
                     task.stop()
                 for task in func_tasks.values():
                     task.wait()
+
+            # If we got a StopIteration, someone asked us to stop, so
+            # don't wait, otherwise make sure we finished
+            if not isinstance(ret, StopIteration):
+                task.wait()
+
+            if isinstance(ret, Exception):
                 raise ret
 
         return return_dict

malcolm/core/spawnable.py

@@ -25,6 +25,7 @@ def start(self, process=None):
         if process is None:
             process = self.process
         self._initialize()
+        self._spawned = []
         for (func, args, _) in self._spawn_functions:
             assert func is not None, "Spawned function is None"
             self._spawned.append(process.spawn(func, *args))
@@ -41,7 +42,7 @@ def wait(self, timeout=None):
         self._initialize()
         for spawned in self._spawned:
             spawned.wait(timeout=timeout)
-        self._spawned = []
+            assert spawned.ready(), "Spawned %r still running" % spawned
 
     def add_spawn_function(self, func, stop_func=None, *args):
         """Register functions to be triggered by self.start and self.stop