Journal Sunday Python Pattern : Une machine à état toute simple

Posté par David Delassus (site web personnel) le 17 avril 2022 à 03:14. Licence CC By‑SA.

Étiquettes :

avr.

2022

Sommaire

Bonjour Nal,

Il y a un "design pattern" que je réutilise souvent dans différent langages pour découper la logique métier en plusieurs petit bout de code bien séparés et facilement testable.

Ce design pattern s'apparente très fortement à une machine à état :

on a la machine a état qui possède un contexte (des données qui seront manipulées et modifiées par l'exécution de l'algorithme)
chaque état va agir sur ce contexte et retourner l'état suivant à exécuter, ou None si on est arrivé au bout de l'algorithme

Ici, les transitions d'un état vers un autre sont modelées par le code : l'état actuel retourne l'état suivant

Pour les tests c'est assez simple : étant donné un contexte, quel est le type de retour de mon état ?

Une implémentation toute simple

Allez, un peu de code Python pour imager la chose… Au passage, j'en ai fait une librairie disponible sur Github.

Tout d'abord, on va définir le contexte :

from typing import TypeVar

Context = TypeVar("Context")

Le contexte peut être tout et n'importe quoi, mais on va explicitement le nommer pour la suite (comme ça, des outils tels que mypy vont pouvoir vérifier que l'on fait correctement les choses).

Ensuite, on va définir ce qu'est un état :

from typing import Protocol, Generic, Optional


class State(Protocol, Generic[Context]):
    def run(self, context: Context) -> Optional["State[Context]"]
        raise NotImplementedError

NB : Les guillemets sur le type de retour sont nécessaires, car State n'est pas encore complètement défini à cet endroit (cela ne pose aucun problème à mypy).

Ici, c'est l'équivalent Python d'une interface générique, en TypeScript cela donnerait :

interface State<Context> {
  run(context: Context): State<Context> | null
}

C'est ce que l'on va utiliser par la suite pour implémenter les différentes étapes de notre algorithme.

Enfin, on peut définir notre machine à état :

class StateMachine(Generic[Context]):
    def __init__(self, context: Context):
        self.context = context

    def run_from(self, state: Optional[State[Context]]):
        while state is not None:
            state = state.run(self.context)

Et c'est tout. Tant que l'on a un état a exécuter, on continu. C'est l'état qui contrôle le déroulement de l'algorithme.

Un premier exemple simple

Prenons un exemple tout simple, calculer une suite de syracuse :

from dataclasses import dataclass, field


@dataclass
class Stats:
    altitude: int = 0
    fly_time: int = 0
    suite: list[int] = field(default_factory=list)


