Integrate statecharts into your code

Sismic provides several ways to integrate executable statecharts into your Python source code. The simplest way is to directly embed the entire code in the statechart’s description. This was illustrated with the Elevator example in Include code in statecharts. Its code is part of the YAML file of the statechart, and interpreted by Sismic during the statechart’s execution.

In order to make a statechart communicate with the source code contained in the environment in which it is executed, there are basically two approaches:

  1. The statechart sends events to, or receives external events from the environment.

  2. The environment stores shared objects in the statechart’s initial context, and the statechart calls operations on these objects and/or accesses the variables contained in it.

Of course, one could also use a hybrid approach, combining ideas from the three approches above.

Running example

In this document, we will present the main differences between the two approaches, on the basis of a simple example of a Graphical User Interface (GUI) whose behaviour is defined by a statechart. All the source code and YAML files for this example, discussed in more detail below, is available in the docs/examples directory of Sismic’s repository.

The example represents a simple stopwatch, i.e., a timer than can be started, stopped and reset. It also provides a split time feature and a display of the elapsed time. A button-controlled GUI of such a stopwatch looks as follows (inactive buttons are greyed out):

_images/stopwatch_gui.png

Essentially, the stopwatch simply displays a value, representing the elapsed time (expressed in seconds), which is initially 0. By clicking on the start button the stopwatch starts running. When clicking on stop, the stopwatch stops running. By using split, the time being displayed is temporarily frozen, although the stopwatch continues to run. Clicking on unsplit while continue to display the actual elapsed time. reset will restart from 0, and quit will quit the stopwatch application.

The idea is that the buttons will trigger state changes and actions carried out by an underlying statechart. Taking abstraction of the concrete implementation, the statechart would essentially look as follows, with one main active state containing two parallel substates timer and display.

_images/stopwatch_overview.png

Controlling a statechart from within the environment

Let us illustrate how to control a statechart through source code that executes in the environment containing the statechart. The statechart’s behaviour is triggered by external events sent to it by the source code each time one of the buttons in the GUI is pressed. Conversely, the statechart itself can send events back to the source code to update its display.

This statechart looks as follows:

_images/stopwatch_with_logic.png

Here is the YAML file containing the textual description of this statechart:

statechart:
  name: Stopwatch
  description: |
    A simple stopwatch which support "start", "stop", "split", and "reset".
    These features are triggered respectively using "start", "stop", "split", and "reset".

    The stopwatch sends an "refresh" event each time the display is updated.
    The value to display is attached to the event under the key "time".

    The statechart is composed of two parallel regions:
     - A "timer" region which increments "elapsed_time" if timer is running
     - A "display" region that refreshes the display according to the actual time/lap time feature

  preamble: elapsed_time = 0
  root state:
    name: active
    parallel states:
      - name: timer
        initial: stopped
        transitions:
          - event: reset
            action: elapsed_time = 0
        states:
          - name: running
            transitions:
              - event: stop
                target: stopped
              - guard: after(1)
                target: running
                action: elapsed_time += 1
          - name: stopped
            transitions:
              - event: start
                target: running
      - name: display
        initial: actual time
        states:
          - name: actual time
            transitions:
              - guard: after(0.2)
                target: actual time
                action: |
                  send('refresh', time=elapsed_time)
              - event: split
                target: lap time
          - name: lap time
            transitions:
              - event: split
                target: actual time

We observe that the statechart contains an elapsed_time variable, that is updated every second while the stopwatch is in the running state. The statechart will modify its behaviour by receiving start, stop, reset and split events from its external environment. In parallel to this, every 100 milliseconds, the display state of the statechart sends a refresh event (parameterised by the time variable containing the elapsed_time value) back to its external environment. In the lap time state (reached through a split event) , this regular refreshing is stopped until a new split event is received.

The source code (shown below) that defines the GUI of the stopwatch, and that controls the statechart by sending it events, is implemented using the Tkinter library. Each button of the GUI is bound to a Python method in which the corresponding event is created and sent to the statechart. The statechart is bound to the source code by defining a new Interpreter that contains the parsed YAML specification, and using the bind() method. The event_handler passed to it allows the Python source code to receive events back from the statechart. In particular, the w_timer field of the GUI will be updated with a new value of the time whenever the statechart sends a refresh event. The run method, which is put in Tk’s mainloop, updates the internal clock of the interpreter and executes the interpreter.

import time
import tkinter as tk

from sismic.interpreter import Interpreter
from sismic.io import import_from_yaml


# The two following lines are NOT needed in a typical environment.
# These lines make sismic available in our testing environment
import sys
sys.path.append('../../..')




