Controlling macro sequences

The sequence module provides macros for conditional execution and retries: cond, select, retry and retry_continue. The failure macro can be used to test error handling. These macros help compose procedures using existing macros as hooks.

Note

These flow-control macros are experimental and may change in future releases. They are separate from the deprecated sequence macro in the same module, which accepts an XML string rather than a file path.

Attaching the macros to execute

The flow-control macros execute the hooks attached to their hook places. Calling them without hooks does not perform the intended procedure. In the Sequencer GUI, add child macros as hooks and select their hook places. In Python macros, use createMacro(), createExecMacroHook() and appendHook(). For example, the following macro runs one of two delays:

from sardana.macroserver.macro import Macro

class conditional_delay(Macro):
    def run(self):
        condition, _ = self.createMacro("cond", "{sys/tg_test/1/ampli} == 0")
        condition.appendHook(
            (self.createExecMacroHook(["sleep", "1"]), ["on-true"])
        )
        condition.appendHook(
            (self.createExecMacroHook(["sleep", "2"]), ["on-false"])
        )
        self.runMacro(condition)

Replace sys/tg_test/1/ampli with an attribute available in your system. For more details about attaching hooks programmatically, see Adding hooks support.

Conditional execution with cond

cond expression evaluates a Taurus evaluation expression at execution time and converts its result to a boolean. Pass the expression without the eval: prefix; the macro adds it internally.

  • If the result is true, all hooks at on-true run in order.

  • Otherwise, all hooks at on-false run in order.

For example, {sys/tg_test/1/ampli} == 0 selects on-true when the attribute is zero. Either branch may be empty. If neither branch has hooks, cond emits a warning and executes no child macros.

Selecting a branch with select

select expression evaluates a Taurus expression, converts the result to an integer and uses it as an index into the hooks attached to body. Index zero selects the first hook, index one the second, and so on. Only the selected hook runs.

If the index is out of range, all hooks attached to default run in order. If no default hooks exist, the macro raises a RuntimeError.

For example, a selection with three possible delays can be built as follows:

from sardana.macroserver.macro import Macro

class selected_delay(Macro):
    def run(self):
        selection, _ = self.createMacro("select", "{sys/tg_test/1/ampli}")
        for delay in (0, 1, 2):
            selection.appendHook(
                (self.createExecMacroHook(["sleep", str(delay)]), ["body"])
            )
        selection.appendHook(
            (self.createExecMacroHook(["sleep", "10"]), ["default"])
        )
        self.runMacro(selection)

An attribute value of two selects sleep 2. A value of three selects the default, sleep 10.

Note

The current implementation uses Python list indexing, so negative indices can also select body hooks: -1 selects the last hook. They only select the default when they are outside the list’s range. Use integer-valued expressions to avoid truncation by the integer conversion.

Retrying a procedure

retry times sleep executes all hooks at body in order. If any of them raises an exception, it retries the whole body from the beginning, including any hooks that already succeeded. A successful attempt ends the retry loop.

The parameters are:

  • times: number of retries after the initial attempt; defaults to one. Zero won’t be very useful as it will immediately fail the whole macro if any hooked macro fails.

  • sleep: delay in seconds between failed attempts; defaults to zero.

For example, retry 3 5 allows failing up to three times, with a five-second delay between them. Hooks at on-failure run after the final failed attempt. The macro then raises a RuntimeError chained from the body exception. An InterruptException is propagated without retrying.

Note

The current implementation calls pre-retry hooks before the third and subsequent attempts, but not before the second attempt. Exceptions raised by pre-retry or on-failure hooks are not retried by this macro.

Attach only procedures that can safely be repeated: retrying does not undo motor movements, acquisitions or other changes made by previous attempts. But you may be able to do such “reset” operations by hooking macros to the pre-retry place.

Continuing after failed retries

retry_continue accepts the same parameters and hook places as retry. After the final failed body attempt, it runs on-failure hooks and logs a warning instead of raising the final RuntimeError. This lets a surrounding sequence continue even when the body never succeeds. Interruptions and errors in the auxiliary hooks still propagate.

Testing failures

failure takes no parameters and always raises RuntimeError("Failure!"). It is useful for checking retry counts and failure hooks without depending on a device error. For example:

from sardana.macroserver.macro import Macro

class retry_example(Macro):
    def run(self):
        attempts, _ = self.createMacro("retry_continue", 2, 0)
        attempts.appendHook(
            (self.createExecMacroHook(["failure"]), ["body"])
        )
        attempts.appendHook(
            (self.createExecMacroHook(["sleep", "1"]), ["on-failure"])
        )
        self.runMacro(attempts)
        self.output("Continuing after three failed attempts")

Here the body fails three times, the failure hook sleeps once, and the outer macro continues. Replacing retry_continue with retry makes the final failure propagate instead.