class ComputeNextNumber(State[Stats]):
    def __init__(self, n: int):
        self.n = n

    def run(self, context: Stats) -> Optional[State[Stats]]:
        context.altitude = max(context.altitude, self.n)
        context.fly_time += 1
        context.suite.append(self.n)

        if self.n == 1:
            return None

        elif self.n % 2 == 0:
            return ComputeNextNumber(self.n // 2)

        else:
            return ComputeNextNumber(3 * self.n + 1)


class Syracuse(StateMachine[Stats]):
    def __init__(self):
        super().__init__(Stats())

    def compute(self, n: int) -> None:
        self.run_from(ComputeNextNumber(n))


fsm = Syracuse()
fsm.compute(5)

assert fsm.context.altitude == 16
assert fsm.context.fly_time == 6
assert fsm.context.suite == [5, 16, 8, 4, 2, 1]

L'entièreté de l'algorithme réside dans la classe ComputeNextNumber.

Dans cet exemple, le type associé à Context est Stats (grâce à State[Stats] et StateMachine[Stats]), et mypy sera en mesure de le vérifier.

Un exemple plus concret

Le second exemple représente un cas plus concret, si on essayait de parser une requête HTTP ?

Pour info, une requête HTTP est composée de :

une ligne de requête (version du protocole, verbe HTTP, URL)
zéro ou plusieurs en-têtes
une ligne vide
le corps de la requête

Par exemple :

POST /hello HTTP/1.1
Host: example.com

world

NB : Dans les exemples de code suivant, je vais omettre le détail de l'implémentation pour se focaliser sur l'utilisation de ce design pattern.

Je vois ici 3 étapes :

lire la ligne de requête
lire une ligne d'en-tête
lire le corps de la requête

En vérité, il y a une 4è étape qui peut intervenir à plusieurs endroits, pour la gestion d'erreur.

Lançons nous dans le code :

class HTTPContext:
  # ...



class ParseRequestLine(State[HTTPContext]):
    def run(self, context: HTTPContext) -> Optional[State[HTTPContext]]:
        line = context.socket.readline()

        if not line:  # EOF
            return UnexpectedError(RuntimeError("EOF"))

        try:
            verb, url, protocol = self._parse_request_line(line)
            return ParseHeaderLine(verb, url, protocol)

        except Exception as err:
            return UnexpectedError(err)

    def _parse_request_line(self, line: str) -> tuple[str, str, str]:
        # ...


class ParseHeaderLine(State[HTTPContext]):
    def __init__(self, verb: str, url: str, protocol: str, headers=None):
        if headers is None:
            headers = {}

        self.verb = verb
        self.url = url
        self.protocol = protocol
        self.headers = headers

    def run(self, context: HTTPContext) -> Optional[State[HTTPContext]]:
        line = context.socket.readline()

        if not line:  # EOF
            return UnexpectedError(RuntimeError("EOF"))

        line = line.strip()  # delete \r\n
        if not line:
            return ReadBody(self.verb, self.url, self.protocol, self.headers)

        try:
            header_name, header_val = self._parse_header_line(line)

        except Exception as err:
            return UnexpectedError(err)

        self.headers[header_name] = header_val
        return self  # MaximumDepthRecursion avoided!!!!

    def _parse_header_line(self, line: str) -> tuple[str, str]:
        # ...


class ReadBody(State[HTTPContext]):
    def __init__(self, verb: str, url: str, protocol: str, headers: dict[str, str]):
        self.verb = verb
        self.url = url
        self.protocol = protocol
        self.headers = headers

    def run(self, context: HTTPContext) -> Optional[State[HTTPContext]]:
        body = context.socket.read()
        context.set_request_info(
            verb=self.verb,
            url=self.url,
            protocol=self.protocol,
            headers=self.headers,
            body=self.body,
        )

        return None


class UnexpectedError(State[HTTPContext]):
    def __init__(self, err: Exception):
        self.err = err

    def run(self, context: HTTPContext) -> Optional[State[HTTPContext]]:
        context.set_error_info(self.err)

        return None

Dans cet exemple, on voit tout de suite les avantages :

on a du code récursif sans limite de récursion, en effet, on quitte la fonction avant de re-rentrer dedans, donc la pile d'appel n'est pas surchargée
chaque étape de l'algorithme est bien séparée, on évite ainsi une fonction gigantesque
une gestion d'erreur plutôt simple et sans goto, en effet il suffit de retourner l'état qui va traiter l'erreur
facile à tester, on va pouvoir "mocker" le HTTPContext et vérifier le type de retour de la méthode run() de chaque étape

Conclusion

Permettre à une fonction de retourner ce que l'appellant doit exécuter ensuite est un pattern très pratique, surtout quand il est question de changement de spec. Imaginons que la syntaxe de notre requête HTTP change, ici seul le code métier devra être revu, et en aucun cas le code qui l'appelle.

Le code qui appelle notre algorithme est entièrement piloté par notre algorithme au final (c'est nous qui lui disons quoi faire ensuite). J'ai simplifié pas mal de code métier dans différent projet en utilisant ce concept, que cela soit en Elixir, en Go, en Python, ou en TypeScript (je serais bien curieux de voir ce que cela donne en Rust).

Et toi Nal ? Tu en as des pattern sympa comme ça ?

implémentation Python : easy_fsm

# Machine à états ?

Posté par barmic 🦦 le 17 avril 2022 à 16:42. Évalué à 4.
Ce design pattern s'apparente très fortement à une machine à état

Je comprends le rapprochement avec une machine à états mais ça ne m'a pas l'air d'en être une du tout. Une machine a état se définie par les transitions entre les états. Un état est inerte et on suit des transitions pour aller d'état en état. Cela donne des propriétés très différentes : un état ne connais pas les autres destinations possibles à partir de lui. L'objectif c'est d'avoir une séparation entre le code métier et la navigation entre chaque partie et d'avoir une vision large du graphe[^1] complet de la machine à états.

Il me semble que c'est plus une manière de chaîner des fonctions de manière très découpées.
```
Optional[State[HTTPContext]]
```
C'est pas un peu lourd à l'usage ça ? Le state pourrait pas lui-même être une sorte de type somme : Next[HTTPContext] | Final | Error ?

J'ai une dernière question, comment tu récupère ton résultat à la fin ? C'est à la charge du dernier State de faire ce qu'il faut ou tu introspecte le dernier état ?

C'est peut être pas clair donc avec ton exemple, si je veux faire quelque chose de cette requête, il faut :
- ajouter un nouvel état qui fait suite à read body pour faire mon traitement
- ou à la fin de StateMachine.run_from() je vais chercher dans mon dernier state les données qui viennent d'être parsées
[^1]: cette notion de graphe est importantes pour certaines machines a états car ça permet de déduire des propriétés de l'algorithme comme par exemple prouver la terminaison.

https://linuxfr.org/users/barmic/journaux/y-en-a-marre-de-ce-gros-troll
- [^] # Re: Machine à états ?
  
  Posté par David Delassus (site web personnel) le 17 avril 2022 à 17:11. Évalué à 4.
  Je comprends le rapprochement avec une machine à états mais ça ne m'a pas l'air d'en être une du tout.
  
  J'avoue, c'est le nom le plus proche que j'ai trouvé pour ce design pattern que j'utilise sans en connaître le nom.
  
  Une machine a état se définie par les transitions entre les états. Un état est inerte et on suit des transitions pour aller d'état en état.
  
  Ici la "transition" c'est le retour de la fonction.
  
  Cela donne des propriétés très différentes : un état ne connais pas les autres destinations possibles à partir de lui. L'objectif c'est d'avoir une séparation entre le code métier et la navigation entre chaque partie et d'avoir une vision large du graphe[1] complet de la machine à états.
  
  En théorie oui, en pratique, je trouve toujours plus simple de donner la possibilité à l'état d'effectuer lui même la transition.
  
  Par exemple, en gamedev, j'utilise pas mal les "Stack Based State Machine", l'état retourne soit :
  - rien, on continu d'exécuter cet état (via OnUpdate)
  - Push new state, on appelle le OnPaused de l'état en court puis le OnEnter de l'état suivant
  - Pop state, on appelle le OnExit de l'état en court, puis le OnResume de l'état précédent
  C'est pratique pour implémenter les menus, la mise en pause, le game over, etc…
  
  Optional[State[HTTPContext]]
  C'est pas un peu lourd à l'usage ça ?
  
  En Python c'est strictement équivalent à State[HTTPContext] | None, le None représentant la fin de l'exécution de la State Machine.
  
  J'ai une dernière question, comment tu récupère ton résultat à la fin ?
  
  CF l'exemple avec la suite de Syracuse, c'est via la propriété context de l'objet StateMachine. Dans un langage immutable, on aurait plutôt :
```
while state is not None:
    state, context = state.run(context)
```
  C'est ce qu'on retrouve notamment avec les GenServer en Erlang/Elixir.
  
  Ici, le contexte ce sont les données en entrée ET en sortie de l'algorithme.
  
  https://link-society.com - https://kubirds.com - https://github.com/link-society/flowg
  - [^] # Re: Machine à états ?
    
    Posté par barmic 🦦 le 17 avril 2022 à 17:28. Évalué à 3.
    
    En théorie oui, en pratique, je trouve toujours plus simple de donner la possibilité à l'état d'effectuer lui même la transition.
    
    Les vérifications de propriétés ça n'est pas que de la théorie c'est très utilisé. Mais oui ton approche et celle des fsm ne répondent pas au même besoin.
    
    Je ne voulais pas du tout invalider ton approche, je montrais juste ce qui le distingue d'une fsm.
    
    Par exemple, en gamedev, j'utilise pas mal les "Stack Based State Machine", l'état retourne soit :
    
    Je connais pas les machines à états basées sur une stack, je regarderais.
    
    CF l'exemple avec la suite de Syracuse, c'est via la propriété context de l'objet StateMachine.
    
    Effectivement quand je me suis posé la question j'ai regardé l'exemple sur le parsing http, je suis pas remonté plus haut.
    
    https://linuxfr.org/users/barmic/journaux/y-en-a-marre-de-ce-gros-troll
    - [^] # Re: Machine à états ?
      
      Posté par David Delassus (site web personnel) le 17 avril 2022 à 17:49. Évalué à 3.
      Je ne voulais pas du tout invalider ton approche
      
      Loin de moi l'idée de penser ça, j'apprécie grandement tes éclaircissements sur le sujet.
      
      Je connais pas les machines à états basées sur une stack, je regarderais.
      
      Pour faire simple, on aurait :
      
      class State: def on_enter(self): pass def on_update(self): pass def on_pause(self): pass def on_resume(self): pass def on_exit(self): pass class StateMachine: def __init__(self, initial_state): self.states = [initial_state] initial_state.on_enter() def update(self): if len(self.states) == 0: return current_state = self.states[-1] action = current_state.on_update() match action: case "pop": current_state.on_exit() self.states.pop() if len(self.states) > 0: current_state = self.states[-1] current_state.on_resume() case ("push", new_state): current_state.on_pause() new_state.on_enter() self.states.append(new_state)
      
      Après, selon ce que tu veux faire, tu modifie la signature des callbacks pour prendre en paramètre un delta_time, un contexte, ou tout ce que tu veux.
      
      Avec ce modèle, mettre le jeu en pause revient à pousser le state Paused, et reprendre le jeu revient à pop ce state. Peu importe le state précédent cela fonctionnera de la même manière. C'est dans le on_update() que l'on va récupérer (ou ignorer) les inputs et autres scripts/actions du jeu. Les autres callbacks sont utiles pour tout ce qui est chargement de ressource ou déclenchement de script.
      
      https://link-society.com - https://kubirds.com - https://github.com/link-society/flowg
  - [^] # Re: Machine à états ?
    
    Posté par chimrod (site web personnel) le 17 avril 2022 à 21:04. Évalué à 6.
    
    J'avoue, c'est le nom le plus proche que j'ai trouvé pour ce design pattern que j'utilise sans en connaître le nom.
    
    Cela ressemble beaucoup à un trampoline, et c’est utilisé depuis au moins l’époque de Lisp. Cela permet de créer un code récursif sans surcharger la pile quand le compilateur n’est pas capable d’optimiser la récursion terminale.
    
    Je laisse les experts préciser s’il s’agit d’un cas particulier de Continuation-passing style (désolé, je saurai pas comment le traduire en français) ou non.
    - [^] # Re: Machine à états ?
      
      Posté par David Delassus (site web personnel) le 17 avril 2022 à 21:40. Évalué à 2.
      Cela ressemble beaucoup à un trampoline
      
      Ah sympa, je connaissais pas. Pour les langages ne supportant pas la récursion, je pensais qu'on utilisait plutôt le Y-Combinator.
      
      Je laisse les experts préciser s’il s’agit d’un cas particulier de Continuation-passing style
      
      Il me semble que les continuations sont le mécanisme derrière les générateurs et coroutines, par exemple :
      
      def generator(): yield 1 yield 2 yield 3 this_is_a_continuation = generator() assert 1 == this_is_a_continuation.send(None) assert 2 == this_is_a_continuation.send(None) assert 3 == this_is_a_continuation.send(None)
      
      Ici je n'interromp pas l'exécution de la fonction pour la reprendre plus tard, je retourne plutôt la fonction suivante à exécuter.
      
      https://link-society.com - https://kubirds.com - https://github.com/link-society/flowg
  - [^] # Re: Machine à états ?
    
    Posté par Cyrille Pontvieux (site web personnel, Mastodon) le 18 avril 2022 à 02:24. Évalué à 7.
    
    Ça s’appelle un Behavior Driven State Machine, autrement raccourci en BDSM ;-)
    
    … j’ai pas résisté :-D
    - [^] # Re: Machine à états ?
      
      Posté par David Delassus (site web personnel) le 18 avril 2022 à 02:38. Évalué à 4.
      
      Si j'avais su, j'aurais pas hésité à publier sur PyPI easy_bdsm.
      
      https://link-society.com - https://kubirds.com - https://github.com/link-society/flowg
# J'aime beaucoup !

Posté par Philippe F (site web personnel) le 17 avril 2022 à 18:18. Évalué à 4.

Ca m'arrive souvent de faire des parser pour des formats de fichiers relativement simple, et je trouvais que mon code manquait d'un bon Design Pattern approprié. Le voici enfin! Je m'en sortais avec des variables globales pour le contexte, une variable pour l'état et plein de if/else qui au bout d'un moment deviennent un peu compliqués à maintenir.

J'ai aussi eu le cas d'une boucle d'évènement d'un jeu à réaliser dernièrement et j'ai rencontré exactement le besoin que du décris (du très très classique en fait).

Je garde ça dans ma besace. J'aime beaucoup la simplicité, tu as avant-tout une classe avec une méthode run(). Question naive, est-ce qu'une fonction ne suffirai pas dans les cas très simple ?
- [^] # Re: J'aime beaucoup !
  
  Posté par David Delassus (site web personnel) le 17 avril 2022 à 18:53. Évalué à 4.
  Question naive, est-ce qu'une fonction ne suffirai pas dans les cas très simple ?
  
  Tout à fait possible d'utiliser une fonction :
```
def step1():
    # do stuff
    return step2


def step2():
    # do stuff
    return step3(data)


def step3(data):
    def run():
        # do stuff
        return None

    return run

state = step1
while state is not None:
    state = state()
```
  Cependant, pour des raisons de typage statique, je préfère utiliser une classe.
  
  https://link-society.com - https://kubirds.com - https://github.com/link-society/flowg
- [^] # Re: J'aime beaucoup !
  
  Posté par Michaël (site web personnel) le 19 avril 2022 à 07:24. Évalué à 3.
  
  Pour les lexers simples l’approche typique est d’utiliser des fonctions mutuellement récursives, type read_whitespace, read_identifier, read_immediate_string , read_expression, etc. Comme Python ne sait pas faire l’optimisation de récursion terminale, on s’en sort en faisant comme l’OP. Bon courage!
# Monade ?

Posté par Thomas Douillard le 18 avril 2022 à 13:05. Évalué à 3.

Ça ressemble furieusement a un structure de monade "état" en Haskell non ?
- [^] # Re: Monade ?
  
  Posté par David Delassus (site web personnel) le 18 avril 2022 à 13:20. Évalué à 3. Dernière modification le 18 avril 2022 à 13:20.
  
  Pas vraiment non.
  
  Le but de la "State Monad" est de pouvoir composer plusieurs changements d'états ensemble. C'est le code appellant (caller) qui s'occupe de la composition.
  
  Ici, c'est le code appelé (callee) qui s'occupe de la composition.
  
  Les différentes instances d'une "State Monad" sont complètement découplées, ici non il y a un couplage fort entre chacune des étapes de l'algorithme, il s'agit plus d'une découpe d'une grosse fonction en plusieurs étapes que d'une "State Monad".
  
  https://link-society.com - https://kubirds.com - https://github.com/link-society/flowg
  - [^] # Re: Monade ?
    
    Posté par Thomas Douillard le 19 avril 2022 à 20:23. Évalué à 3. Dernière modification le 19 avril 2022 à 20:23.
    
    En tout cas, il semble exister des monades pour les machines à état : https://coot.me/posts/categories-with-monadic-effects.html
# Automate fini en C avec switch

Posté par PR le 18 avril 2022 à 18:41. Évalué à 5.
Ça ressemble vaguement à une technique que j’avais vu en C, dont je ne retrouve malheureusement plus le lien.

L’idée est grosso modo la même : plutôt que d’avoir des données qui indiquent l’état, on a une variable qui indique le prochain code à exécuter. Ça se fait avec un switch :
```
void
machine(struct event *ev)
{
    static int state = INIT;
    switch (state) {
    ...
    case A:
        ...
        if (condition_X(ev))
            state = X;
        ...
        return;
    case B:
        ...
    ...
    }
}
```
Il faut savoir qu’un switch optimisé se résoud avec une jump table. Ainsi la variable state représente bien la partie du code qui va être exécutée au prochain appel de la fonction. On peut faire avec un pointeur de fonction aussi bien sûr.

Mort aux cons !
# J'ai aussi beaucoup utilisé un pattern similaire

Posté par syj le 20 avril 2022 à 23:31. Évalué à 2.
Pendant des années , j'ai utilisé un pattern similaire. Il y a encore des traces sur ce projet sourceforge.
https://sourceforge.net/projects/planningrh/

Spécialement, le framework associé à ce projet.
https://sourceforge.net/p/planningrh/code/HEAD/tree/stateengine/

La force de ce pattern pour décrire un screen flow, c'est qu'une "Action | Transition" peut-être agnostique de son état de départ.
Résultat, tu peux très facilement créer des actions qui vont manipuler ta machine à état et l'utiliser depuis différents endroit .

Par exemple:

Action: "Editer une fiche client"
Cette action crée un état "fiche client" avec l'état précédent en paramètre.
L'état "fiche client" supporte alors une action ActionBack qui permet de revenir à l'état précédent.
```
                            |------------------|
ActionEditCustomer          |                  |    ActionBack
--------------------------->|EditCustomerState |----------------------->
                            |                  |
                            |------------------|
```
Tu peux alors plugger ton action éditer la fiche client dans tous les écrans qui ont un client sélectionné.
Sans ajouter de code, tu peux alors revenir sur l'écran de départ.

D'une certaines manières, VueX reprend un peu ce principe.

Suivre le flux des commentaires

Note : les commentaires appartiennent à celles et ceux qui les ont postés. Nous n’en sommes pas responsables.