Pyfsm

pyfsm

Module for the management, representation, and analysis of finite state machines (FSM).

Features

Introduction

This module provides classes and functions to create, manage, and analyze finite state machines (FSM). It offers tools to represent FSMs, evaluate their structure, and identify important properties such as state accessibility and cycle detection. I often code state machines to drive, test hardware, and make measurement tools. So, tired about verbosity I decide to make my own tool to describe FSMs in very direct fashion. Then I decided to add a tool to visualize the states on real time.

Description

The FSM is represented by square matrix form where the matrix is an adjacency matrix that represents the directed graph of the deterministic automata. Here, the rows represent the current state and the columns the next state; each element $a(i,j) = \delta_k$ represents a transition condition which can be 1 or 0, (True or False), so in this case i-th state goes to j-th state if k-th condition is true.

This matrix represents the following FSM:

In order to describe a FSM you need a 3-element Tuple: $s_{origin}\to s_{destiny}:t_{condition}$ where $s_{k}$ are the states and $t_k$ transition conditions: functions or expressions to be evaluated if theu are true or false.

To describe correctly FSM in deterministic case, the following conditions must be fullfilled:

  1. A transition condition from state $s_i$ to $s_j$ is unique
  2. Given a state $s_i$, whose outgoing flow branches into several paths toward the possible states $s_{k},s_{r},s_{t},...,s_{v}$ according to the associated transitions $\delta_{i,k},\delta_{i,r},\delta_{i,t},...,\delta_{i,v}$, these transitions must satisfy the condition of being mutually exclusive. That is, only one of them must be true in the s i-th state.
    $$\bigcap^{\infty} \delta_{i,m} = 0$$
    $$s := s_{k} | s_{i} \cap \delta_{i,k}$$
  3. Transition condition $s_{origin}\to s_{destiny}:t_{condition}$ must be unique.

State Machine Actions

On FSM there are 4 types of actions: 1. On enter: When FSM enters new state.
1. On state: When FSM is in particular state and step() cycle is called.
1. On exit: When FSM exits from current state.
1. On transition: When a particular transition condition is met.

When two or more conditions are met, the order is the following: 1. On State action callback is called on current state 1. On Transition action callback is called to current state associated transition condition which is True 1. On Exit action callback is called on current state 1. On Enter action callback is called on new state

Actions order transtitioning from state A to B

See example below:

if __name__ == "__main__": 
    f = fsm()
    # Define transitions
    f.add_transition('A => B : t0')
    f.add_transition('B => C : t1')
    f.add_transition('C => D : t2')
    f.add_transition('D => A : t3')

    # Adds condition
    f.add_condition('t0', 'a%10 == 0')
    f.add_condition('t1', 'a%10 == 0')
    f.add_condition('t2', 'a%10 == 0')
    f.add_condition('t3', 'a%10 == 0')

    # ******************** Define Actions ********************

    # On entry (to state) actions 
    f.add_action_on_entry('A', onEnter_A)
    f.add_action_on_entry('B', onEnter_B)
    f.add_action_on_entry('C', onEnter_C)
    f.add_action_on_entry('D', onEnter_D)

    # On state actions 
    f.add_action_on_state('A', onState_A)
    f.add_action_on_state('B', onState_B)
    f.add_action_on_state('C', onState_C)
    f.add_action_on_state('D', onState_D)

    # On exit (from state) actions
    f.add_action_on_exit('A', onExit_A)
    f.add_action_on_exit('B', onExit_B)
    f.add_action_on_exit('C', onExit_C)
    f.add_action_on_exit('D', onExit_D)

    # On transition actions
    f.add_action_on_transition('t0', onTransition_t0)
    f.add_action_on_transition('t1', onTransition_t1)
    f.add_action_on_transition('t2', onTransition_t2)
    f.add_action_on_transition('t3', onTransition_t3)

    # ********************************************************

    f.compile() # Compile

    a = 0

    # Execute 130 times
    for j in range(130): 
        f.step()
        if (j > 100) and (len(f.true_transitions) != 0):
            break
        a += 1

Dead states detection

Dead states are non-reachable states or states described on FSM that are not reachable because all entries on a given column related to this state are zero.
To ensure good FSM description a static check for dead states can be performed through accesibility matrix $R$.

The accessibility matrix (also known as the reachability matrix) is a matrix that shows which nodes in a graph can reach which other nodes, either directly or through a path of one or more steps. If you start with an adjacency matrix $M$, which tells you which nodes are directly connected, the accessibility matrix $R$ tells you whether there is any path at all from node $i$ to node $j$, regardless of how many steps it takes. $$R = \bigwedge_{k=1}^{N} M^k$$ I prefer to perfom matrix summation instead OR operation, first because all non zero entries are the reachability between states like boolean matrix, and second, this will give us the number of paths of any length (1 to N), from state $i$ to state $j$. So: $$R = \sum_{k=1}^{N} M^k$$ - $R_{i,j}$ show us how many paths of length 1 to N exists from state $i$ to state $j$. - If $R_{i,j}$ = 0 there is no path between i-th and j-th states. - If $R_{i,j}$ > 0 there is one or more paths between i-th and j-th states.

For dead states detection, please, refer to get_allPaths() method.

Loops or cycle detection

