Comments (2)
I think i found a solution by myself how work within the python-statemachine in combination with ros2.
import rclpy
from rclpy.node import Node
from rclpy.executors import MultiThreadedExecutor
from std_msgs.msg import String
from statemachine import StateMachine, State
class MyStateMachine(StateMachine):
state_a = State(initial=True)
state_b = State()
state_c = State()
trigger = state_a.to(state_b,cond='condition_success') |state_a.to(state_c,cond='condition_success2') | state_b.to(state_c) | state_c.to(state_a)
class StateMachineNode(Node):
def __init__(self):
super().__init__('state_machine_node')
self.myfsm = MyStateMachine(self)
self.subscription = self.create_subscription(
String,
'state_transition',
self.state_transition_callback,
10)
def condition_success(self):
return self.new_state == 'StateA'
def condition_success2(self):
return self.new_state == 'StateB'
def state_transition_callback(self, msg):
self.new_state = msg.data
if self.new_state in ['StateA', 'StateB', 'StateC']:
self.myfsm.send('trigger')
self.get_logger().info(f'Transitioning trigger got it {self.new_state}')
def on_enter_state_a(self):
self.get_logger().info('Executing function for StateA')
def on_enter_state_b(self):
self.get_logger().info('Executing function for StateB')
def on_enter_state_c(self):
self.get_logger().info('Executing function for StateC')
def main(args=None):
rclpy.init(args=args)
node = StateMachineNode()
#fsm = MyStateMachine(node)
executor = MultiThreadedExecutor()
executor.add_node(node)
try:
node.get_logger().info('State machine node running')
executor.spin()
except KeyboardInterrupt:
node.get_logger().info('Shutting down')
rclpy.shutdown()
if __name__ == '__main__':
main()
This implementation allows me to use the trigger form a ros2 topic within the state machine. So my full functionality can be in the ros2 node.
from python-statemachine.
Nice @slyandsmart . After reading your solution the fix was to "send" to the state machine the event name, not the name of the target state. Right? In your case, the event name is trigger
.
Changing this line;
self.my_fsm.send(new_state)
By this one:
self.myfsm.send('trigger')
Also, I think that you'll better served if the code surrounding the state machine don't know anything about the internal states.
So if you could refactor the event loop to receive the performed actions instead the desired state, your design will improve:
Something like this:
def state_transition_callback(self, msg):
event = msg.data
if event in self.myfsm.allowed_events:
self.myfsm.send(event)
self.get_logger().info(f'Transitioning trigger got it {event}. Current state is {self.myfsm.current_state.id}')
As you've resolved the problem, I'll close the issue for now. Feel free to reopen it or open another one.
from python-statemachine.
Related Issues (20)
- Generate the code from PlantUML diagrams HOT 2
- Type annotations missing for `initial_state` and `final_states`; leading to linting errors with Pylint/Pyright HOT 3
- Expensive instantiation of StateMachine HOT 3
- AttributeError: object has no attribute 'model' HOT 3
- Enable Multithreading HOT 2
- [Feature request] State machine based workflow for database storage on entity state tracking HOT 1
- Action callback gets called twice when mixing definition methods HOT 1
- Check that all state transitions can reach a final state. HOT 6
- Execute a state machine from another state machine HOT 1
- Guards and conditions passed as strings really necessary? HOT 6
- Send a state machine over MQTT / serializing-deserializing a SM HOT 6
- Question: How can I use cond functions to gatekeep, but not fail? HOT 5
- Failed to generate a graphic image HOT 2
- Models instantiated in Djangos setUpTestData seem to skip conditions HOT 5
- @<transition>.cond decorator bug, general question about running code on every transition even if TransitionNotAllowed is raised HOT 3
- Persistence, timeouts, and branching conditionals? HOT 7
- feature request: return to previous state
- sm.current_state.name capitalizes the state name and replaces underscores with spaces HOT 2
- Introduction of asyncio causes RuntimeError exception in multi-threaded application HOT 3
Recommend Projects
-
React
A declarative, efficient, and flexible JavaScript library for building user interfaces.
-
Vue.js
🖖 Vue.js is a progressive, incrementally-adoptable JavaScript framework for building UI on the web.
-
Typescript
TypeScript is a superset of JavaScript that compiles to clean JavaScript output.
-
TensorFlow
An Open Source Machine Learning Framework for Everyone
-
Django
The Web framework for perfectionists with deadlines.
-
Laravel
A PHP framework for web artisans
-
D3
Bring data to life with SVG, Canvas and HTML. 📊📈🎉
-
Recommend Topics
-
javascript
JavaScript (JS) is a lightweight interpreted programming language with first-class functions.
-
web
Some thing interesting about web. New door for the world.
-
server
A server is a program made to process requests and deliver data to clients.
-
Machine learning
Machine learning is a way of modeling and interpreting data that allows a piece of software to respond intelligently.
-
Visualization
Some thing interesting about visualization, use data art
-
Game
Some thing interesting about game, make everyone happy.
Recommend Org
-
Facebook
We are working to build community through open source technology. NB: members must have two-factor auth.
-
Microsoft
Open source projects and samples from Microsoft.
-
Google
Google ❤️ Open Source for everyone.
-
Alibaba
Alibaba Open Source for everyone
-
D3
Data-Driven Documents codes.
-
Tencent
China tencent open source team.
from python-statemachine.