feature complete version, tests and debug pending

.gitignore

@@ -1 +1,3 @@
-.idea
+.idea
+.venv
+**/__pycache__

src/snanosm/mealy.py

@@ -1,4 +1,5 @@
-from typing import Optional, Callable, Protocol, Self, TypeVar, Generic, List
+from enum import Enum
+from typing import Optional, Callable, Protocol, Self, TypeVar, Generic, List, Union
 
 
 # Anything hashable and equatable can be used as an input
@@ -10,7 +11,12 @@ class InputProtocol(Protocol):
         ...
 
 
+class TransitionInputEnum(Enum):
+    MATCH_REST = "#TRANSITION_MATCH_REST"
+
+
 Input = TypeVar("Input", bound=InputProtocol)
+TransitionInput = TypeVar("TransitionInput", bound=Union[InputProtocol, TransitionInputEnum])
 
 
 class State:
@@ -32,19 +38,26 @@ class State:
 
 
 class Transition(Generic[Input]):
-    def __init__(self, state_input: Input, origin_name: str, destination_name: str,
+    def __init__(self, transition_input: TransitionInput, origin_name: str, destination_name: str,
-                 output_function: Callable[[Self, object], None]) -> None:
+                 output_function: Callable[[Self, Optional[object]], None]) -> None:
-        self.__state_input = state_input
+        self.__transition_input = transition_input
         self.__origin_name = origin_name
         self.__destination_name = destination_name
+        self.__destination_name_hash = hash(destination_name)
         self.__output_function = output_function
 
+    def execute_output(self, context):
+        self.__output_function(context, self)
+
+    def get_destination_name_hash(self) -> int:
+        return self.__destination_name_hash
+
     def __str__(self) -> str:
-        return f"[Transition ({self.__origin_name}, {self.__destination_name}, {self.__state_input})]"
+        return f"[Transition ({self.__origin_name}, {self.__destination_name}, {self.__transition_input})]"
 
     def __hash__(self) -> int:
         # A tuple is only hashable if all its elements are hashable
-        return hash((self.__origin_name, self.__destination_name, self.__state_input))
+        return hash((self.__origin_name, self.__destination_name, self.__transition_input))
 
 
 class Machine(Generic[Input]):
@@ -53,9 +66,11 @@ class Machine(Generic[Input]):
         self.__end_states: List[int] = []
         self.__current_state: Optional[int] = None
         self.__states: dict[int, State] = {}
-        self.__transitions: dict[int, Transition] = {}
+        # dict[hash_origin, dict[hash_transition_input, Transition]]
+        self.__transitions: dict[int, dict[int, Transition]] = {}
+        self.__context = {}
 
-    def add_transition(self, state_input: Input, origin_name: str, destination_name: str,
+    def add_transition(self, transition_input: TransitionInput, origin_name: str, destination_name: str,
                        output_function: Callable[[Self, object], None]) -> None:
         ho = hash(origin_name)
         if ho not in self.__states.keys():
@@ -63,14 +78,22 @@ class Machine(Generic[Input]):
         hd = hash(destination_name)
         if hd not in self.__states.keys():
             raise ValueError(f"Destination state {destination_name} does not exist.")
-        t = Transition(state_input, origin_name, destination_name, output_function)
+        if isinstance(transition_input, TransitionInputEnum):
-        ht = hash(t)
+            hi = hash(transition_input.name)
-        if ht in self.__transitions.keys():
+        else:
-            raise ValueError(f"Transition ({origin_name}, {destination_name}, {state_input}) already exists.")
+            hi = hash(transition_input)
-        self.__transitions[ht] = t
+        t = Transition(transition_input, origin_name, destination_name, output_function)
+        if ho not in self.__transitions.keys():
+            self.__transitions[ho] = {hi: t}
+        elif hi not in self.__transitions[ho]:
+            self.__transitions[ho][hi] = t
+        elif hi in self.__transitions[ho]:
+            raise ValueError(f"Transition already exists for origin state {origin_name} and transition input {transition_input}.")
 
     def add_state(self, state_name: str, state_data: object, is_start_state: bool = False,
                   is_end_state: bool = False) -> None:
+        if state_name.startswith("#"):
+            raise ValueError("State names cannot start with # (reserved for special states)")
         h: int = hash(state_name)
         if is_start_state:
             if self.__start_state is None:
@@ -95,19 +118,35 @@ class Machine(Generic[Input]):
             raise ValueError("Resetting a machine without starting state.")
         self.__current_state = self.__start_state
 
-    def get_current_state(self) -> State:
+    def get_current_state(self) -> Optional[State]:
         if self.__current_state is None:
-            raise ValueError("Getting current state before the machine has one.")
+            return None # The machine has not started yet
         return self.__states[self.__current_state]
 
     def process_input(self, i: Input) -> None:
+        if self.__start_state is None:
+            raise ValueError("Processing input without a start state.")
         if self.__current_state is None:
-            if self.__start_state is None:
-                raise ValueError("Processing input without a start state.")
             self.__current_state = self.__start_state
-        # Get all transitions for the current state
+        hs = hash(self.__current_state)
-        #transition = self.__transitions[(self.__current_state, i)]
+        # dict[hash_origin, dict[hash_transition_input, Transition]]
-        #self.__current_state = transition.get_next_state()
+        if hs not in self.__transitions.keys():
+            raise ValueError("Processing input for a state with no transitions.")
+        transitions = self.__transitions[hs]
+        hi = hash(i)
+        transition = None
+        if hi in transitions.keys():
+            # Transition found: execute and move to the destination state
+            transition = transitions[hi]
+        # No transition found, check if there's a special transition
+        if transition is None:
+            hc = hash(TransitionInputEnum.MATCH_REST)
+            if hc in transitions.keys():
+                transition = transitions[hc]
+            else:
+                raise ValueError(f"Found invalid transition while processing input {i} [{transition}].]")
+        transition.execute_output(self.__context)
+        self.__current_state = transition.get_destination_name_hash()
 
     def __str__(self) -> str:
         r = [f"=MACHINE=", "\tSTATES"]

test/test_mealy.py

@@ -7,4 +7,15 @@ class TestMachine(unittest.TestCase):
 
     def test_init(self):
         m = Machine()
-        self.assertIsNotNone(m)
+        self.assertIsNotNone(m)
+
+    def test_add_state_two_start_states(self):
+        m = Machine()
+        m.add_state("A", {}, True, False)
+        self.assertRaises(ValueError, m.add_state, "B", {}, True, False)
+
+    def test_add_two_end_states(self):
+        m = Machine()
+        m.add_state("A", {}, True, False)
+        m.add_state("B", {}, False, True)
+        m.add_state("C", {}, False, True)