Tutorial: FSM¶
An FSM (finite state machine) is coord-dsl’s smallest coordination unit. This tutorial models one, generates it, and runs it — with a full Python section and a full C++ section. They share the same model and the same event-loop semantics; only the runtime language differs.
Modelling¶
Following the FSM design in Prof. Herman Bruyninckx’s Composable and
Explainable Systems of Systems, an FSM is represented as a pure data
structure. A .fsm file contains:
states — the stateful behaviours, including a
startand anendstate;an event loop — named events representing occurrences or monitored state changes to which the machine reacts;
transitions — directed
from/torelationships between states; andreactions — the policy relating an event to a transition and, optionally, further events to fire.
Each reaction matches one event from the referenced event loop. Event compositions from the broader FSM design are not implemented.
The control loop and behaviour implementations are separate from the model.
Event-loop syntax.
Declare an event loop before the FSM. It has its own namespace and name; each
comma-separated event is introduced by evt:
evt loop (ns=ex) el {
evt START,
evt STOP
}
The FSM selects one declared loop with evt loop: <el>. Event references use
the loop-qualified name in both when and fires:
R_START {
when: <el.START>,
do: <T_START>,
fires { <el.STOP> }
}
Within the same scope, e.g. in the same FSM declaration, references can be
direct, e.g., start: <IDLE> or from: <IDLE>, to: <GRASPING>.
ns ex = "http://example.org/"
evt loop (ns=ex) el {
evt E-CONFIGURE-ENTERED,
evt E-CONFIGURE-EXIT,
evt E_IDLE_ENTERED,
evt E_IDLE_EXIT_EXECUTE,
evt E_IDLE_EXIT_COMPILE,
evt E-COMPILE-ENTERED,
evt E-COMPILE-EXIT,
evt E_EXECUTE_ENTERED,
evt E_EXECUTE_EXIT,
evt e-step,
evt e-exit
}
fsm (ns=ex) ex_fsm {
description: "Example of a simple FSM"
states {
S_START,
S-CONFIGURE,
S_IDLE,
S-COMPILE,
S_EXECUTE,
S_EXIT
}
evt loop: <el>
start: <S_START>
end: <S_EXIT>
transitions {
T_START_CONFIGURE { from: <S_START>, to: <S-CONFIGURE> },
T_CONFIGURE_IDLE { from: <S-CONFIGURE>, to: <S_IDLE> },
T_IDLE_IDLE { from: <S_IDLE>, to: <S_IDLE> },
T_IDLE_EXECUTE { from: <S_IDLE>, to: <S_EXECUTE> },
T_IDLE_COMPILE { from: <S_IDLE>, to: <S-COMPILE> },
T_COMPILE_EXECUTE { from: <S-COMPILE>, to: <S_EXECUTE> },
T_EXECUTE_EXECUTE { from: <S_EXECUTE>, to: <S_EXECUTE> },
T_EXECUTE_IDLE { from: <S_EXECUTE>, to: <S_IDLE> },
T_IDLE_EXIT { from: <S_IDLE>, to: <S_EXIT> }
}
reactions {
R_E_CONFIGURE_EXIT {
when: <el.E-CONFIGURE-EXIT>,
do: <T_CONFIGURE_IDLE>,
fires { <el.E_IDLE_ENTERED> }
},
R_E_IDLE_EXIT_EXECUTE {
when: <el.E_IDLE_EXIT_EXECUTE>,
do: <T_IDLE_EXECUTE>,
fires { <el.E_EXECUTE_ENTERED> }
},
R_E_IDLE_EXIT_COMPILE {
when: <el.E_IDLE_EXIT_COMPILE>,
do: <T_IDLE_COMPILE>,
fires { <el.E-COMPILE-ENTERED> }
},
R_E_COMPILE_EXIT {
when: <el.E-COMPILE-EXIT>,
do: <T_COMPILE_EXECUTE>,
fires { <el.E_EXECUTE_ENTERED> }
},
R_E_EXECUTE_EXIT {
when: <el.E_EXECUTE_EXIT>,
do: <T_EXECUTE_IDLE>,
fires { <el.E_IDLE_ENTERED> }
},
R_E_EXIT { when: <el.e-exit>, do: <T_IDLE_EXIT> },
R_E_STEP1 {
when: <el.e-step>,
do: <T_START_CONFIGURE>,
fires { <el.E-CONFIGURE-ENTERED>, <el.e-step> }
},
R_E_STEP2 { when: <el.e-step>, do: <T_IDLE_IDLE> },
R_E_STEP3 { when: <el.e-step>, do: <T_EXECUTE_EXECUTE> }
}
}
Two rules matter:
Reactions are ordered. On each step the first reaction whose event is present and whose transition starts from the current state is taken; the rest are ignored that step.
During graph generation, the
namespace(ns) is combined with the FSM and declaration names to create their full URIs.
Generate¶
textx generate example.fsm --target python -o ex_fsm.py
textx generate example.fsm --target cpp -o ex_fsm.hpp
textx generate example.fsm --target graph --format json-ld --autocompact # writes ex_fsm.ld.json beside example.fsm
textx generate example.fsm --target dot --format png # a picture of the machine
The graph and console targets call rdflib.Graph.serialize with
format from --format (default: json-ld), indent=2, and
auto_compact=True when --autocompact is present (False otherwise).
RDFLib derives the compacted JSON-LD context from the graph’s namespace
manager.
The code targets emit create_fsm(), destroy_fsm() (C++), the
state/event/transition/reaction enums, and the IRI tables below. No control loop —
you own that. Each also writes a provenance.ld.json beside the artifact,
recording what produced it.
Model IRIs in the generated code¶
Every entity the model names keeps its IRI in the generated runtime, so a running
machine can identify itself and its parts against the RDF graph — the same IRIs
the graph target serialises:
Symbol |
Names |
|---|---|
|
the machine itself |
|
indexed by |
|
indexed by |
from ex_fsm import FSM_URI, STATE_URIS, StateID, create_fsm
fsm = create_fsm()
print(FSM_URI) # which machine
print(STATE_URIS[StateID(fsm.current_state_index)]) # which state it is in
The C++ header carries the same names, as static constexpr const char *
arrays indexed by the enums.
Running in Python¶
The Python runtime is coord_dsl.event_loop (the buffer ops) and
coord_dsl.fsm (fsm_step). The generated module gives you EventID and
StateID enums, create_fsm() and STATE_URIS.
The loop is: run your behaviour (produce/consume events), fsm_step, then
reconfig_event_buffers:
from coord_dsl.event_loop import produce_event, consume_event, reconfig_event_buffers
from coord_dsl.fsm import fsm_step
from ex_fsm import EventID, StateID, create_fsm, STATE_URIS
fsm = create_fsm()
while fsm.current_state_index != StateID.S_EXIT:
behavior(fsm) # YOUR code: produce/consume events
fsm_step(fsm) # apply the first matching reaction
reconfig_event_buffers(fsm.event_data)
fsm.current_state_index is the live state; fsm.event_data holds the
buffers. You “define behaviour” by writing behavior — typically it inspects
current_state_index and, when a state’s work is done, produces the event
that advances the machine. A complete, time-driven controller ships with the
repo:
#!/usr/bin/env python3
@dataclass
class UserData:
current_time: float
state_duration: float
compile: bool = True
cycles: int = 0 # idle visits so far
max_cycles: int = 0 # 0 keeps the machine running until interrupted
transition_time: float | None = None
def __post_init__(self):
if self.transition_time is None:
self.transition_time = self.current_time + self.state_duration
def generic_on_end(fsm: FSMData, ud: UserData, end_events: list[EventID]):
del ud
# print(f"State '{StateID(fsm.current_state_index).name}' finished")
for evt in end_events:
produce_event(fsm.event_data, evt)
def idle_on_end(fsm: FSMData, ud: UserData):
ud.cycles += 1
if ud.max_cycles and ud.cycles >= ud.max_cycles:
# leave through the machine's end state rather than being killed
generic_on_end(fsm, ud, [EventID.E_EXIT])
return
if ud.compile:
generic_on_end(fsm, ud, [EventID.E_IDLE_EXIT_COMPILE])
else:
generic_on_end(fsm, ud, [EventID.E_IDLE_EXIT_EXECUTE])
# Toggle compile flag for next time
ud.compile = not ud.compile
def generic_step(fsm: FSMData, ud: UserData, start_event: EventID) -> bool:
"""Return True if timeout has occurred, i.e., state finished."""
if consume_event(fsm.event_data, start_event):
print(f"State: {StateID(fsm.current_state_index).name} ({STATE_URIS[StateID(fsm.current_state_index)]})")
ud.current_time = time.time()
assert ud.transition_time is not None
if ud.current_time < ud.transition_time:
return False
# ensure loop period
while ud.transition_time < ud.current_time:
ud.transition_time += ud.state_duration
return True
Run it:
textx generate example.fsm --target python -o ex_fsm.py
python generated_fsm_bhv.py # cycles until Ctrl+C
python generated_fsm_bhv.py --cycles 3 # fires e-exit, ends in S_EXIT
The second form is what makes the loop above terminate: the controller produces
e-exit after three idle visits, and R_E_EXIT — which is ordered above
the e-step reactions, so it wins while both events are present — takes the
machine to its end state.
Running in C++¶
The C++ header is self-contained and depends on coord2b. The API mirrors
Python one-to-one (produce_event / consume_event /
reconfig_event_buffers / fsm_step_nbx), and the loop has the same shape:
auto fsm = ex_fsm::create_fsm();
bool compile = true;
printf("Starting generated FSM example. Press Ctrl+C to exit.\n\n");
while (!stopFlag) {
next_tick += tick_period;
produce_event(fsm->eventData, ex_fsm::E_STEP);
// run state machine, event loop
fsm_step_nbx(fsm);
reconfig_event_buffers(fsm->eventData);
// handle print
now = clock::now();
if (now > next_print) {
next_print += print_period;
std::cout << "State: " << fsm->states[fsm->currentStateIndex].name << " ("
<< ex_fsm::STATE_URIS[fsm->currentStateIndex] << ")" << std::endl;
if (fsm->currentStateIndex == ex_fsm::S_CONFIGURE) {
produce_event(fsm->eventData, ex_fsm::E_CONFIGURE_EXIT);
} else if (fsm->currentStateIndex == ex_fsm::S_IDLE) {
if (compile) {
produce_event(fsm->eventData, ex_fsm::E_IDLE_EXIT_COMPILE);
} else {
produce_event(fsm->eventData, ex_fsm::E_IDLE_EXIT_EXECUTE);
}
compile = !compile;
} else if (fsm->currentStateIndex == ex_fsm::S_COMPILE) {
produce_event(fsm->eventData, ex_fsm::E_COMPILE_EXIT);
} else if (fsm->currentStateIndex == ex_fsm::S_EXECUTE) {
produce_event(fsm->eventData, ex_fsm::E_EXECUTE_EXIT);
}
}
std::this_thread::sleep_until(next_tick);
}
fsm->currentStateIndex and fsm->states[i].name read the live machine;
fsm->eventData holds the buffers. Build against coord2b with CMake:
cmake_minimum_required(VERSION 3.20)
project(fsm_example)
find_package(coord2b REQUIRED)
include_directories(
${coord2b_INCLUDE_DIRS}
)
add_executable(fsm_test
test_fsm.cpp
ex_fsm.hpp
)
target_compile_features(fsm_test PUBLIC
c_std_99 cxx_std_17 cxx_std_20
)
target_link_libraries(fsm_test
coord2b
)
cmake -S . -B build && cmake --build build
./build/fsm_test