Sometimes, state machines are correctly coded in theoretical terms. However, when they interact with the physical world, unforeseen conditions may arise. Even if the code is correct, the system can enter a limit cycle that may affect an automated process. For example, let's suppose we have implemented an FSM to characterize and measure battery charging curves. Suppose that full charge detection is based on voltage measurement rather than a coulomb-counting gauge. If, for some reason (battery degradation, a fault in the charging system, etc.), the system fails to reach the threshold voltage, it could enter a cycle like charge → verification → reset → charge → verification → reset... Since such cycles can be arbitrarily long and vary in nature, two functions have been provided for cycle detection. Given cycle detection, one can detet the abnormal behavior and take decisions based on that. This makes the code more robust.

Closed cycle detection

Detects if exists a closed cycles on state machine. Refer to detect_closed_cycle()

Windowed cycle detection

Detects cycles on a window. Refer to detect_windowed_cycles()

Usage

    def test_fcn():
        return True

    f = fsm()

    f.add_transition('A => B : t0')
    f.add_transition('B => C : t1')
    f.add_transition('C => A : t2')
    f.add_transition('D => A : t3')

    f.add_condition('t0', 'a%10 == 0')
    f.add_condition('t1', 'a%10 == 0')
    f.add_condition('t2', 'a%10 == 0')
    f.add_condition('t3', test_fcn)

    f.compile()
    a = 0

    for j in range(130): 
        f.step()
        if (j > 100) and (len(f.true_transitions) != 0): 
            break
        a += 1

    print('done')
    print(f)

Exception classes

Ligth mode

Exception hierarchy

There are two main types of exceptions: - Invalid Syntax exceptions
Invalid syntax exceptions are triggered when we are defining FSM before executing it if the user violates the rules of FSM definition. The Syntax exceptions are: - Redundant condition : When one transition is use on more than one state transition condition. python f = fsm() f.add_transition('A => B : t0') # OK f.add_transition('B => C : t1') # OK f.add_transition('C => D : t2') # OK f.add_transition('D => A : t3') # OK f.add_transition('D => D : t2') # ERROR : Already defined transition for C = > D - Inconsistent transition : When the symbol used to describe a transition is not the same.
You can express transitions using 3 types of symbols: '->', '=>' and ',' (comma). Once one is chosen, it cannot be changed.

    For example: 
    * This is valid:

    ```python
            f = fsm()
            f.add_transition('A => B : t0')
            f.add_transition('B => C : t1')
            f.add_transition('C => A : t3')
            f.add_transition('D => A : t4')
            f.add_transition('D => D : t2')
    ```
    * This is also valid: 
    ```python 
            f = fsm()
            f.add_transition('A -> B : t0')
            f.add_transition('B -> C : t1')
            f.add_transition('C -> A : t3')
            f.add_transition('D -> A : t4')
            f.add_transition('D -> D : t2')

    ``` 
    * This too: 
    ```python 
            f = fsm()
            f.add_transition('A,B : t0')
            f.add_transition('B,C : t1')
            f.add_transition('C,A : t3')
            f.add_transition('D,A : t4')
            f.add_transition('D,D : t2')
    ``` 
    * But this is not valid:
    ```python
            f = fsm()
            f.add_transition('A => B : t0')
            f.add_transition('B => C : t1')
            f.add_transition('C -> A : t3') # Transition symbol syntax inconsistency
            f.add_transition('D -> A : t4')

    ```



t1, t2 and t3 are True at same time (non disjoint transitions)

This raises FSMNondisjoinctTransitions

pyfsmview

Remote HTTP FSM viewer in real time

pyfsmview is a minimalist http/websocket server to visualize and debug FSM from pyfsm module in real-time to visualize and debug FSM from pyfsm module on real time.

Dark mode

Dark mode

Ligth mode

Ligth mode

Minimal Example:

import asyncio
from pyfsm import fsm
from pyfsmview import pyfsm_http_visualizer

if __name__ == '__main__': 

    # FSM Overloading
    class test_fsm(fsm):
        def __init__(self, history_len=10) -> None:
            super().__init__(history_len)
            self.a = 0

        def tcondition(self)->bool:
            return True #(self.a % 10) == 0 

        def step(self) -> None:
            self.a += 1
            self.a %= 2
            super().step()

        # Custom co-routine (optional)
        async def printstate(self)->None:
            print("Prinstate running...")
            while True:
                print(f'{self.a} {self.state}')
                await asyncio.sleep(0.1)

    f = test_fsm()

    f.add_transition('A => B : t0')
    f.add_transition('B => C : t1')
    f.add_transition('C => D : t2')
    f.add_transition('D => A : t3')

    f.add_condition('t0', f.tcondition)
    f.add_condition('t1', f.tcondition)
    f.add_condition('t2', f.tcondition)
    f.add_condition('t3', f.tcondition)

    f.compile()

    # Create http visualizer instance in dark mode 
    service = pyfsm_http_visualizer(mode = 'dark')
    service.bind(f) # binds FSM to visualizer
    service.tasks.append(f.printstate) # Adds additional coroutine if you want

    service.fsmbind.ev_loop_flag.set() # Event loop flag must be set
    # Runs until Ctrl-C is pressed 
    try: 
        asyncio.run(service.start()) # Classic asyncio running coroutines
    except KeyboardInterrupt: 
        service.fsmbind.ev_running.clear() # Signal stop all 
        print("Ctrl-C detected. exit()")

Open a web browser and connect to url: localhost:8000

Installation

Coming soon....

License

Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

Source Code & Repository

The complete implementation of this project is open-source and hosted on GitHub.

📂 View Project on GitHub