examples/Frequently asked questions.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Frequently asked questions\n",
    "\n",
    "## Table of contents\n",
    "\n",
    "Links won't work on Github but should work with [nbviewer](https://nbviewer.org/github/pytransitions/transitions/blob/master/examples/Frequently%20asked%20questions.ipynb).\n",
    "\n",
    "* [How do I load/save state machine configurations with json/yaml](#How-do-I-load/save-state-machine-configurations-with-json/yaml)\n",
    "* [How to use transitions with django models?](#How-to-use-transitions-with-django-models?)\n",
    "* [transitions memory footprint is too large for my Django app and adding models takes too long.](#transitions-memory-footprint-is-too-large-for-my-Django-app-and-adding-models-takes-too-long.) \n",
    "* [Is there a during callback which is called when no transition has been successful?](#Is-there-a-'during'-callback-which-is-called-when-no-transition-has-been-successful?)\n",
    "* [How to have a dynamic transition destination based on a function's return value?](#How-to-have-a-dynamic-transition-destination-based-on-a-function's-return-value)\n",
    "* [Machine.get_triggers should only show valid transitions based on some conditions.](#Machine.get_triggers-should-only-show-valid-transitions-based-on-some-conditions.)\n",
    "* [Transitions does not add convencience methods to my model](#Transitions-does-not-add-convencience-methods-to-my-model)\n",
    "* [I have several inter-dependent machines/models and experience deadlocks](#How-can-I-edit-a-graph's-styling)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### How do I load/save state machine configurations with json/yaml\n",
    "\n",
    "The easiest way to load a configuration is by making sure it is structured just as the `Machine` constructor. Your first level elements should be `name`, `transitions`, `states` and so on. When your yaml/json configuration is loaded, you can add your model programatically and pass the whole object to `Machine`.\n",
    "\n",
    "#### Loading a JSON configuration"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from transitions import Machine\n",
    "import json\n",
    "\n",
    "\n",
    "class Model:\n",
    "\n",
    "    def say_hello(self, name):\n",
    "        print(f\"Hello {name}!\")\n",
    "\n",
    "\n",
    "# import json\n",
    "json_config = \"\"\"\n",
    "{\n",
    "  \"name\": \"MyMachine\",\n",
    "  \"states\": [\n",
    "    \"A\",\n",
    "    \"B\",\n",
    "    { \"name\": \"C\", \"on_enter\": \"say_hello\" }\n",
    "  ],\n",
    "  \"transitions\": [\n",
    "    [\"go\", \"A\", \"B\"],\n",
    "    {\"trigger\": \"hello\", \"source\": \"*\", \"dest\": \"C\"}\n",
    "  ],\n",
    "  \"initial\": \"A\"\n",
    "}\n",
    "\"\"\"\n",
    "\n",
    "model = Model()\n",
    "\n",
    "config = json.loads(json_config)\n",
    "config['model'] = model  # adding a model to the configuration\n",
    "m = Machine(**config)  # **config unpacks arguments as kwargs\n",
    "assert model.is_A()\n",
    "model.go()\n",
    "assert model.is_B()\n",
    "model.hello(\"world\")  # >>> Hello world!\n",
    "assert model.state == 'C'"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "####  Loading a YAML configuration\n",
    "\n",
    "This example uses [pyyaml](https://pypi.org/project/PyYAML/)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from transitions import Machine\n",
    "import yaml\n",
    "\n",
    "\n",
    "class Model:\n",
    "\n",
    "    def say_hello(self, name):\n",
    "        print(f\"Hello {name}!\")\n",
    "\n",
    "        \n",
    "yaml_config = \"\"\"\n",
    "---\n",
    "\n",
    "name: \"MyMachine\"\n",
    "\n",
    "states:\n",
    "  - \"A\"\n",
    "  - \"B\"\n",
    "  - name: \"C\"\n",
    "    on_enter: \"say_hello\"\n",
    "\n",
    "transitions:\n",
    "  - [\"go\", \"A\", \"B\"]\n",
    "  - {trigger: \"hello\", source: \"*\", dest: \"C\"}\n",
    "\n",
    "initial: \"A\"\n",
    "\"\"\"\n",
    "\n",
    "model = Model()\n",
    "\n",
    "config = yaml.safe_load(yaml_config)  \n",
    "config['model'] = model  # adding a model to the configuration\n",
    "m = Machine(**config)  # **config unpacks arguments as kwargs\n",
    "assert model.is_A()\n",
    "model.go()\n",
    "assert model.is_B()\n",
    "model.hello(\"world\")  # >>> Hello world!\n",
    "assert model.state == 'C'"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Exporting YAML or JSON\n",
    "\n",
    "A default `Machine` does not keep track of its configuration but `transitions.extensions.markup.MarkupMachine` does. \n",
    "`MarkupMachine` cannot just be used to export your configuration but also to visualize or instrospect your configuration conveniently.\n",
    "Is is also the foundation for `GraphMachine`. You will see that `MarkupMachine` will always export every attribute even unset values. This makes such exports visually cluttered but easier to automatically process.\n",
    "If you plan to use such a configuration with a 'normal' `Machine`, you should remove the `models` attribute from the markup since `Machine` cannot process it properly.\n",
    "If you pass the (stored and loaded) configuration to another `MarkupMachine` however, it will attempt to create and initialize models for you."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "#export\n",
    "from transitions.extensions.markup import MarkupMachine\n",
    "import json\n",
    "import yaml\n",
    "\n",
    "\n",
    "class Model:\n",
    "\n",
    "    def say_hello(self, name):\n",
    "        print(f\"Hello {name}!\")\n",
    "\n",
    "\n",
    "model = Model()\n",
    "m = MarkupMachine(model=None, name=\"ExportedMachine\")\n",
    "m.add_state('A')\n",
    "m.add_state('B')\n",
    "m.add_state('C', on_enter='say_hello')\n",
    "m.add_transition('go', 'A', 'B')\n",
    "m.add_transition(trigger='hello', source='*', dest='C')\n",
    "m.initial = 'A'\n",
    "m.add_model(model)\n",
    "model.go()\n",
    "\n",
    "print(\"JSON:\")\n",
    "print(json.dumps(m.markup, indent=2))\n",
    "print('\\nYAML:')\n",
    "print(yaml.dump(m.markup))\n",
    "\n",
    "config2 = json.loads(json.dumps(m.markup))  # simulate saving and loading\n",
    "m2 = MarkupMachine(markup=config2)\n",
    "model2 = m2.models[0]  # get the initialized model\n",
    "assert model2.is_B()  # the model state was preserved\n",
    "model2.hello('again')  # >>> Hello again!\n",
    "assert model2.state == 'C'"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### How to use `transitions` with django models?\n",
    "\n",
    "In [this comment](https://github.com/pytransitions/transitions/issues/146#issuecomment-300277397) **proofit404** provided a nice example about how to use `transitions` and django together:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from django.db import models\n",
    "from django.db.models.signals import post_init\n",
    "from django.dispatch import receiver\n",
    "from django.utils.translation import ugettext_lazy as _\n",
    "from transitions import Machine\n",
    "\n",
    "\n",
    "class ModelWithState(models.Model):\n",
    "    ASLEEP = 'asleep'\n",
    "    HANGING_OUT = 'hanging out'\n",
    "    HUNGRY = 'hungry'\n",
    "    SWEATY = 'sweaty'\n",
    "    SAVING_THE_WORLD = 'saving the world'\n",
    "    STATE_TYPES = [\n",
    "        (ASLEEP, _('asleep')),\n",
    "        (HANGING_OUT, _('hanging out')),\n",
    "        (HUNGRY, _('hungry')),\n",
    "        (SWEATY, _('sweaty')),\n",
    "        (SAVING_THE_WORLD, _('saving the world')),\n",
    "    ]\n",
    "    state = models.CharField(\n",
    "        _('state'),\n",
    "        max_length=100,\n",
    "        choices=STATE_TYPES,\n",
    "        default=ASLEEP,\n",
    "        help_text=_('actual state'),\n",
    "    )\n",
    "\n",
    "\n",
    "@receiver(post_init, sender=ModelWithState)\n",
    "def init_state_machine(instance, **kwargs):\n",
    "\n",
    "    states = [state for state, _ in instance.STATE_TYPES]\n",
    "    machine = instance.machine = Machine(model=instance, states=states, initial=instance.state)\n",
    "    machine.add_transition('work_out', instance.HANGING_OUT, instance.HUNGRY)\n",
    "    machine.add_transition('eat', instance.HUNGRY, instance.HANGING_OUT)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### `transitions` memory footprint is too large for my Django app and adding models takes too long.\n",
    "\n",
    "We analyzed the memory footprint of `transitions` in [this discussion](https://github.com/pytransitions/transitions/issues/146) and could verify that the standard approach is not suitable to handle thousands of models. However, with a static (class) machine and some `__getattribute__` tweaking we can keep the convenience loss minimal:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from transitions import Machine\n",
    "from functools import partial\n",
    "from mock import MagicMock\n",
    "\n",
    "\n",
    "class Model(object):\n",
    "\n",
    "    machine = Machine(model=None, states=['A', 'B', 'C'], initial=None,\n",
    "                      transitions=[\n",
    "                          {'trigger': 'go', 'source': 'A', 'dest': 'B', 'before': 'before'},\n",
    "                          {'trigger': 'check', 'source': 'B', 'dest': 'C', 'conditions': 'is_large'},\n",
    "                      ], finalize_event='finalize')\n",
    "\n",
    "    def __init__(self):\n",
    "        self.state = 'A'\n",
    "        self.before = MagicMock()\n",
    "        self.after = MagicMock()\n",
    "        self.finalize = MagicMock()\n",
    "\n",
    "    @staticmethod\n",
    "    def is_large(value=0):\n",
    "        return value > 9000\n",
    "\n",
    "    def __getattribute__(self, item):\n",
    "        try:\n",
    "            return super(Model, self).__getattribute__(item)\n",
    "        except AttributeError:\n",
    "            if item in self.machine.events:\n",
    "                return partial(self.machine.events[item].trigger, self)\n",
    "            raise\n",
    "\n",
    "\n",
    "model = Model()\n",
    "model.go()\n",
    "assert model.state == 'B'\n",
    "assert model.before.called\n",
    "assert model.finalize.called\n",
    "model.check()\n",
    "assert model.state == 'B'\n",
    "model.check(value=500)\n",
    "assert model.state == 'B'\n",
    "model.check(value=9001)\n",
    "assert model.state == 'C'\n",
    "assert model.finalize.call_count == 4"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "You lose `model.is_<state>` convenience functions and the ability to add callbacks such as `Model.on_enter_<state>` automatically. However, the second limitation can be tackled with dynamic resolution in states as pointed out by [mvanderlee](https://github.com/mvanderlee) [here](https://github.com/pytransitions/transitions/issues/146#issuecomment-869049925):"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "pycharm": {
     "name": "#%%\n"
    }
   },
   "outputs": [],
   "source": [
    "from transitions import State\n",
    "import logging\n",
    "\n",
    "logger = logging.getLogger(__name__)\n",
    "\n",
    "\n",
    "class DynamicState(State):\n",
    "    \"\"\" Need to dynamically get the on_enter and on_exit callbacks since the\n",
    "        model can not be registered to the Machine due to Memory limitations\n",
    "    \"\"\"\n",
    "\n",
    "    def enter(self, event_data):\n",
    "        \"\"\" Triggered when a state is entered. \"\"\"\n",
    "        logger.debug(\"%sEntering state %s. Processing callbacks...\", event_data.machine.name, self.name)\n",
    "        if hasattr(event_data.model, f'on_enter_{self.name}'):\n",
    "            event_data.machine.callbacks([getattr(event_data.model, f'on_enter_{self.name}')], event_data)\n",
    "        logger.info(\"%sFinished processing state %s enter callbacks.\", event_data.machine.name, self.name)\n",
    "\n",
    "    def exit(self, event_data):\n",
    "        \"\"\" Triggered when a state is exited. \"\"\"\n",
    "        logger.debug(\"%sExiting state %s. Processing callbacks...\", event_data.machine.name, self.name)\n",
    "        if hasattr(event_data.model, f'on_exit_{self.name}'):\n",
    "            event_data.machine.callbacks([getattr(event_data.model, f'on_exit_{self.name}')], event_data)\n",
    "        logger.info(\"%sFinished processing state %s exit callbacks.\", event_data.machine.name, self.name)\n",
    "\n",
    "\n",
    "class DynamicMachine(Machine):\n",
    "    \"\"\"Required to use DynamicState\"\"\"\n",
    "    state_cls = DynamicState"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": false,
    "pycharm": {
     "name": "#%% md\n"
    }
   },
   "source": [
    "### Is there a 'during' callback which is called when no transition has been successful?\n",
    "\n",
    "Currently, `transitions` has no such callback. This example from the issue discussed [here](https://github.com/pytransitions/transitions/issues/342) might give you a basic idea about how to extend `Machine` with such a feature:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from transitions.core import Machine, State, Event, EventData, listify\n",
    "\n",
    "\n",
    "class DuringState(State):\n",
    "\n",
    "    # add `on_during` to the dynamic callback methods\n",
    "    # this way on_during_<state> can be recognized by `Machine`\n",
    "    dynamic_methods = State.dynamic_methods + ['on_during']\n",
    "    \n",
    "    # parse 'during' and remove the keyword before passing the rest along to state\n",
    "    def __init__(self, *args, **kwargs):\n",
    "        during = kwargs.pop('during', [])\n",
    "        self.on_during = listify(during)\n",
    "        super(DuringState, self).__init__(*args, **kwargs)\n",
    "\n",
    "    def during(self, event_data):\n",
    "        for handle in self.on_during:\n",
    "            event_data.machine.callback(handle, event_data)\n",
    "\n",
    "\n",
    "class DuringEvent(Event):\n",
    "\n",
    "    def _trigger(self, model, *args, **kwargs):\n",
    "        # a successful transition returns `res=True` if res is False, we know that\n",
    "        # no transition has been executed\n",
    "        res = super(DuringEvent, self)._trigger(model, *args, **kwargs)\n",
    "        if res is False:\n",
    "            state = self.machine.get_state(model.state)\n",
    "            event_data = EventData(state, self, self.machine, model, args=args, kwargs=kwargs)\n",
    "            event_data.result = res\n",
    "            state.during(event_data)\n",
    "        return res\n",
    "\n",
    "\n",
    "class DuringMachine(Machine):\n",
    "    # we need to override the state and event classes used by `Machine`\n",
    "    state_cls = DuringState\n",
    "    event_cls = DuringEvent\n",
    "\n",
    "\n",
    "class Model:\n",
    "\n",
    "    def on_during_A(self):\n",
    "        print(\"Dynamically assigned callback\")\n",
    "\n",
    "    def another_callback(self):\n",
    "        print(\"Explicitly assigned callback\")\n",
    "\n",
    "\n",
    "model = Model()\n",
    "machine = DuringMachine(model=model, states=[{'name': 'A', 'during': 'another_callback'}, 'B'],\n",
    "                        transitions=[['go', 'B', 'A']], initial='A', ignore_invalid_triggers=True)\n",
    "machine.add_transition('test', source='A', dest='A', conditions=lambda: False)\n",
    "\n",
    "assert not model.go()\n",
    "assert not model.test()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### How to have a dynamic transition destination based on a function's return value\n",
    "\n",
    "This has been a feature request [here](https://github.com/pytransitions/transitions/issues/269). We'd encourage to write a wrapper which converts a condensed statement into individual condition-based transitions. However, a less expressive version could look like this:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from transitions import Machine, Transition\n",
    "from six import string_types\n",
    "\n",
    "class DependingTransition(Transition):\n",
    "\n",
    "    def __init__(self, source, dest, conditions=None, unless=None, before=None,\n",
    "                 after=None, prepare=None, **kwargs):\n",
    "\n",
    "        self._result = self._dest = None\n",
    "        super(DependingTransition, self).__init__(source, dest, conditions, unless, before, after, prepare)\n",
    "        if isinstance(dest, dict):\n",
    "            try:\n",
    "                self._func = kwargs.pop('depends_on')\n",
    "            except KeyError:\n",
    "                raise AttributeError(\"A multi-destination transition requires a 'depends_on'\")\n",
    "        else:\n",
    "            # use base version in case transition does not need special handling\n",
    "            self.execute = super(DependingTransition, self).execute\n",
    "\n",
    "    def execute(self, event_data):\n",
    "        func = getattr(event_data.model, self._func) if isinstance(self._func, string_types) \\\n",
    "               else self._func\n",
    "        self._result = func(*event_data.args, **event_data.kwargs)\n",
    "        super(DependingTransition, self).execute(event_data)\n",
    "\n",
    "    @property\n",
    "    def dest(self):\n",
    "        return self._dest[self._result] if self._result is not None else self._dest\n",
    "\n",
    "    @dest.setter\n",
    "    def dest(self, value):\n",
    "        self._dest = value\n",
    "\n",
    "# subclass Machine to use DependingTransition instead of standard Transition\n",
    "class DependingMachine(Machine):\n",
    "    transition_cls = DependingTransition\n",
    "    \n",
    "\n",
    "def func(value):\n",
    "    return value\n",
    "\n",
    "m = DependingMachine(states=['A', 'B', 'C', 'D'], initial='A')\n",
    "# define a dynamic transition with a 'depends_on' function which will return the required value\n",
    "m.add_transition(trigger='shuffle', source='A', dest=({1: 'B', 2: 'C', 3: 'D'}), depends_on=func)\n",
    "m.shuffle(value=2)  # func returns 2 which makes the transition dest to be 'C'\n",
    "assert m.is_C()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Note that this solution has some drawbacks. For instance, the generated graph might not include all possible outcomes."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### `Machine.get_triggers` should only show valid transitions based on some conditions.\n",
    "\n",
    "This has been requested [here](https://github.com/pytransitions/transitions/issues/256). `Machine.get_triggers` is usually quite naive and only checks for theoretically possible transitions. If you need more sophisticated peeking, this `PeekMachine._can_trigger` might be a solution:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from transitions import Machine, EventData\n",
    "from transitions.core import listify\n",
    "from functools import partial\n",
    "\n",
    "\n",
    "class Model(object):\n",
    "\n",
    "    def fails(self, condition=False):\n",
    "        return False\n",
    "\n",
    "    def success(self, condition=False):\n",
    "        return True\n",
    "\n",
    "    # condition is passed by EventData\n",
    "    def depends_on(self, condition=False):\n",
    "        return condition\n",
    "\n",
    "    def is_state_B(self, condition=False):\n",
    "        return self.state == 'B'\n",
    "\n",
    "\n",
    "class PeekMachine(Machine):\n",
    "\n",
    "    def _can_trigger(self, model, *args, **kwargs):\n",
    "        # We can omit the first two arguments state and event since they are only needed for\n",
    "        # actual state transitions. We do have to pass the machine (self) and the model as well as\n",
    "        # args and kwargs meant for the callbacks.\n",
    "        e = EventData(None, None, self, model, args, kwargs)\n",
    "\n",
    "        return [trigger_name for trigger_name in self.get_triggers(model.state)\n",
    "                if any(all(c.check(e) for c in t.conditions)\n",
    "                       for t in self.events[trigger_name].transitions[model.state])]\n",
    "\n",
    "    # override Machine.add_model to assign 'can_trigger' to the model\n",
    "    def add_model(self, model, initial=None):\n",
    "        for mod in listify(model):\n",
    "            mod = self if mod is self.self_literal else mod\n",
    "            if mod not in self.models:\n",
    "                setattr(mod, 'can_trigger', partial(self._can_trigger, mod))\n",
    "                super(PeekMachine, self).add_model(mod, initial)\n",
    "\n",
    "states = ['A', 'B', 'C', 'D']\n",
    "transitions = [\n",
    "    dict(trigger='go_A', source='*', dest='A', conditions=['depends_on']),  # only available when condition=True is passed\n",
    "    dict(trigger='go_B', source='*', dest='B', conditions=['success']),  # always available\n",
    "    dict(trigger='go_C', source='*', dest='C', conditions=['fails']),  # never available\n",
    "    dict(trigger='go_D', source='*', dest='D', conditions=['is_state_B']),  # only available in state B\n",
    "    dict(trigger='reset', source='D', dest='A', conditions=['success', 'depends_on']), # only available in state D when condition=True is passed\n",
    "    dict(trigger='forwards', source='A', dest='D', conditions=['success', 'fails']),  # never available\n",
    "    dict(trigger='forwards', source='D', dest='D', unless=['depends_on'])\n",
    "]\n",
    "\n",
    "model = Model()\n",
    "machine = PeekMachine(model, states=states, transitions=transitions, initial='A', auto_transitions=False)\n",
    "assert model.can_trigger() == ['go_B']\n",
    "assert set(model.can_trigger(condition=True)) == set(['go_A', 'go_B'])\n",
    "model.go_B(condition=True)\n",
    "assert set(model.can_trigger()) == set(['go_B', 'go_D'])\n",
    "model.go_D()\n",
    "assert model.can_trigger() == ['go_B', 'forwards']\n",
    "assert set(model.can_trigger(condition=True)) == set(['go_A', 'go_B', 'reset'])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Transitions does not add convencience methods to my model\n",
    "\n",
    "There is a high chance that your model *already contained* a `trigger` method or methods with the same name as your even trigger. In this case, `transitions` will not add convenience methods to not accidentaly break your model and only emit a warning. If you defined these methods on purpose and *want* them to be overrided or maybe even call *both* -- the trigger event AND your predefined method, you can extend/override `Machine._checked_assignment` which is always called when something needs to be added to a model:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from transitions import State, Machine\n",
    "\n",
    "class StateMachineModel:\n",
    "\n",
    "    state = None\n",
    "\n",
    "    def __init__(self):\n",
    "        pass\n",
    "\n",
    "    def transition_one(self):\n",
    "        print('transitioning states...')\n",
    "\n",
    "    def transition_two(self):\n",
    "        print('transitioning states...')\n",
    "\n",
    "\n",
    "class OverrideMachine(Machine):\n",
    "\n",
    "    def _checked_assignment(self, model, name, func):\n",
    "        setattr(model, name, func)\n",
    "\n",
    "\n",
    "class CallingMachine(Machine):\n",
    "\n",
    "    def _checked_assignment(self, model, name, func):\n",
    "        if hasattr(model, name):\n",
    "            predefined_func = getattr(model, name)\n",
    "            def nested_func(*args, **kwargs):\n",
    "                predefined_func()\n",
    "                func(*args, **kwargs)\n",
    "            setattr(model, name, nested_func)\n",
    "        else:\n",
    "            setattr(model, name, func)\n",
    "\n",
    "\n",
    "states = [State(name='A'), State(name='B'), State(name='C'), State(name='D')]\n",
    "transitions = [\n",
    "    {'trigger': 'transition_one', 'source': 'A', 'dest': 'B'},\n",
    "    {'trigger': 'transition_two', 'source': 'B', 'dest': 'C'},\n",
    "    {'trigger': 'transition_three', 'source': 'C', 'dest': 'D'}\n",
    "]\n",
    "state_machine_model = StateMachineModel()\n",
    "\n",
    "print('OverrideMachine ...')\n",
    "state_machine = OverrideMachine(model=state_machine_model, states=states, transitions=transitions, initial=states[0])\n",
    "print('state_machine_model (current state): %s' % state_machine_model.state)\n",
    "state_machine_model.transition_one()\n",
    "print('state_machine_model (current state): %s' % state_machine_model.state)\n",
    "state_machine_model.transition_two()\n",
    "print('state_machine_model (current state): %s' % state_machine_model.state)\n",
    "\n",
    "print('\\nCallingMachine ...')\n",
    "state_machine_model = StateMachineModel()\n",
    "state_machine = CallingMachine(model=state_machine_model, states=states, transitions=transitions, initial=states[0])\n",
    "print('state_machine_model (current state): %s' % state_machine_model.state)\n",
    "state_machine_model.transition_one()\n",
    "print('state_machine_model (current state): %s' % state_machine_model.state)\n",
    "state_machine_model.transition_two()\n",
    "print('state_machine_model (current state): %s' % state_machine_model.state)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "pycharm": {
     "name": "#%% md\n"
    }
   },
   "source": [
    "### I have several inter-dependent machines/models and experience deadlocks\n",
    "\n",
    "A common use case involves multiple machines where one machine should react to events emitted by the other(s).\n",
    "Sometimes this involves 'waiting' as in 'all machines are triggered at the same time but machine 1 needs to wait until\n",
    "machine 2 is ready'. `Machine` will process callbacks sequentially. Thus, if your callbacks contain passages like this\n",
    "\n",
    "```python\n",
    "class Model:\n",
    "    def on_enter_state(self):\n",
    "        while not event:\n",
    "            time.sleep(1)\n",
    "```\n",
    "\n",
    "it is very likely that `event` will never happen because the callback will block the event processing forever.\n",
    "\n",
    "Bad news first: there is no one-fits-all-solution for this kind of problem.\n",
    "Now the good news: There is a solution that fits many use cases. An event bus! We consider transitions to be events that\n",
    "can be emitted by user input, system events or other machines.\n",
    "\n",
    "The event bus approach decouples the need of individual machines to know each other. They communicate via events.\n",
    "Thus, we can model quite complex inter-dependent behaviour without threading or asynchronous processing.\n",
    "Furthermore, other components do not need to know which machine processes which event, they can broadcast the message on\n",
    "the bus and rest assured that whoever is interested in the event will get it.\n",
    "The challenge is to wrap one's head around the concept of modelling transitions as events rather than actions to be conducted.\n",
    "\n",
    "Since we expect events to be emitted in callbacks, and we also expect that not every event bus member will be able to\n",
    "process every event sent across the bus, we pass `queued=True` (every machine processes one event at a time) and\n",
    "`ignore_invalid_triggers=True` (when the event cannot be triggered from the current state or is unknown, ignore it)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "pycharm": {
     "name": "#%%\n"
    }
   },
   "outputs": [],
   "source": [
    "from transitions import Machine\n",
    "import logging\n",
    "\n",
    "\n",
    "class EventBus:\n",
    "\n",
    "    def __init__(self):\n",
    "        self.members = []\n",
    "\n",
    "    def add_member(self, member):\n",
    "        \"\"\"Member can be a model or a machine acting as a model\"\"\"\n",
    "        # We decorate each member with an 'emit' function to fire events.\n",
    "        # EventBus will then broadcast that event to ALL members, including the one that triggered the event.\n",
    "        # Furthermore, we can pass a payload in case there is data that needs to be sent with an event.\n",
    "        setattr(member, 'emit', self.broadcast)\n",
    "        self.members.append(member)\n",
    "\n",
    "    def broadcast(self, event, payload=None):\n",
    "        for member in self.members:\n",
    "            member.trigger(event, payload)\n",
    "\n",
    "\n",
    "# Our machines can either be off or started\n",
    "states = ['off', 'started']\n",
    "\n",
    "\n",
    "class Machine1(Machine):\n",
    "\n",
    "    # this machine can only boot once.\n",
    "    transitions = [['boot', 'off', 'started']]\n",
    "\n",
    "    def __init__(self):\n",
    "        # we pass 'ignore_invalid_triggers' since a machine on an event bus might get events it cannot process\n",
    "        # right now and we do not want to throw an exception every time that happens.\n",
    "        # Furthermore, we will set 'queued=True' to process events sequentially instead of nested.\n",
    "        super(Machine1, self).__init__(states=states, transitions=self.transitions,\n",
    "                                       ignore_invalid_triggers=True, initial='off', queued=True)\n",
    "\n",
    "    def on_enter_started(self, payload=None):\n",
    "        print(\"Starting successful\")\n",
    "        # We emit out start event and attach ourselves as payload just in case\n",
    "        self.emit(\"Machine1Started\", self)\n",
    "\n",
    "\n",
    "class Machine2(Machine):\n",
    "\n",
    "    # This machine can also reboot (boot from every state) but only when the 'ready' flag has been set.\n",
    "    # 'ready' is set once the event 'Machine1Started' has been processed (before the transition is from 'off' to 'on'\n",
    "    # is actually executed). Furthermore, we will also boot the machine when we catch that event.\n",
    "    transitions = [{'trigger': 'boot', 'source': '*', 'dest': 'started', 'conditions': 'ready'},\n",
    "                   {'trigger': 'Machine1Started', 'source': 'off', 'dest': 'started', 'before': 'on_machine1_started'}]\n",
    "\n",
    "    def __init__(self):\n",
    "        super(Machine2, self).__init__(states=states, transitions=self.transitions,\n",
    "                                       ignore_invalid_triggers=True, initial='off', queued=True)\n",
    "        self._ready = False\n",
    "\n",
    "    # Callbacks also work with properties. Passing the string 'ready' will evaluate this property\n",
    "    @property\n",
    "    def ready(self):\n",
    "        return self._ready\n",
    "\n",
    "    @ready.setter\n",
    "    def ready(self, value):\n",
    "        self._ready = value\n",
    "\n",
    "    def on_machine1_started(self, payload=None):\n",
    "        self.ready = True\n",
    "        print(\"I am ready now!\")\n",
    "\n",
    "    def on_enter_started(self, payload=None):\n",
    "        print(\"Booting successful\")\n",
    "\n",
    "\n",
    "logging.basicConfig(level=logging.DEBUG)\n",
    "bus = EventBus()\n",
    "machine1 = Machine1()\n",
    "machine2 = Machine2()\n",
    "bus.add_member(machine2)\n",
    "bus.add_member(machine1)\n",
    "bus.broadcast('boot')\n",
    "# what will happen:\n",
    "# - bus will broadcast 'boot' event to machine2\n",
    "# - machine2 will attempt to boot but fail and return since ready is set to false\n",
    "# - bus will broadcast 'boot' event to machine1\n",
    "# - machine1 will boot and emit the 'Machine1Started'\n",
    "# - bus will broadcast 'Machine1Started' to machine2\n",
    "# - machine2 will handle the event, boot and return\n",
    "# - bus will broadcast 'Machine1Started' to machine1\n",
    "# - machine1 will add that event to its event queue\n",
    "# - bus broadcast of 'Machine1Started' returns\n",
    "# - machine1 is done with handling 'boot' and process the next event in the event queue\n",
    "# - machine1 cannot handle 'Machine1Started' and will ignore it\n",
    "# - bus broadcast of 'boot' returns\n",
    "assert machine1.state == machine2.state\n",
    "bus.broadcast('boot')\n",
    "# broadcast 'boot' event to all members:\n",
    "# - machine2 will reboot\n",
    "# - machine1 won't do anything"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": false,
    "pycharm": {
     "name": "#%% md\n"
    }
   },
   "source": [
    "If you consider this too much boilerplate and you don't mind some dependencies and less generalization\n",
    "you can of course go the leaner asynchronous or threaded route and process your callbacks in parallel.\n",
    "Having `while event` loops as mentioned above in `async` callbacks is not uncommon. You should consider, however, that\n",
    "the execution order of callbacks as described in the README is kept for `AsyncMachine` as well.\n",
    "All callbacks of one stage (e.g. `prepare`) must return before callbacks of the next state (e.g. `conditions`) are triggered."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "pycharm": {
     "name": "#%%\n"
    }
   },
   "outputs": [],
   "source": [
    "from transitions.extensions.asyncio import AsyncMachine\n",
    "import asyncio\n",
    "\n",
    "states = ['off', 'started']\n",
    "\n",
    "\n",
    "class Machine1(AsyncMachine):\n",
    "\n",
    "    transitions = [{'trigger': 'boot', 'source': 'off', 'dest': 'started', 'before': 'heavy_processing'}]\n",
    "\n",
    "    def __init__(self):\n",
    "        super(Machine1, self).__init__(states=states, transitions=self.transitions, initial='off')\n",
    "\n",
    "    async def heavy_processing(self):\n",
    "        # we need to do some heavy lifting before we can proceed with booting\n",
    "        await asyncio.sleep(0.5)\n",
    "        print(\"Processing done!\")\n",
    "\n",
    "\n",
    "class Machine2(AsyncMachine):\n",
    "\n",
    "    transitions = [['boot', 'off', 'started']]\n",
    "\n",
    "    def __init__(self, dependency):\n",
    "        super(Machine2, self).__init__(states=states, transitions=self.transitions, initial='off')\n",
    "        self.dependency = dependency\n",
    "\n",
    "    async def on_enter_started(self):\n",
    "        while not self.dependency.is_started():\n",
    "            print(\"Waiting for dependency to be ready...\")\n",
    "            await asyncio.sleep(0.1)\n",
    "        print(\"Machine2 up and running\")\n",
    "\n",
    "\n",
    "machine1 = Machine1()\n",
    "machine2 = Machine2(machine1)\n",
    "asyncio.get_event_loop().run_until_complete(asyncio.gather(machine1.boot(), machine2.boot()))\n",
    "assert machine1.state == machine2.state"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### How can I edit a graph's styling\n",
    "\n",
    "There are multiple ways to alter a graph's layout. Some are only available when you chose `pygraphviz` as backend. You can [1] edit `GraphMachine.machine_attributes` for Graph attributes such as flow direction and rank separation or `GraphMachine.style_attributes` for node and edge properties. The easiest way is to deepcopy the standard values and override properties you would like to change. Furthermore you can [2] work directly with the returned [pygraphviz.AGraph](https://pygraphviz.github.io/documentation/stable/reference/agraph.html#pygraphviz.AGraph.edges) object that `model.get_graph()` returns. However, values set via `edge_attr` and `node_attr` will be overriden by GraphMachine.style_attributes if they haven't been reset. `AGraph.draw` features `args` as a parameter to pass arguments to Graphviz layout engine directly. Eventually, you can also get the raw dot representation (e.g. `AGraph.string()`) as a string. Some styling examples can be found in the [Graph MIxin Demo](./Graph%20MIxin%20Demo.ipynb) notebook."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from copy import deepcopy\n",
    "from transitions.extensions import GraphMachine\n",
    "from collections import defaultdict\n",
    "\n",
    "states = ['A', 'B', 'C', 'D']\n",
    "transitions = [\n",
    "    {'trigger': 'go', 'source': 'A', 'dest': 'B'},\n",
    "    {'trigger': 'go', 'source': 'B', 'dest': 'C'},\n",
    "    {'trigger': 'back', 'source': 'C', 'dest': 'B'},\n",
    "    {'trigger': 'back', 'source': 'B', 'dest': 'A'},\n",
    "    {'trigger': 'forward', 'source': 'A', 'dest': 'C'},\n",
    "    {'trigger': 'forward', 'source': 'C', 'dest': 'D'},\n",
    "    {'trigger': 'backward', 'source': 'D', 'dest': 'C'},\n",
    "    {'trigger': 'backward', 'source': 'C', 'dest': 'A'}\n",
    "]\n",
    "\n",
    "class CustomGraphMachine(GraphMachine):\n",
    "\n",
    "    # Override Graphmachine's default styling for nodes and edges [1]\n",
    "    # Copy default graph attributes but change direction to top to bottom\n",
    "    machine_attributes = deepcopy(GraphMachine.machine_attributes)\n",
    "    machine_attributes[\"rankdir\"] = \"TB\"  # Left to right layout\n",
    "\n",
    "     # Reset styling\n",
    "    style_attributes = defaultdict(dict) \n",
    "    style_attributes[\"node\"][\"default\"] = { \"fontname\": \"arial\", \"shape\": \"circle\"}\n",
    "    style_attributes[\"edge\"][\"default\"] = { \"fontname\": \"arial\" }\n",
    "\n",
    "\n",
    "machine = CustomGraphMachine(states=states, transitions=transitions, initial='A', title=\"State Diagram\")\n",
    "graph = machine.get_graph()\n",
    "# directly modify the graph's attributes (works only with pygraphviz) [2]\n",
    "# note: this will only work for attributes that haven't been set by the GraphMachine\n",
    "graph.node_attr.update(fillcolor=\"turquoise\")\n",
    "graph.edge_attr.update(color=\"blue\")\n",
    "# modify the attributes of a specific edge\n",
    "graph.edges([\"A\", \"B\"])[0].attr.update(color=\"red\")\n",
    "# pass additional arguments to the graphviz layout engine via args [3]\n",
    "graph.draw('state_diagram.png', prog='dot', args='-Gnodesep=1')\n",
    "# get the graphs source code [4]\n",
    "source = graph.string()"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "transitions",
   "language": "python",
   "name": "python3"
  },
  "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.10.13"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}