# Create a tiny GUI
class StopwatchApplication(tk.Frame):
    def __init__(self, master=None):
        super().__init__(master)

        # Initialize widgets
        self.create_widgets()

        # Create a Stopwatch interpreter
        with open('stopwatch.yaml') as f:
            statechart = import_from_yaml(f)
        self.interpreter = Interpreter(statechart)
        self.interpreter.clock.time = time.time()

        # Bind interpreter events to the GUI
        self.interpreter.bind(self.event_handler)

        # Run the interpreter
        self.run()

    def run(self):
        # Update internal clock and execute interpreter
        self.interpreter.clock.time = time.time()
        self.interpreter.execute()

        # Queue a call every 100ms on tk's mainloop
        self.after(100, self.run)

        # Update the widget that contains the list of active states.
        self.w_states['text'] = 'active states: ' + ', '.join(self.interpreter.configuration)

    def create_widgets(self):
        self.pack()

        # Add buttons
        self.w_btn_start = tk.Button(self, text='start', command=self._start)
        self.w_btn_stop = tk.Button(self, text='stop', command=self._stop)
        self.w_btn_split = tk.Button(self, text='split', command=self._split)
        self.w_btn_unsplit = tk.Button(self, text='unsplit', command=self._unsplit)
        self.w_btn_reset = tk.Button(self, text='reset', command=self._reset)
        self.w_btn_quit = tk.Button(self, text='quit', command=self._quit)

        # Initial button states
        self.w_btn_stop['state'] = tk.DISABLED
        self.w_btn_unsplit['state'] = tk.DISABLED

        # Pack
        self.w_btn_start.pack(side=tk.LEFT,)
        self.w_btn_stop.pack(side=tk.LEFT,)
        self.w_btn_split.pack(side=tk.LEFT,)
        self.w_btn_unsplit.pack(side=tk.LEFT,)
        self.w_btn_reset.pack(side=tk.LEFT,)
        self.w_btn_quit.pack(side=tk.LEFT,)

        # Active states label
        self.w_states = tk.Label(root)
        self.w_states.pack(side=tk.BOTTOM, fill=tk.X)

        # Timer label
        self.w_timer = tk.Label(root, font=("Helvetica", 16), pady=5)
        self.w_timer.pack(side=tk.BOTTOM, fill=tk.X)

    def event_handler(self, event):
        # Update text widget when timer value is updated
        if event.name == 'refresh':
            self.w_timer['text'] = event.time

    def _start(self):
        self.interpreter.queue('start')
        self.w_btn_start['state'] = tk.DISABLED
        self.w_btn_stop['state'] = tk.NORMAL

    def _stop(self):
        self.interpreter.queue('stop')
        self.w_btn_start['state'] = tk.NORMAL
        self.w_btn_stop['state'] = tk.DISABLED

    def _reset(self):
        self.interpreter.queue('reset')

    def _split(self):
        self.interpreter.queue('split')
        self.w_btn_split['state'] = tk.DISABLED
        self.w_btn_unsplit['state'] = tk.NORMAL

    def _unsplit(self):
        self.interpreter.queue('split')
        self.w_btn_split['state'] = tk.NORMAL
        self.w_btn_unsplit['state'] = tk.DISABLED

    def _quit(self):
        self.master.destroy()


if __name__ == '__main__':
    # Create GUI
    root = tk.Tk()
    root.wm_title('StopWatch')
    app = StopwatchApplication(master=root)

    app.mainloop()

Controlling the environment from within the statechart

In this second example, we basically reverse the idea: now the Python code that resides in the environment contains the logic (e.g., the elapsed_time variable), and this code is exposed to, and controlled by, a statechart that represents the main loop of the program and calls the necessary methods in the source code. These method calls are associated to actions on the statechart’s transitions. With this solution, the statechart is no longer a black box, since it needs to be aware of the source code, in particular the methods it needs to call in this code.

An example of the Python code that is controlled by the statechart is given below:

class Stopwatch:
    def __init__(self):
        self.elapsed_time = 0
        self.split_time = 0
        self.is_split = False
        self.running = False

    def start(self):
        # Start internal timer
        self.running = True

    def stop(self):
        # Stop internal timer
        self.running = False

    def reset(self):
        # Reset internal timer
        self.elapsed_time = 0

    def split(self):
        # Split time
        if not self.is_split:
            self.is_split = True
            self.split_time = self.elapsed_time

    def unsplit(self):
        # Unsplit time
        if self.is_split:
            self.is_split = False

    def display(self):
        # Return the value to display
        if self.is_split:
            return int(self.split_time)
        else:
            return int(self.elapsed_time)

    def update(self, delta):
        # Update internal timer of ``delta`` seconds
        if self.running:
            self.elapsed_time += delta

The statechart expects such a Stopwatch instance to be created and provided in its initial context. Recall that an Interpreter accepts an optional initial_context parameter. In this example, initial_context={'stopwatch': Stopwatch()}.

The statechart is simpler than in the previous example: one parallel region handles the running status of the stopwatch, and a second one handles its split features.

