Implement FSM core components and example usage

examples/config_example.yml

@@ -1,26 +1,30 @@
 name: "SimpleFSM"
 states:
   Idle:
-    on_enter: "Entering Idle"
-    on_exit: "Exiting Idle"
+    on_enter: "Entering Idle hehe"
+    on_exit: "Exiting Idle hehe"
   Active:
-    on_enter: "Entering Active"
-    on_exit: "Exiting Active"
+    on_enter: "Entering Active ehehe"
+    on_exit: "Exiting Active ehehe"
 transitions:
   - source: "Idle"
     target: "Active"
     condition: "lambda: True"
-    action: "lambda: print('Activating')"
+    action: "print('====Activating====')"
   - source: "Active"
     target: "Idle"
     condition: "lambda: True"
-    action: "lambda: print('Deactivating')"
+    action: "print('====Deactivating====')"
 events:
   Activate:
     transitions:
       - source: "Idle"
         target: "Active"
+    terminal:
+      - "Active"
   Deactivate:
     transitions:
       - source: "Active"
         target: "Idle"
+    terminal:
+      - "Idle"

examples/workflow_example.py

@@ -0,0 +1,15 @@
+from ..flowfsm.config.parser import parse_fsm_config
+from ..flowfsm.config.loader import load_fsm_from_config
+from ..flowfsm.runtime.executor import Executor
+
+# Load FSM configuration
+config = parse_fsm_config("./config_example.yml")
+
+# Create FSM
+workflow = load_fsm_from_config(config)
+
+print(f"Workflow '{workflow.name}' initialized with states: {workflow.states}")
+
+# Execute FSM
+executor = Executor(workflow)
+executor.run()

flow-fsm/config/loader.py → flowfsm/config/loader.py

@@ -12,7 +12,7 @@ def create_states(states_config):
     for state_name, state_config in states_config.items():
         on_enter = state_config.get("on_enter")
         on_exit = state_config.get("on_exit")
-        state = State(state_name, on_enter=on_enter, on_exit=on_exit)
+        state = StateRegistry.register(state_name, on_enter=on_enter, on_exit=on_exit)
         states[state_name] = state
     return states
 
@@ -23,15 +23,17 @@ def create_transitions(transitions_config):
         source, target = transition["source"], transition["target"]
         condition = transition.get("condition")
         action = transition.get("action")
-        transition = Transition(source, target, condition=condition, action=action)
+        source = StateRegistry.get(source) if source in StateRegistry._states else StateRegistry.register(source)
+        target = StateRegistry.get(target) if target in StateRegistry._states else StateRegistry.register(target)
+        transition = TransitionRegistry.register(source, target, condition=condition, action=action)
         transitions.append(transition)
     return transitions
 
 def create_events(events_config, transitions):
     """Dynamically create events and bind them to transitions."""
     events = {}
     for event_name, event_config in events_config.items():
-        event = Event(event_name)
+        event = EventRegistry.register(event_name)
         for transition in event_config["transitions"]:
             # Find the transition object based on source and target
             for t in transitions:
@@ -43,14 +45,14 @@ def create_events(events_config, transitions):
 
 def load_fsm_from_config(config):
     """Create FSM components from the configuration file."""
+    # Clear existing registries to avoid conflicts
+    StateRegistry.clear()
+    TransitionRegistry.clear()
+    EventRegistry.clear()
     states = create_states(config['states'])
     transitions = create_transitions(config['transitions'])
     events = create_events(config['events'], transitions)
 
-    workflow = Workflow(config["name"], {
-        "states": list(states.keys()),
-        "transitions": [(t.source, t.target, {"condition": t.is_valid, "action": t.execute}) for t in transitions],
-        "events": list(events.keys())
-    })
+    workflow = Workflow(config["name"], list(states.items()), transitions, events)
 
     return workflow

flow-fsm/core/base.py → flowfsm/core/base.py

@@ -13,11 +13,11 @@ def __new__(cls, name, bases, dct):
     @staticmethod
     def generate_trigger_event_method():
         def trigger_event(self, event):
-            transition = event.trigger(self.current_state)
+            transition = event.trigger(self, self.current_state)
             if transition:
-                self.current_state.exit()
+                self.current_state().exit()
                 transition.execute()
-                self.current_state = StateRegistry.get(transition.target)()
+                self.current_state = StateRegistry.get(transition.target.name)()
                 self.current_state.enter()
             else:
                 raise InvalidTransitionError(f"No valid transition for event '{event}' from state '{self.current_state}'.")

flow-fsm/core/event.py → flowfsm/core/event.py

@@ -3,14 +3,16 @@ class EventRegistry:
     _events = {}
 
     @classmethod
-    def register(cls, name):
+    def register(cls, name, terminal_states=None):
+        """Register events and optionally define terminal states."""
         if name in cls._events:
             raise ValueError(f"Event '{name}' is already registered.")
 
-        def create_event_class(name):
-            """Dynamically creates an event class."""
+        def create_event_class(name, terminal_states):
+            """Dynamically creates an event class with terminal states."""
             def __init__(self):
                 self.transitions = []
+                self.terminal_states = terminal_states or []  # Default to an empty list if no terminal states are provided
 
             def add_transition(self, transition):
                 self.transitions.append(transition)
@@ -29,26 +31,37 @@ def trigger(self, current_state):
             }
             return type(name, (object,), methods)
 
-        event_class = create_event_class(name)
+        # Create and register the event class with terminal states
+        event_class = create_event_class(name, terminal_states)
         cls._events[name] = event_class
         return event_class
 
     @classmethod
     def get(cls, name):
