Merge pull request #86 from simonsobs/dev

Replace a factory method of FSM with a config dict

src/nextline_schedule/auto/state_machine/config.py

@@ -0,0 +1,102 @@
+'''The configuration of the finite state machine of the auto mode states.
+
+The package "transitions" is used: https://github.com/pytransitions/transitions
+
+State Diagram:
+
+                     .-------------.
+                     |   Created   |
+                     '-------------'
+                            | start()
+                            |
+                            v
+                     .-------------.
+     .-------------->|     Off     |<--------------.
+     |               '-------------'               |
+     |                      | turn_on()            |
+   turn_off()               |                 on_raised()
+     |                      |                      |
+     |                      |                      |
+     |   .------------------+------------------.   |
+     |   |   Auto           |                  |   |
+     |   |                  v                  |   |
+     |   |           .-------------.           |   |
+     |   |           |   Waiting   |           |   |
+     |   |           '-------------'           |   |
+     |   |                  | on_initialized() |   |
+     |   |                  | on_finished()    |   |
+     |   |                  v                  |   |
+     |   |           .-------------.           |   |
+     |   |           |   Pulling   |           |   |
+     '---|           '-------------'           |---'
+         |         run() |     ^               |
+         |               |     |               |
+         |               v     | on_finished() |
+         |           .-------------.           |
+         |           |   Running   |           |
+         |           '-------------'           |
+         |                                     |
+         '-------------------------------------'
+
+>>> class Model:
+...     def on_enter_auto_waiting(self):
+...         print('enter the waiting state')
+...         self.on_finished()
+...
+...     def on_exit_auto_waiting(self):
+...        print('exit the waiting state')
+...
+...     def on_enter_auto_pulling(self):
+...        print('enter the pulling state')
+
+>>> from transitions.extensions import HierarchicalMachine
+
+>>> model = Model()
+>>> machine = HierarchicalMachine(model=model, **CONFIG)
+>>> model.state
+'created'
+
+>>> _ = model.start()
+>>> model.state
+'off'
+
+>>> _ = model.turn_on()
+enter the waiting state
+exit the waiting state
+enter the pulling state
+
+>>> model.state
+'auto_pulling'
+'''
+
+
+_AUTO_SUB_STATE_CONFIG = {
+    'name': 'auto',
+    'children': ['waiting', 'pulling', 'running'],
+    'initial': 'waiting',
+    'transitions': [
+        ['on_initialized', 'waiting', 'pulling'],
+        ['on_finished', 'waiting', 'pulling'],
+        ['run', 'pulling', 'running'],
+        ['on_finished', 'running', 'pulling'],
+    ],
+}
+
+CONFIG = {
+    'name': 'global',
+    'states': ['created', 'off', _AUTO_SUB_STATE_CONFIG],
+    'transitions': [
+        ['start', 'created', 'off'],
+        ['turn_on', 'off', 'auto'],
+        ['on_raised', 'auto', 'off'],
+        {
+            'trigger': 'turn_off',
+            'source': 'auto',
+            'dest': 'off',
+            'before': 'cancel_task',
+        },
+    ],
+    'initial': 'created',
+    'queued': True,
+    'ignore_invalid_triggers': True,
+}

src/nextline_schedule/auto/state_machine/machine.py

@@ -4,8 +4,9 @@
 from typing import Any, Protocol
 
 from nextline.utils import pubsub
+from transitions.extensions.asyncio import HierarchicalAsyncMachine
 
-from .factory import build_state_machine
+from .config import CONFIG
 
 
 class CallbackType(Protocol):
@@ -31,7 +32,7 @@ def __init__(self, callback: CallbackType):
         self._pubsub_state = pubsub.PubSubItem[str]()
         self._logger = getLogger(__name__)
 
-        machine = build_state_machine(model=self)
+        machine = HierarchicalAsyncMachine(model=self, **CONFIG)  # type: ignore
         machine.after_state_change = [self.after_state_change.__name__]
 
         self.state: str  # attached by machine