_images/stopwatch_with_object.png
statechart:
  name: Stopwatch
  description: |
    A simple stopwatch which support "start", "stop", "split", and "reset".
    These features are triggered respectively using "start", "stop", "split", and "reset".

    The stopwatch expects a "stopwatch" object in its initial context.
    This object should support the following methods: "start", "stop", "split", "reset", and "unsplit".
  root state:
    name: active
    parallel states:
      - name: timer
        initial: stopped
        transitions:
          - event: reset
            action: stopwatch.reset()
        states:
          - name: running
            transitions:
              - event: stop
                target: stopped
                action: stopwatch.stop()
          - name: stopped
            transitions:
              - event: start
                target: running
                action: stopwatch.start()
      - name: display
        initial: actual time
        states:
          - name: actual time
            transitions:
              - event: split
                target: lap time
                action: stopwatch.split()
          - name: lap time
            transitions:
              - event: split
                target: actual time
                action: stopwatch.unsplit()

The Python code of the GUI no longer needs to listen to the events sent by the interpreter. It should, of course, continue to send events (corresponding to button presses) to the statechart using send. The binding between the statechart and the GUI is now achieved differently, by simply passing the stopwatch object to the Interpreter as its initial_context.

import time
import tkinter as tk

from sismic.interpreter import Interpreter
from sismic.io import import_from_yaml
from stopwatch import Stopwatch


# The two following lines are NOT needed in a typical environment.
# These lines make sismic available in our testing environment
import sys
sys.path.append('../../..')




# Create a tiny GUI
class StopwatchApplication(tk.Frame):
    def __init__(self, master=None):
        super().__init__(master)

        # Initialize widgets
        self.create_widgets()

        # Create a Stopwatch interpreter
        with open('stopwatch_external.yaml') as f:
            statechart = import_from_yaml(f)

        # Create a stopwatch object and pass it to the interpreter
        self.stopwatch = Stopwatch()
        self.interpreter = Interpreter(statechart, initial_context={'stopwatch': self.stopwatch})
        self.interpreter.clock.start()
        
        # Run the interpreter
        self.run()

        # Update the stopwatch every 100ms
        self.after(100, self.update_stopwatch)

    def update_stopwatch(self):
        self.stopwatch.update(delta=0.1)
        self.after(100, self.update_stopwatch)

        # Update timer label
        self.w_timer['text'] = self.stopwatch.display()

    def run(self):
        # Queue a call every 100ms on tk's mainloop
        self.interpreter.execute()
        self.after(100, self.run)
        self.w_states['text'] = 'active states: ' + ', '.join(self.interpreter.configuration)

    def create_widgets(self):
        self.pack()

        # Add buttons
        self.w_btn_start = tk.Button(self, text='start', command=self._start)
        self.w_btn_stop = tk.Button(self, text='stop', command=self._stop)
        self.w_btn_split = tk.Button(self, text='split', command=self._split)
        self.w_btn_unsplit = tk.Button(self, text='unsplit', command=self._unsplit)
        self.w_btn_reset = tk.Button(self, text='reset', command=self._reset)
        self.w_btn_quit = tk.Button(self, text='quit', command=self._quit)

        # Initial button states
        self.w_btn_stop['state'] = tk.DISABLED
        self.w_btn_unsplit['state'] = tk.DISABLED

        # Pack
        self.w_btn_start.pack(side=tk.LEFT,)
        self.w_btn_stop.pack(side=tk.LEFT,)
        self.w_btn_split.pack(side=tk.LEFT,)
        self.w_btn_unsplit.pack(side=tk.LEFT,)
        self.w_btn_reset.pack(side=tk.LEFT,)
        self.w_btn_quit.pack(side=tk.LEFT,)

        # Active states label
        self.w_states = tk.Label(root)
        self.w_states.pack(side=tk.BOTTOM, fill=tk.X)

        # Timer label
        self.w_timer = tk.Label(root, font=("Helvetica", 16), pady=5)
        self.w_timer.pack(side=tk.BOTTOM, fill=tk.X)

    def _start(self):
        self.interpreter.queue('start')
        self.w_btn_start['state'] = tk.DISABLED
        self.w_btn_stop['state'] = tk.NORMAL

    def _stop(self):
        self.interpreter.queue('stop')
        self.w_btn_start['state'] = tk.NORMAL
        self.w_btn_stop['state'] = tk.DISABLED

    def _reset(self):
        self.interpreter.queue('reset')

    def _split(self):
        self.interpreter.queue('split')
        self.w_btn_split['state'] = tk.DISABLED
        self.w_btn_unsplit['state'] = tk.NORMAL

    def _unsplit(self):
        self.interpreter.queue('split')
        self.w_btn_split['state'] = tk.NORMAL
        self.w_btn_unsplit['state'] = tk.DISABLED

    def _quit(self):
        self.master.destroy()


if __name__ == '__main__':
    # Create GUI
    root = tk.Tk()
    root.wm_title('StopWatch (external)')
    app = StopwatchApplication(master=root)

    app.mainloop()