+        """Retrieve an event class by name."""
         if name not in cls._events:
             raise ValueError(f"Event '{name}' is not registered.")
         return cls._events[name]
+
+    @classmethod
+    def clear(cls):
+        """Clear all registered events."""
+        cls._events = {}
 
 
 class Event:
     """User API for creating and managing events."""
-    def __init__(self, name):
+    def __init__(self, name, terminal_states=None):
         self.name = name
-        self._event_class = EventRegistry.register(name)
+        self.terminal_states = terminal_states
+        self._event_class = EventRegistry.register(name, terminal_states)
         self._event_instance = self._event_class()
 
     def __getattr__(self, attr):
+        """Delegate attribute access to the event instance."""
         return getattr(self._event_instance, attr)
 
     def __repr__(self):
+        """Represent the event instance."""
         return repr(self._event_instance)
+

flow-fsm/core/state.py → flowfsm/core/state.py

@@ -7,16 +7,25 @@ class StateRegistry:
     def register(cls, name, on_enter=None, on_exit=None):
         def create_state_class(name, on_enter=None, on_exit=None):
             """Dynamically creates a state class with custom behavior."""
-            def default_enter(self):
+            def enter(self):
+                if on_enter:
+                    print(on_enter)
+                    return
                 print(f"Entering {name}")
 
-            def default_exit(self):
+            def exit(self):
+                if on_exit:
+                    print(on_exit)
+                    return
                 print(f"Exiting {name}")
 
             methods = {
-                "enter": on_enter or default_enter,
-                "exit": on_exit or default_exit,
+                "enter": enter,
+                "exit": exit,
                 "__repr__": lambda self: f"<State: {name}>",
+                "name": name,
+                "on_enter": on_enter,
+                "on_exit": on_exit,
             }
             return type(name, (object,), methods)
 
@@ -32,6 +41,11 @@ def get(cls, name):
         if name not in cls._states:
             raise ValueError(f"State '{name}' is not registered.")
         return cls._states[name]
+
+    @classmethod
+    def clear(cls):
+        """Clear all registered states."""
+        cls._states = {}
 
 
 # User-facing State API

flow-fsm/core/transition.py → flowfsm/core/transition.py

@@ -6,19 +6,21 @@ class TransitionRegistry:
     def register(cls, source, target, condition=None, action=None):
         def create_transition_class(source, target, condition=None, action=None):
             """Dynamically creates a transition class."""
-            def is_valid(self):
-                return condition() if condition else True
+            def is_valid(*args):
+                return eval(condition) if condition else True
 
-            def execute(self):
+            def execute():
                 if action:
-                    action()
+                    eval(action)
 
             methods = {
                 "is_valid": is_valid,
                 "execute": execute,
                 "__repr__": lambda self: f"<Transition: {source} -> {target}>",
                 "source": source,
                 "target": target,
+                "condition": condition,
+                "action": action,
             }
             return type(f"Transition_{source}_to_{target}", (object,), methods)
 
@@ -30,6 +32,11 @@ def execute(self):
     def get_all(cls):
         """Retrieve all registered transitions."""
         return cls._transitions
+
+    @classmethod
+    def clear(cls):
+        """Clear all registered transitions."""
+        cls._transitions = []
 
 
 class Transition:
@@ -42,4 +49,4 @@ def __getattr__(self, attr):
         return getattr(self._transition_instance, attr)
 
     def __repr__(self):
-        return repr(self._transition_instance)
+        return repr(self._transition_instance)

flowfsm/core/workflow.py

@@ -0,0 +1,16 @@
+from .base import FSMBase
+from .state import StateRegistry
+from .transition import Transition
+from .event import EventRegistry
+
+
+class Workflow(FSMBase):
+    """Workflow class extending FSMBase."""
+    def __init__(self, name, states, transitions, events):
+        super().__init__(name)
+
+        self.states = states
+        self.current_state = self.states[0][1]
+        self.current_state().enter()
+        self.transitions = transitions
+        self.events = events

flowfsm/runtime/executor.py

@@ -0,0 +1,16 @@
+class Executor:
+    """The executor runs the FSM based on events and transitions."""
+    def __init__(self, workflow):
+        self.workflow = workflow
+
+    def run(self):
+        """Run the FSM, processing events and triggering transitions."""
+        print(f"Starting FSM: {self.workflow.name}")
+        while True:
+            # Wait for events to trigger transitions
+            event_name = "Activate" #input("Enter event name: ").strip()
+            if event_name in self.workflow.events:
+                event = self.workflow.events[event_name]
+                self.workflow.trigger_event(event)
+            else:
+                print(f"Event '{event_name}' not found.")

flowfsm/utils/__init__.py

@@ -0,0 +1,5 @@
+def print_params_decorator(func):
+    def wrapper(*args, **kwargs):
+        print(f"Parameters passed to {func.__name__}: args={args}, kwargs={kwargs}")
+        return func(*args, **kwargs)
+    return wrapper

test.py

@@ -0,0 +1,14 @@
+from flowfsm.config.parser import parse_fsm_config
+from flowfsm.config.loader import load_fsm_from_config
+from flowfsm.runtime.executor import Executor
+
+
+# Load FSM configuration
+config = parse_fsm_config("./examples/config_example.yml")
+
+# Create FSM
+workflow = load_fsm_from_config(config)
+
+# Execute FSM
+executor = Executor(workflow)
+executor.run()