tests/auto/state_machine/test_config.py

@@ -0,0 +1,97 @@
+from copy import deepcopy
+from pathlib import Path
+from unittest.mock import AsyncMock
+
+import pytest
+from hypothesis import given, settings
+from hypothesis import strategies as st
+from transitions import Machine
+from transitions.extensions import HierarchicalAsyncGraphMachine
+from transitions.extensions.asyncio import HierarchicalAsyncMachine
+from transitions.extensions.markup import HierarchicalMarkupMachine
+
+from nextline_schedule.auto.state_machine.config import CONFIG
+
+SELF_LITERAL = Machine.self_literal
+
+
+def test_model_default() -> None:
+    machine = HierarchicalAsyncMachine(model=None, **CONFIG)  # type: ignore
+    assert not machine.models
+
+
+def test_model_self_literal() -> None:
+    machine = HierarchicalAsyncMachine(model=SELF_LITERAL, **CONFIG)  # type: ignore
+    assert machine.models[0] is machine
+    assert len(machine.models) == 1
+
+
+def test_restore_from_markup() -> None:
+    machine = HierarchicalMarkupMachine(model=None, **CONFIG)  # type: ignore
+    assert isinstance(machine.markup, dict)
+    markup = deepcopy(machine.markup)
+    del markup['models']  # type: ignore
+    rebuild = HierarchicalMarkupMachine(model=None, **markup)  # type: ignore
+    assert rebuild.markup == machine.markup
+
+
+@pytest.mark.skip
+def test_graph(tmp_path: Path) -> None:  # pragma: no cover
+    FILE_NAME = 'states.png'
+    path = tmp_path / FILE_NAME
+    # print(f'Saving the state diagram to {path}...')
+    machine = HierarchicalAsyncGraphMachine(model=SELF_LITERAL, **CONFIG)  # type: ignore
+    machine.get_graph().draw(path, prog='dot')
+
+
+STATE_MAP = {
+    'created': {
+        'start': {'dest': 'off'},
+    },
+    'off': {
+        'turn_on': {'dest': 'auto_waiting'},
+    },
+    'auto_waiting': {
+        'turn_off': {'dest': 'off', 'before': 'cancel_task'},
+        'on_initialized': {'dest': 'auto_pulling'},
+        'on_finished': {'dest': 'auto_pulling'},
+        'on_raised': {'dest': 'off'},
+    },
+    'auto_pulling': {
+        'run': {'dest': 'auto_running'},
+        'turn_off': {'dest': 'off', 'before': 'cancel_task'},
+        'on_raised': {'dest': 'off'},
+    },
+    'auto_running': {
+        'on_finished': {'dest': 'auto_pulling'},
+        'turn_off': {'dest': 'off', 'before': 'cancel_task'},
+        'on_raised': {'dest': 'off'},
+    },
+}
+
+TRIGGERS = list({trigger for v in STATE_MAP.values() for trigger in v.keys()})
+
+
+@settings(max_examples=200)
+@given(triggers=st.lists(st.sampled_from(TRIGGERS)))
+async def test_transitions(triggers: list[str]) -> None:
+    machine = HierarchicalAsyncMachine(model=SELF_LITERAL, **CONFIG)  # type: ignore
+    assert machine.is_created()
+
+    for trigger in triggers:
+        prev = machine.state
+        if (map_ := STATE_MAP[prev].get(trigger)) is None:
+            await getattr(machine, trigger)()
+            assert machine.state == prev
+            continue
+
+        if before := map_.get('before'):
+            setattr(machine, before, AsyncMock())
+
+        assert await getattr(machine, trigger)() is True
+        dest = map_['dest']
+        assert getattr(machine, f'is_{dest}')()
+
+        if before:
+            assert getattr(machine, before).call_count == 1
+            assert getattr(machine, before).await_count == 1