{ "cells": [ { "cell_type": "markdown", "metadata": {}, "source": [ "# Playground\n", "\n", "Make sure to read the [documentation](https://github.com/pytransitions/transitions#table-of-contents) first.\n", "\n", "Links won't work on Github but should work with [nbviewer](https://nbviewer.org/github/pytransitions/transitions/blob/master/examples/Playground.ipynb).\n", "\n", "* [Rescue those kittens!](#Rescue-those-kittens!)\n", "* [Too much coffee with my hierarchical state machines!](#Too-much-coffee-with-my-hierarchical-state-machines!)\n", "* [Very asynchronous dancing](#Very-asynchronous-dancing)\n", "* [Fun with graphs](#Fun-with-graphs)\n", "\n", "\n", "## Rescue those kittens!" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "from transitions import Machine\n", "import random\n", "\n", "class NarcolepticSuperhero(object):\n", "\n", " # Define some states. Most of the time, narcoleptic superheroes are just like\n", " # everyone else. Except for...\n", " states = ['asleep', 'hanging out', 'hungry', 'sweaty', 'saving the world']\n", " # A more compact version of the quickstart transitions\n", " transitions = [['wakeup', 'asleep', 'hanging out'],\n", " ['work_out', 'hanging out', 'hungry'],\n", " ['eat', 'hungry', 'hanging out'],\n", " {'trigger': 'distress_call', 'source': '*', 'dest': 'saving the world', 'before': 'change_into_super_secret_costume'},\n", " {'trigger': 'complete_mission', 'source': 'saving the world', 'dest': 'sweaty', 'after': 'update_journal'},\n", " {'trigger': 'clean_up', 'source': 'sweaty', 'dest': 'asleep', 'conditions': 'is_exhausted'},\n", " ['clean_up', 'sweaty', 'hanging out'],\n", " ['nap', '*', 'asleep']]\n", "\n", "\n", " def __init__(self, name):\n", "\n", " # No anonymous superheroes on my watch! Every narcoleptic superhero gets\n", " # a name. Any name at all. SleepyMan. SlumberGirl. You get the idea.\n", " self.name = name\n", " self.kittens_rescued = 0 # What have we accomplished today?\n", "\n", " # Initialize the state machine\n", " self.machine = Machine(model=self, states=NarcolepticSuperhero.states,\n", " transitions=NarcolepticSuperhero.transitions, initial='asleep')\n", "\n", " def update_journal(self):\n", " \"\"\" Dear Diary, today I saved Mr. Whiskers. Again. \"\"\"\n", " self.kittens_rescued += 1\n", "\n", " @property\n", " def is_exhausted(self):\n", " \"\"\" Basically a coin toss. \"\"\"\n", " return random.random() < 0.5\n", "\n", " def change_into_super_secret_costume(self):\n", " print(\"Beauty, eh?\")\n", " \n", " def yell(self):\n", " print(f\"I am {self.name} and I am {self.state}!\")\n", " \n", "batman = NarcolepticSuperhero(\"Batman\")\n", "batman.wakeup()\n", "assert batman.state == 'hanging out'\n", "batman.yell()\n", "# the rest is up to you ..." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Too much coffee with my hierarchical state machines!" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "from transitions.extensions import HierarchicalMachine as Machine\n", "\n", "states = ['standing', 'walking', {'name': 'caffeinated', 'children':['dithering', 'running']}]\n", "transitions = [\n", " ['walk', 'standing', 'walking'],\n", " ['stop', 'walking', 'standing'],\n", " ['drink', '*', 'caffeinated'],\n", " ['walk', ['caffeinated', 'caffeinated_dithering'], 'caffeinated_running'],\n", " ['relax', 'caffeinated', 'standing']\n", "]\n", "\n", "machine = Machine(states=states, transitions=transitions, initial='standing', ignore_invalid_triggers=True)\n", "\n", "assert machine.walk() # Walking now\n", "# I fancy a coffee right now ..." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Very asynchronous dancing" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "scrolled": false }, "outputs": [], "source": [ "from transitions.extensions.asyncio import AsyncMachine\n", "import asyncio\n", "\n", "class Dancer:\n", " \n", " states = ['start', 'left_food_left', 'left', 'right_food_right']\n", " \n", " def __init__(self, name, beat):\n", " self.my_name = name\n", " self.my_beat = beat\n", " self.moves_done = 0\n", " \n", " async def on_enter_start(self):\n", " self.moves_done += 1\n", " \n", " async def wait(self):\n", " print(f'{self.my_name} stepped {self.state}')\n", " await asyncio.sleep(self.my_beat)\n", "\n", " async def dance(self):\n", " while self.moves_done < 5:\n", " await self.step()\n", " \n", "dancer1 = Dancer('Tick', 1)\n", "dancer2 = Dancer('Tock', 1.1)\n", "\n", "m = AsyncMachine(model=[dancer1, dancer2], states=Dancer.states, initial='start', after_state_change='wait')\n", "m.add_ordered_transitions(trigger='step')\n", "\n", "# it starts okay but becomes quite a mess\n", "_ = await asyncio.gather(dancer1.dance(), dancer2.dance()) " ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Fun with graphs\n", "\n", "This requires `pygraphviz` or `graphviz`" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "from transitions.extensions.states import Timeout, Tags, add_state_features\n", "from transitions.extensions.diagrams import GraphMachine\n", "\n", "import io\n", "from IPython.display import Image, display, display_png\n", "\n", "\n", "@add_state_features(Timeout, Tags)\n", "class CustomMachine(GraphMachine):\n", " pass\n", "\n", "\n", "states = ['new', 'approved', 'ready', 'finished', 'provisioned',\n", " {'name': 'failed', 'on_enter': 'notify', 'on_exit': 'reset',\n", " 'tags': ['error', 'urgent'], 'timeout': 10, 'on_timeout': 'shutdown'},\n", " 'in_iv', 'initializing', 'booting', 'os_ready', {'name': 'testing', 'on_exit': 'create_report'},\n", " 'provisioning']\n", "\n", "transitions = [{'trigger': 'approve', 'source': ['new', 'testing'], 'dest':'approved',\n", " 'conditions': 'is_valid', 'unless': 'abort_triggered'},\n", " ['fail', '*', 'failed'],\n", " ['add_to_iv', ['approved', 'failed'], 'in_iv'],\n", " ['create', ['failed','in_iv'], 'initializing'],\n", " ['init', 'in_iv', 'initializing'],\n", " ['finish', 'approved', 'finished'],\n", " ['boot', ['booting', 'initializing'], 'booting'],\n", " ['ready', ['booting', 'initializing'], 'os_ready'],\n", " ['run_checks', ['failed', 'os_ready'], 'testing'],\n", " ['provision', ['os_ready', 'failed'], 'provisioning'],\n", " ['provisioning_done', 'provisioning', 'os_ready']]\n", "\n", "\n", "class Model:\n", " \n", " # graph object is created by the machine\n", " def show_graph(self, **kwargs):\n", " stream = io.BytesIO()\n", " self.get_graph(**kwargs).draw(stream, prog='dot', format='png')\n", " display(Image(stream.getvalue()))\n", " \n", " def is_valid(self):\n", " return True\n", " \n", " def abort_triggered(self):\n", " return False\n", "\n", "model = Model()\n", "machine = CustomMachine(model=model, states=states, transitions=transitions, initial='new', title='System State',\n", " show_conditions=True, show_state_attributes=True)\n", "model.approve()\n", "model.show_graph()\n", "\n", "# Your turn! What happens next? " ] } ], "metadata": { "kernelspec": { "display_name": "Python 3.8.3 64-bit ('transitions': conda)", "language": "python", "name": "python38364bittransitionsconda9f9fdeb4313741768b0dccf7fd8ce480" }, "language_info": { "codemirror_mode": { "name": "ipython", "version": 3 }, "file_extension": ".py", "mimetype": "text/x-python", "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", "version": "3.8.3" } }, "nbformat": 4, "nbformat_minor": 4 }