add exercises
This commit is contained in:
parent
8a24549cdb
commit
2fb8318506
11 changed files with 2706 additions and 0 deletions
241
StartingPoint/srcgen/a.py
Normal file
241
StartingPoint/srcgen/a.py
Normal file
|
|
@ -0,0 +1,241 @@
|
|||
"""Implementation of statechart a.
|
||||
Generated by itemis CREATE code generator.
|
||||
"""
|
||||
|
||||
import queue
|
||||
import sys, os
|
||||
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '../lib')))
|
||||
from yakindu.rx import Observable
|
||||
|
||||
class A:
|
||||
"""Implementation of the state machine A.
|
||||
"""
|
||||
|
||||
class State:
|
||||
""" State Enum
|
||||
"""
|
||||
(
|
||||
main_region_state_a,
|
||||
null_state
|
||||
) = range(2)
|
||||
|
||||
|
||||
def __init__(self):
|
||||
""" Declares all necessary variables including list of states, histories etc.
|
||||
"""
|
||||
|
||||
self.x = None
|
||||
self.x_observable = Observable()
|
||||
self.y = None
|
||||
self.y_observable = Observable()
|
||||
|
||||
self.in_event_queue = queue.Queue()
|
||||
# enumeration of all states:
|
||||
self.__State = A.State
|
||||
self.__state_conf_vector_changed = None
|
||||
self.__state_vector = [None] * 1
|
||||
for __state_index in range(1):
|
||||
self.__state_vector[__state_index] = self.State.null_state
|
||||
|
||||
# for timed statechart:
|
||||
self.timer_service = None
|
||||
self.__time_events = [None] * 2
|
||||
|
||||
# initializations:
|
||||
self.__is_executing = False
|
||||
|
||||
def is_active(self):
|
||||
"""Checks if the state machine is active.
|
||||
"""
|
||||
return self.__state_vector[0] is not self.__State.null_state
|
||||
|
||||
def is_final(self):
|
||||
"""Checks if the statemachine is final.
|
||||
Always returns 'false' since this state machine can never become final.
|
||||
"""
|
||||
return False
|
||||
|
||||
def is_state_active(self, state):
|
||||
"""Checks if the state is currently active.
|
||||
"""
|
||||
s = state
|
||||
if s == self.__State.main_region_state_a:
|
||||
return self.__state_vector[0] == self.__State.main_region_state_a
|
||||
return False
|
||||
|
||||
def time_elapsed(self, event_id):
|
||||
"""Add time events to in event queue
|
||||
"""
|
||||
if event_id in range(2):
|
||||
self.in_event_queue.put(lambda: self.raise_time_event(event_id))
|
||||
self.run_cycle()
|
||||
|
||||
def raise_time_event(self, event_id):
|
||||
"""Raise timed events using the event_id.
|
||||
"""
|
||||
self.__time_events[event_id] = True
|
||||
|
||||
def __execute_queued_event(self, func):
|
||||
func()
|
||||
|
||||
def __get_next_event(self):
|
||||
if not self.in_event_queue.empty():
|
||||
return self.in_event_queue.get()
|
||||
return None
|
||||
|
||||
def __entry_action_main_region_state_a(self):
|
||||
"""Entry action for state 'StateA'..
|
||||
"""
|
||||
#Entry action for state 'StateA'.
|
||||
self.timer_service.set_timer(self, 0, (1 * 1000), False)
|
||||
self.timer_service.set_timer(self, 1, (2 * 1000), False)
|
||||
|
||||
def __exit_action_main_region_state_a(self):
|
||||
"""Exit action for state 'StateA'..
|
||||
"""
|
||||
#Exit action for state 'StateA'.
|
||||
self.timer_service.unset_timer(self, 0)
|
||||
self.timer_service.unset_timer(self, 1)
|
||||
|
||||
def __enter_sequence_main_region_state_a_default(self):
|
||||
"""'default' enter sequence for state StateA.
|
||||
"""
|
||||
#'default' enter sequence for state StateA
|
||||
self.__entry_action_main_region_state_a()
|
||||
self.__state_vector[0] = self.State.main_region_state_a
|
||||
self.__state_conf_vector_changed = True
|
||||
|
||||
def __enter_sequence_main_region_default(self):
|
||||
"""'default' enter sequence for region main region.
|
||||
"""
|
||||
#'default' enter sequence for region main region
|
||||
self.__react_main_region__entry_default()
|
||||
|
||||
def __exit_sequence_main_region_state_a(self):
|
||||
"""Default exit sequence for state StateA.
|
||||
"""
|
||||
#Default exit sequence for state StateA
|
||||
self.__state_vector[0] = self.State.null_state
|
||||
self.__exit_action_main_region_state_a()
|
||||
|
||||
def __exit_sequence_main_region(self):
|
||||
"""Default exit sequence for region main region.
|
||||
"""
|
||||
#Default exit sequence for region main region
|
||||
state = self.__state_vector[0]
|
||||
if state == self.State.main_region_state_a:
|
||||
self.__exit_sequence_main_region_state_a()
|
||||
|
||||
def __react_main_region__entry_default(self):
|
||||
"""Default react sequence for initial entry .
|
||||
"""
|
||||
#Default react sequence for initial entry
|
||||
self.__enter_sequence_main_region_state_a_default()
|
||||
|
||||
def __react(self, transitioned_before):
|
||||
"""Implementation of __react function.
|
||||
"""
|
||||
#State machine reactions.
|
||||
return transitioned_before
|
||||
|
||||
|
||||
def __main_region_state_a_react(self, transitioned_before):
|
||||
"""Implementation of __main_region_state_a_react function.
|
||||
"""
|
||||
#The reactions of state StateA.
|
||||
transitioned_after = transitioned_before
|
||||
if transitioned_after < 0:
|
||||
if self.__time_events[0]:
|
||||
self.__exit_sequence_main_region_state_a()
|
||||
self.x_observable.next()
|
||||
self.__time_events[0] = False
|
||||
self.__enter_sequence_main_region_state_a_default()
|
||||
self.__react(0)
|
||||
transitioned_after = 0
|
||||
elif self.__time_events[1]:
|
||||
self.__exit_sequence_main_region_state_a()
|
||||
self.y_observable.next()
|
||||
self.__time_events[1] = False
|
||||
self.__enter_sequence_main_region_state_a_default()
|
||||
self.__react(0)
|
||||
transitioned_after = 0
|
||||
#If no transition was taken
|
||||
if transitioned_after == transitioned_before:
|
||||
#then execute local reactions.
|
||||
transitioned_after = self.__react(transitioned_before)
|
||||
return transitioned_after
|
||||
|
||||
|
||||
def __clear_in_events(self):
|
||||
"""Implementation of __clear_in_events function.
|
||||
"""
|
||||
self.__time_events[0] = False
|
||||
self.__time_events[1] = False
|
||||
|
||||
|
||||
def __micro_step(self):
|
||||
"""Implementation of __micro_step function.
|
||||
"""
|
||||
state = self.__state_vector[0]
|
||||
if state == self.State.main_region_state_a:
|
||||
self.__main_region_state_a_react(-1)
|
||||
|
||||
|
||||
def run_cycle(self):
|
||||
"""Implementation of run_cycle function.
|
||||
"""
|
||||
#Performs a 'run to completion' step.
|
||||
if self.timer_service is None:
|
||||
raise ValueError('Timer service must be set.')
|
||||
|
||||
if self.__is_executing:
|
||||
return
|
||||
self.__is_executing = True
|
||||
next_event = self.__get_next_event()
|
||||
if next_event is not None:
|
||||
self.__execute_queued_event(next_event)
|
||||
condition_0 = True
|
||||
while condition_0:
|
||||
self.__micro_step()
|
||||
self.__clear_in_events()
|
||||
condition_0 = False
|
||||
next_event = self.__get_next_event()
|
||||
if next_event is not None:
|
||||
self.__execute_queued_event(next_event)
|
||||
condition_0 = True
|
||||
self.__is_executing = False
|
||||
|
||||
|
||||
def enter(self):
|
||||
"""Implementation of enter function.
|
||||
"""
|
||||
#Activates the state machine.
|
||||
if self.timer_service is None:
|
||||
raise ValueError('Timer service must be set.')
|
||||
|
||||
if self.__is_executing:
|
||||
return
|
||||
self.__is_executing = True
|
||||
#Default enter sequence for statechart A
|
||||
self.__enter_sequence_main_region_default()
|
||||
self.__is_executing = False
|
||||
|
||||
|
||||
def exit(self):
|
||||
"""Implementation of exit function.
|
||||
"""
|
||||
#Deactivates the state machine.
|
||||
if self.__is_executing:
|
||||
return
|
||||
self.__is_executing = True
|
||||
#Default exit sequence for statechart A
|
||||
self.__exit_sequence_main_region()
|
||||
self.__state_vector[0] = self.State.null_state
|
||||
self.__is_executing = False
|
||||
|
||||
|
||||
def trigger_without_event(self):
|
||||
"""Implementation of triggerWithoutEvent function.
|
||||
"""
|
||||
self.run_cycle()
|
||||
|
||||
306
StartingPoint/srcgen/b.py
Normal file
306
StartingPoint/srcgen/b.py
Normal file
|
|
@ -0,0 +1,306 @@
|
|||
"""Implementation of statechart b.
|
||||
Generated by itemis CREATE code generator.
|
||||
"""
|
||||
|
||||
import queue
|
||||
import sys, os
|
||||
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '../lib')))
|
||||
from yakindu.rx import Observable
|
||||
|
||||
class B:
|
||||
"""Implementation of the state machine B.
|
||||
"""
|
||||
|
||||
class State:
|
||||
""" State Enum
|
||||
"""
|
||||
(
|
||||
main_region_outer,
|
||||
main_region_outer_r1inner,
|
||||
null_state
|
||||
) = range(3)
|
||||
|
||||
|
||||
def __init__(self):
|
||||
""" Declares all necessary variables including list of states, histories etc.
|
||||
"""
|
||||
|
||||
self.outer = None
|
||||
self.outer_observable = Observable()
|
||||
self.inner = None
|
||||
self.inner_observable = Observable()
|
||||
|
||||
self.in_event_queue = queue.Queue()
|
||||
# enumeration of all states:
|
||||
self.__State = B.State
|
||||
self.__state_conf_vector_changed = None
|
||||
self.__state_vector = [None] * 1
|
||||
for __state_index in range(1):
|
||||
self.__state_vector[__state_index] = self.State.null_state
|
||||
|
||||
# for timed statechart:
|
||||
self.timer_service = None
|
||||
self.__time_events = [None] * 2
|
||||
|
||||
# initializations:
|
||||
self.__is_executing = False
|
||||
|
||||
def is_active(self):
|
||||
"""Checks if the state machine is active.
|
||||
"""
|
||||
return self.__state_vector[0] is not self.__State.null_state
|
||||
|
||||
def is_final(self):
|
||||
"""Checks if the statemachine is final.
|
||||
Always returns 'false' since this state machine can never become final.
|
||||
"""
|
||||
return False
|
||||
|
||||
def is_state_active(self, state):
|
||||
"""Checks if the state is currently active.
|
||||
"""
|
||||
s = state
|
||||
if s == self.__State.main_region_outer:
|
||||
return (self.__state_vector[0] >= self.__State.main_region_outer)\
|
||||
and (self.__state_vector[0] <= self.__State.main_region_outer_r1inner)
|
||||
if s == self.__State.main_region_outer_r1inner:
|
||||
return self.__state_vector[0] == self.__State.main_region_outer_r1inner
|
||||
return False
|
||||
|
||||
def time_elapsed(self, event_id):
|
||||
"""Add time events to in event queue
|
||||
"""
|
||||
if event_id in range(2):
|
||||
self.in_event_queue.put(lambda: self.raise_time_event(event_id))
|
||||
self.run_cycle()
|
||||
|
||||
def raise_time_event(self, event_id):
|
||||
"""Raise timed events using the event_id.
|
||||
"""
|
||||
self.__time_events[event_id] = True
|
||||
|
||||
def __execute_queued_event(self, func):
|
||||
func()
|
||||
|
||||
def __get_next_event(self):
|
||||
if not self.in_event_queue.empty():
|
||||
return self.in_event_queue.get()
|
||||
return None
|
||||
|
||||
def __entry_action_main_region_outer(self):
|
||||
"""Entry action for state 'Outer'..
|
||||
"""
|
||||
#Entry action for state 'Outer'.
|
||||
self.timer_service.set_timer(self, 0, (3 * 1000), False)
|
||||
|
||||
def __entry_action_main_region_outer_r1_inner(self):
|
||||
"""Entry action for state 'Inner'..
|
||||
"""
|
||||
#Entry action for state 'Inner'.
|
||||
self.timer_service.set_timer(self, 1, (2 * 1000), False)
|
||||
|
||||
def __exit_action_main_region_outer(self):
|
||||
"""Exit action for state 'Outer'..
|
||||
"""
|
||||
#Exit action for state 'Outer'.
|
||||
self.timer_service.unset_timer(self, 0)
|
||||
|
||||
def __exit_action_main_region_outer_r1_inner(self):
|
||||
"""Exit action for state 'Inner'..
|
||||
"""
|
||||
#Exit action for state 'Inner'.
|
||||
self.timer_service.unset_timer(self, 1)
|
||||
|
||||
def __enter_sequence_main_region_outer_default(self):
|
||||
"""'default' enter sequence for state Outer.
|
||||
"""
|
||||
#'default' enter sequence for state Outer
|
||||
self.__entry_action_main_region_outer()
|
||||
self.__enter_sequence_main_region_outer_r1_default()
|
||||
|
||||
def __enter_sequence_main_region_outer_r1_inner_default(self):
|
||||
"""'default' enter sequence for state Inner.
|
||||
"""
|
||||
#'default' enter sequence for state Inner
|
||||
self.__entry_action_main_region_outer_r1_inner()
|
||||
self.__state_vector[0] = self.State.main_region_outer_r1inner
|
||||
self.__state_conf_vector_changed = True
|
||||
|
||||
def __enter_sequence_main_region_default(self):
|
||||
"""'default' enter sequence for region main region.
|
||||
"""
|
||||
#'default' enter sequence for region main region
|
||||
self.__react_main_region__entry_default()
|
||||
|
||||
def __enter_sequence_main_region_outer_r1_default(self):
|
||||
"""'default' enter sequence for region r1.
|
||||
"""
|
||||
#'default' enter sequence for region r1
|
||||
self.__react_main_region_outer_r1__entry_default()
|
||||
|
||||
def __exit_sequence_main_region_outer(self):
|
||||
"""Default exit sequence for state Outer.
|
||||
"""
|
||||
#Default exit sequence for state Outer
|
||||
self.__exit_sequence_main_region_outer_r1()
|
||||
self.__state_vector[0] = self.State.null_state
|
||||
self.__exit_action_main_region_outer()
|
||||
|
||||
def __exit_sequence_main_region_outer_r1_inner(self):
|
||||
"""Default exit sequence for state Inner.
|
||||
"""
|
||||
#Default exit sequence for state Inner
|
||||
self.__state_vector[0] = self.State.main_region_outer
|
||||
self.__exit_action_main_region_outer_r1_inner()
|
||||
|
||||
def __exit_sequence_main_region(self):
|
||||
"""Default exit sequence for region main region.
|
||||
"""
|
||||
#Default exit sequence for region main region
|
||||
state = self.__state_vector[0]
|
||||
if state == self.State.main_region_outer:
|
||||
self.__exit_sequence_main_region_outer()
|
||||
elif state == self.State.main_region_outer_r1inner:
|
||||
self.__exit_sequence_main_region_outer_r1_inner()
|
||||
self.__exit_action_main_region_outer()
|
||||
|
||||
def __exit_sequence_main_region_outer_r1(self):
|
||||
"""Default exit sequence for region r1.
|
||||
"""
|
||||
#Default exit sequence for region r1
|
||||
state = self.__state_vector[0]
|
||||
if state == self.State.main_region_outer_r1inner:
|
||||
self.__exit_sequence_main_region_outer_r1_inner()
|
||||
|
||||
def __react_main_region_outer_r1__entry_default(self):
|
||||
"""Default react sequence for initial entry .
|
||||
"""
|
||||
#Default react sequence for initial entry
|
||||
self.__enter_sequence_main_region_outer_r1_inner_default()
|
||||
|
||||
def __react_main_region__entry_default(self):
|
||||
"""Default react sequence for initial entry .
|
||||
"""
|
||||
#Default react sequence for initial entry
|
||||
self.__enter_sequence_main_region_outer_default()
|
||||
|
||||
def __react(self, transitioned_before):
|
||||
"""Implementation of __react function.
|
||||
"""
|
||||
#State machine reactions.
|
||||
return transitioned_before
|
||||
|
||||
|
||||
def __main_region_outer_react(self, transitioned_before):
|
||||
"""Implementation of __main_region_outer_react function.
|
||||
"""
|
||||
#The reactions of state Outer.
|
||||
transitioned_after = transitioned_before
|
||||
if transitioned_after < 0:
|
||||
if self.__time_events[0]:
|
||||
self.__exit_sequence_main_region_outer()
|
||||
self.outer_observable.next()
|
||||
self.__time_events[0] = False
|
||||
self.__enter_sequence_main_region_outer_default()
|
||||
self.__react(0)
|
||||
transitioned_after = 0
|
||||
#If no transition was taken
|
||||
if transitioned_after == transitioned_before:
|
||||
#then execute local reactions.
|
||||
transitioned_after = self.__react(transitioned_before)
|
||||
return transitioned_after
|
||||
|
||||
|
||||
def __main_region_outer_r1_inner_react(self, transitioned_before):
|
||||
"""Implementation of __main_region_outer_r1_inner_react function.
|
||||
"""
|
||||
#The reactions of state Inner.
|
||||
transitioned_after = transitioned_before
|
||||
if transitioned_after < 0:
|
||||
if self.__time_events[1]:
|
||||
self.__exit_sequence_main_region_outer_r1_inner()
|
||||
self.inner_observable.next()
|
||||
self.__time_events[1] = False
|
||||
self.__enter_sequence_main_region_outer_r1_inner_default()
|
||||
self.__main_region_outer_react(0)
|
||||
transitioned_after = 0
|
||||
#If no transition was taken
|
||||
if transitioned_after == transitioned_before:
|
||||
#then execute local reactions.
|
||||
transitioned_after = self.__main_region_outer_react(transitioned_before)
|
||||
return transitioned_after
|
||||
|
||||
|
||||
def __clear_in_events(self):
|
||||
"""Implementation of __clear_in_events function.
|
||||
"""
|
||||
self.__time_events[0] = False
|
||||
self.__time_events[1] = False
|
||||
|
||||
|
||||
def __micro_step(self):
|
||||
"""Implementation of __micro_step function.
|
||||
"""
|
||||
state = self.__state_vector[0]
|
||||
if state == self.State.main_region_outer_r1inner:
|
||||
self.__main_region_outer_r1_inner_react(-1)
|
||||
|
||||
|
||||
def run_cycle(self):
|
||||
"""Implementation of run_cycle function.
|
||||
"""
|
||||
#Performs a 'run to completion' step.
|
||||
if self.timer_service is None:
|
||||
raise ValueError('Timer service must be set.')
|
||||
|
||||
if self.__is_executing:
|
||||
return
|
||||
self.__is_executing = True
|
||||
next_event = self.__get_next_event()
|
||||
if next_event is not None:
|
||||
self.__execute_queued_event(next_event)
|
||||
condition_0 = True
|
||||
while condition_0:
|
||||
self.__micro_step()
|
||||
self.__clear_in_events()
|
||||
condition_0 = False
|
||||
next_event = self.__get_next_event()
|
||||
if next_event is not None:
|
||||
self.__execute_queued_event(next_event)
|
||||
condition_0 = True
|
||||
self.__is_executing = False
|
||||
|
||||
|
||||
def enter(self):
|
||||
"""Implementation of enter function.
|
||||
"""
|
||||
#Activates the state machine.
|
||||
if self.timer_service is None:
|
||||
raise ValueError('Timer service must be set.')
|
||||
|
||||
if self.__is_executing:
|
||||
return
|
||||
self.__is_executing = True
|
||||
#Default enter sequence for statechart B
|
||||
self.__enter_sequence_main_region_default()
|
||||
self.__is_executing = False
|
||||
|
||||
|
||||
def exit(self):
|
||||
"""Implementation of exit function.
|
||||
"""
|
||||
#Deactivates the state machine.
|
||||
if self.__is_executing:
|
||||
return
|
||||
self.__is_executing = True
|
||||
#Default exit sequence for statechart B
|
||||
self.__exit_sequence_main_region()
|
||||
self.__state_vector[0] = self.State.null_state
|
||||
self.__is_executing = False
|
||||
|
||||
|
||||
def trigger_without_event(self):
|
||||
"""Implementation of triggerWithoutEvent function.
|
||||
"""
|
||||
self.run_cycle()
|
||||
|
||||
332
StartingPoint/srcgen/c.py
Normal file
332
StartingPoint/srcgen/c.py
Normal file
|
|
@ -0,0 +1,332 @@
|
|||
"""Implementation of statechart c.
|
||||
Generated by itemis CREATE code generator.
|
||||
"""
|
||||
|
||||
import queue
|
||||
import sys, os
|
||||
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '../lib')))
|
||||
|
||||
class C:
|
||||
"""Implementation of the state machine C.
|
||||
"""
|
||||
|
||||
class State:
|
||||
""" State Enum
|
||||
"""
|
||||
(
|
||||
main_region_temp,
|
||||
main_region_zero,
|
||||
main_region_one,
|
||||
main_region_initial,
|
||||
null_state
|
||||
) = range(5)
|
||||
|
||||
|
||||
def __init__(self):
|
||||
""" Declares all necessary variables including list of states, histories etc.
|
||||
"""
|
||||
|
||||
self.x = None
|
||||
|
||||
self.in_event_queue = queue.Queue()
|
||||
# enumeration of all states:
|
||||
self.__State = C.State
|
||||
self.__state_conf_vector_changed = None
|
||||
self.__state_vector = [None] * 1
|
||||
for __state_index in range(1):
|
||||
self.__state_vector[__state_index] = self.State.null_state
|
||||
|
||||
# for timed statechart:
|
||||
self.timer_service = None
|
||||
self.__time_events = [None] * 1
|
||||
|
||||
# initializations:
|
||||
#Default init sequence for statechart C
|
||||
self.x = 0
|
||||
self.__is_executing = False
|
||||
|
||||
def is_active(self):
|
||||
"""Checks if the state machine is active.
|
||||
"""
|
||||
return self.__state_vector[0] is not self.__State.null_state
|
||||
|
||||
def is_final(self):
|
||||
"""Checks if the statemachine is final.
|
||||
Always returns 'false' since this state machine can never become final.
|
||||
"""
|
||||
return False
|
||||
|
||||
def is_state_active(self, state):
|
||||
"""Checks if the state is currently active.
|
||||
"""
|
||||
s = state
|
||||
if s == self.__State.main_region_temp:
|
||||
return self.__state_vector[0] == self.__State.main_region_temp
|
||||
if s == self.__State.main_region_zero:
|
||||
return self.__state_vector[0] == self.__State.main_region_zero
|
||||
if s == self.__State.main_region_one:
|
||||
return self.__state_vector[0] == self.__State.main_region_one
|
||||
if s == self.__State.main_region_initial:
|
||||
return self.__state_vector[0] == self.__State.main_region_initial
|
||||
return False
|
||||
|
||||
def time_elapsed(self, event_id):
|
||||
"""Add time events to in event queue
|
||||
"""
|
||||
if event_id in range(1):
|
||||
self.in_event_queue.put(lambda: self.raise_time_event(event_id))
|
||||
self.run_cycle()
|
||||
|
||||
def raise_time_event(self, event_id):
|
||||
"""Raise timed events using the event_id.
|
||||
"""
|
||||
self.__time_events[event_id] = True
|
||||
|
||||
def __execute_queued_event(self, func):
|
||||
func()
|
||||
|
||||
def __get_next_event(self):
|
||||
if not self.in_event_queue.empty():
|
||||
return self.in_event_queue.get()
|
||||
return None
|
||||
|
||||
def __entry_action_main_region_initial(self):
|
||||
"""Entry action for state 'Initial'..
|
||||
"""
|
||||
#Entry action for state 'Initial'.
|
||||
self.timer_service.set_timer(self, 0, (1 * 1000), False)
|
||||
|
||||
def __exit_action_main_region_initial(self):
|
||||
"""Exit action for state 'Initial'..
|
||||
"""
|
||||
#Exit action for state 'Initial'.
|
||||
self.timer_service.unset_timer(self, 0)
|
||||
|
||||
def __enter_sequence_main_region_temp_default(self):
|
||||
"""'default' enter sequence for state Temp.
|
||||
"""
|
||||
#'default' enter sequence for state Temp
|
||||
self.__state_vector[0] = self.State.main_region_temp
|
||||
self.__state_conf_vector_changed = True
|
||||
|
||||
def __enter_sequence_main_region_zero_default(self):
|
||||
"""'default' enter sequence for state Zero.
|
||||
"""
|
||||
#'default' enter sequence for state Zero
|
||||
self.__state_vector[0] = self.State.main_region_zero
|
||||
self.__state_conf_vector_changed = True
|
||||
|
||||
def __enter_sequence_main_region_one_default(self):
|
||||
"""'default' enter sequence for state One.
|
||||
"""
|
||||
#'default' enter sequence for state One
|
||||
self.__state_vector[0] = self.State.main_region_one
|
||||
self.__state_conf_vector_changed = True
|
||||
|
||||
def __enter_sequence_main_region_initial_default(self):
|
||||
"""'default' enter sequence for state Initial.
|
||||
"""
|
||||
#'default' enter sequence for state Initial
|
||||
self.__entry_action_main_region_initial()
|
||||
self.__state_vector[0] = self.State.main_region_initial
|
||||
self.__state_conf_vector_changed = True
|
||||
|
||||
def __enter_sequence_main_region_default(self):
|
||||
"""'default' enter sequence for region main region.
|
||||
"""
|
||||
#'default' enter sequence for region main region
|
||||
self.__react_main_region__entry_default()
|
||||
|
||||
def __exit_sequence_main_region_temp(self):
|
||||
"""Default exit sequence for state Temp.
|
||||
"""
|
||||
#Default exit sequence for state Temp
|
||||
self.__state_vector[0] = self.State.null_state
|
||||
|
||||
def __exit_sequence_main_region_zero(self):
|
||||
"""Default exit sequence for state Zero.
|
||||
"""
|
||||
#Default exit sequence for state Zero
|
||||
self.__state_vector[0] = self.State.null_state
|
||||
|
||||
def __exit_sequence_main_region_one(self):
|
||||
"""Default exit sequence for state One.
|
||||
"""
|
||||
#Default exit sequence for state One
|
||||
self.__state_vector[0] = self.State.null_state
|
||||
|
||||
def __exit_sequence_main_region_initial(self):
|
||||
"""Default exit sequence for state Initial.
|
||||
"""
|
||||
#Default exit sequence for state Initial
|
||||
self.__state_vector[0] = self.State.null_state
|
||||
self.__exit_action_main_region_initial()
|
||||
|
||||
def __exit_sequence_main_region(self):
|
||||
"""Default exit sequence for region main region.
|
||||
"""
|
||||
#Default exit sequence for region main region
|
||||
state = self.__state_vector[0]
|
||||
if state == self.State.main_region_temp:
|
||||
self.__exit_sequence_main_region_temp()
|
||||
elif state == self.State.main_region_zero:
|
||||
self.__exit_sequence_main_region_zero()
|
||||
elif state == self.State.main_region_one:
|
||||
self.__exit_sequence_main_region_one()
|
||||
elif state == self.State.main_region_initial:
|
||||
self.__exit_sequence_main_region_initial()
|
||||
|
||||
def __react_main_region__entry_default(self):
|
||||
"""Default react sequence for initial entry .
|
||||
"""
|
||||
#Default react sequence for initial entry
|
||||
self.__enter_sequence_main_region_initial_default()
|
||||
|
||||
def __react(self, transitioned_before):
|
||||
"""Implementation of __react function.
|
||||
"""
|
||||
#State machine reactions.
|
||||
return transitioned_before
|
||||
|
||||
|
||||
def __main_region_temp_react(self, transitioned_before):
|
||||
"""Implementation of __main_region_temp_react function.
|
||||
"""
|
||||
#The reactions of state Temp.
|
||||
transitioned_after = transitioned_before
|
||||
if transitioned_after < 0:
|
||||
if self.x == 0:
|
||||
self.__exit_sequence_main_region_temp()
|
||||
self.__enter_sequence_main_region_zero_default()
|
||||
self.__react(0)
|
||||
transitioned_after = 0
|
||||
elif self.x == 1:
|
||||
self.__exit_sequence_main_region_temp()
|
||||
self.__enter_sequence_main_region_one_default()
|
||||
self.__react(0)
|
||||
transitioned_after = 0
|
||||
#If no transition was taken
|
||||
if transitioned_after == transitioned_before:
|
||||
#then execute local reactions.
|
||||
transitioned_after = self.__react(transitioned_before)
|
||||
return transitioned_after
|
||||
|
||||
|
||||
def __main_region_zero_react(self, transitioned_before):
|
||||
"""Implementation of __main_region_zero_react function.
|
||||
"""
|
||||
#The reactions of state Zero.
|
||||
transitioned_after = transitioned_before
|
||||
#Always execute local reactions.
|
||||
transitioned_after = self.__react(transitioned_before)
|
||||
return transitioned_after
|
||||
|
||||
|
||||
def __main_region_one_react(self, transitioned_before):
|
||||
"""Implementation of __main_region_one_react function.
|
||||
"""
|
||||
#The reactions of state One.
|
||||
transitioned_after = transitioned_before
|
||||
#Always execute local reactions.
|
||||
transitioned_after = self.__react(transitioned_before)
|
||||
return transitioned_after
|
||||
|
||||
|
||||
def __main_region_initial_react(self, transitioned_before):
|
||||
"""Implementation of __main_region_initial_react function.
|
||||
"""
|
||||
#The reactions of state Initial.
|
||||
transitioned_after = transitioned_before
|
||||
if transitioned_after < 0:
|
||||
if self.__time_events[0]:
|
||||
self.__exit_sequence_main_region_initial()
|
||||
self.x = self.x + 1
|
||||
self.__time_events[0] = False
|
||||
self.__enter_sequence_main_region_temp_default()
|
||||
self.__react(0)
|
||||
transitioned_after = 0
|
||||
#If no transition was taken
|
||||
if transitioned_after == transitioned_before:
|
||||
#then execute local reactions.
|
||||
transitioned_after = self.__react(transitioned_before)
|
||||
return transitioned_after
|
||||
|
||||
|
||||
def __clear_in_events(self):
|
||||
"""Implementation of __clear_in_events function.
|
||||
"""
|
||||
self.__time_events[0] = False
|
||||
|
||||
|
||||
def __micro_step(self):
|
||||
"""Implementation of __micro_step function.
|
||||
"""
|
||||
state = self.__state_vector[0]
|
||||
if state == self.State.main_region_temp:
|
||||
self.__main_region_temp_react(-1)
|
||||
elif state == self.State.main_region_zero:
|
||||
self.__main_region_zero_react(-1)
|
||||
elif state == self.State.main_region_one:
|
||||
self.__main_region_one_react(-1)
|
||||
elif state == self.State.main_region_initial:
|
||||
self.__main_region_initial_react(-1)
|
||||
|
||||
|
||||
def run_cycle(self):
|
||||
"""Implementation of run_cycle function.
|
||||
"""
|
||||
#Performs a 'run to completion' step.
|
||||
if self.timer_service is None:
|
||||
raise ValueError('Timer service must be set.')
|
||||
|
||||
if self.__is_executing:
|
||||
return
|
||||
self.__is_executing = True
|
||||
next_event = self.__get_next_event()
|
||||
if next_event is not None:
|
||||
self.__execute_queued_event(next_event)
|
||||
condition_0 = True
|
||||
while condition_0:
|
||||
self.__micro_step()
|
||||
self.__clear_in_events()
|
||||
condition_0 = False
|
||||
next_event = self.__get_next_event()
|
||||
if next_event is not None:
|
||||
self.__execute_queued_event(next_event)
|
||||
condition_0 = True
|
||||
self.__is_executing = False
|
||||
|
||||
|
||||
def enter(self):
|
||||
"""Implementation of enter function.
|
||||
"""
|
||||
#Activates the state machine.
|
||||
if self.timer_service is None:
|
||||
raise ValueError('Timer service must be set.')
|
||||
|
||||
if self.__is_executing:
|
||||
return
|
||||
self.__is_executing = True
|
||||
#Default enter sequence for statechart C
|
||||
self.__enter_sequence_main_region_default()
|
||||
self.__is_executing = False
|
||||
|
||||
|
||||
def exit(self):
|
||||
"""Implementation of exit function.
|
||||
"""
|
||||
#Deactivates the state machine.
|
||||
if self.__is_executing:
|
||||
return
|
||||
self.__is_executing = True
|
||||
#Default exit sequence for statechart C
|
||||
self.__exit_sequence_main_region()
|
||||
self.__state_vector[0] = self.State.null_state
|
||||
self.__is_executing = False
|
||||
|
||||
|
||||
def trigger_without_event(self):
|
||||
"""Implementation of triggerWithoutEvent function.
|
||||
"""
|
||||
self.run_cycle()
|
||||
|
||||
464
StartingPoint/srcgen/d.py
Normal file
464
StartingPoint/srcgen/d.py
Normal file
|
|
@ -0,0 +1,464 @@
|
|||
"""Implementation of statechart d.
|
||||
Generated by itemis CREATE code generator.
|
||||
"""
|
||||
|
||||
import queue
|
||||
import sys, os
|
||||
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '../lib')))
|
||||
|
||||
class D:
|
||||
"""Implementation of the state machine D.
|
||||
"""
|
||||
|
||||
class State:
|
||||
""" State Enum
|
||||
"""
|
||||
(
|
||||
main_region_orthogonal,
|
||||
main_region_orthogonal_r1temp,
|
||||
main_region_orthogonal_r1one,
|
||||
main_region_orthogonal_r1initial,
|
||||
main_region_orthogonal_r1zero,
|
||||
main_region_orthogonal_r2s,
|
||||
main_region_orthogonal_r2t,
|
||||
null_state
|
||||
) = range(8)
|
||||
|
||||
|
||||
def __init__(self):
|
||||
""" Declares all necessary variables including list of states, histories etc.
|
||||
"""
|
||||
|
||||
self.x = None
|
||||
|
||||
self.in_event_queue = queue.Queue()
|
||||
# enumeration of all states:
|
||||
self.__State = D.State
|
||||
self.__state_conf_vector_changed = None
|
||||
self.__state_vector = [None] * 2
|
||||
for __state_index in range(2):
|
||||
self.__state_vector[__state_index] = self.State.null_state
|
||||
|
||||
# for timed statechart:
|
||||
self.timer_service = None
|
||||
self.__time_events = [None] * 2
|
||||
|
||||
# initializations:
|
||||
#Default init sequence for statechart D
|
||||
self.x = 0
|
||||
self.__is_executing = False
|
||||
self.__state_conf_vector_position = None
|
||||
|
||||
def is_active(self):
|
||||
"""Checks if the state machine is active.
|
||||
"""
|
||||
return self.__state_vector[0] is not self.__State.null_state or self.__state_vector[1] is not self.__State.null_state
|
||||
|
||||
def is_final(self):
|
||||
"""Checks if the statemachine is final.
|
||||
Always returns 'false' since this state machine can never become final.
|
||||
"""
|
||||
return False
|
||||
|
||||
def is_state_active(self, state):
|
||||
"""Checks if the state is currently active.
|
||||
"""
|
||||
s = state
|
||||
if s == self.__State.main_region_orthogonal:
|
||||
return (self.__state_vector[0] >= self.__State.main_region_orthogonal)\
|
||||
and (self.__state_vector[0] <= self.__State.main_region_orthogonal_r2t)
|
||||
if s == self.__State.main_region_orthogonal_r1temp:
|
||||
return self.__state_vector[0] == self.__State.main_region_orthogonal_r1temp
|
||||
if s == self.__State.main_region_orthogonal_r1one:
|
||||
return self.__state_vector[0] == self.__State.main_region_orthogonal_r1one
|
||||
if s == self.__State.main_region_orthogonal_r1initial:
|
||||
return self.__state_vector[0] == self.__State.main_region_orthogonal_r1initial
|
||||
if s == self.__State.main_region_orthogonal_r1zero:
|
||||
return self.__state_vector[0] == self.__State.main_region_orthogonal_r1zero
|
||||
if s == self.__State.main_region_orthogonal_r2s:
|
||||
return self.__state_vector[1] == self.__State.main_region_orthogonal_r2s
|
||||
if s == self.__State.main_region_orthogonal_r2t:
|
||||
return self.__state_vector[1] == self.__State.main_region_orthogonal_r2t
|
||||
return False
|
||||
|
||||
def time_elapsed(self, event_id):
|
||||
"""Add time events to in event queue
|
||||
"""
|
||||
if event_id in range(2):
|
||||
self.in_event_queue.put(lambda: self.raise_time_event(event_id))
|
||||
self.run_cycle()
|
||||
|
||||
def raise_time_event(self, event_id):
|
||||
"""Raise timed events using the event_id.
|
||||
"""
|
||||
self.__time_events[event_id] = True
|
||||
|
||||
def __execute_queued_event(self, func):
|
||||
func()
|
||||
|
||||
def __get_next_event(self):
|
||||
if not self.in_event_queue.empty():
|
||||
return self.in_event_queue.get()
|
||||
return None
|
||||
|
||||
def __entry_action_main_region_orthogonal_r1_initial(self):
|
||||
"""Entry action for state 'Initial'..
|
||||
"""
|
||||
#Entry action for state 'Initial'.
|
||||
self.timer_service.set_timer(self, 0, (1 * 1000), False)
|
||||
|
||||
def __entry_action_main_region_orthogonal_r2_s(self):
|
||||
"""Entry action for state 'S'..
|
||||
"""
|
||||
#Entry action for state 'S'.
|
||||
self.timer_service.set_timer(self, 1, (5 * 1000), False)
|
||||
|
||||
def __exit_action_main_region_orthogonal_r1_initial(self):
|
||||
"""Exit action for state 'Initial'..
|
||||
"""
|
||||
#Exit action for state 'Initial'.
|
||||
self.timer_service.unset_timer(self, 0)
|
||||
|
||||
def __exit_action_main_region_orthogonal_r2_s(self):
|
||||
"""Exit action for state 'S'..
|
||||
"""
|
||||
#Exit action for state 'S'.
|
||||
self.timer_service.unset_timer(self, 1)
|
||||
|
||||
def __enter_sequence_main_region_orthogonal_default(self):
|
||||
"""'default' enter sequence for state Orthogonal.
|
||||
"""
|
||||
#'default' enter sequence for state Orthogonal
|
||||
self.__enter_sequence_main_region_orthogonal_r1_default()
|
||||
self.__enter_sequence_main_region_orthogonal_r2_default()
|
||||
|
||||
def __enter_sequence_main_region_orthogonal_r1_temp_default(self):
|
||||
"""'default' enter sequence for state Temp.
|
||||
"""
|
||||
#'default' enter sequence for state Temp
|
||||
self.__state_vector[0] = self.State.main_region_orthogonal_r1temp
|
||||
self.__state_conf_vector_position = 0
|
||||
self.__state_conf_vector_changed = True
|
||||
|
||||
def __enter_sequence_main_region_orthogonal_r1_one_default(self):
|
||||
"""'default' enter sequence for state One.
|
||||
"""
|
||||
#'default' enter sequence for state One
|
||||
self.__state_vector[0] = self.State.main_region_orthogonal_r1one
|
||||
self.__state_conf_vector_position = 0
|
||||
self.__state_conf_vector_changed = True
|
||||
|
||||
def __enter_sequence_main_region_orthogonal_r1_initial_default(self):
|
||||
"""'default' enter sequence for state Initial.
|
||||
"""
|
||||
#'default' enter sequence for state Initial
|
||||
self.__entry_action_main_region_orthogonal_r1_initial()
|
||||
self.__state_vector[0] = self.State.main_region_orthogonal_r1initial
|
||||
self.__state_conf_vector_position = 0
|
||||
self.__state_conf_vector_changed = True
|
||||
|
||||
def __enter_sequence_main_region_orthogonal_r1_zero_default(self):
|
||||
"""'default' enter sequence for state Zero.
|
||||
"""
|
||||
#'default' enter sequence for state Zero
|
||||
self.__state_vector[0] = self.State.main_region_orthogonal_r1zero
|
||||
self.__state_conf_vector_position = 0
|
||||
self.__state_conf_vector_changed = True
|
||||
|
||||
def __enter_sequence_main_region_orthogonal_r2_s_default(self):
|
||||
"""'default' enter sequence for state S.
|
||||
"""
|
||||
#'default' enter sequence for state S
|
||||
self.__entry_action_main_region_orthogonal_r2_s()
|
||||
self.__state_vector[1] = self.State.main_region_orthogonal_r2s
|
||||
self.__state_conf_vector_position = 1
|
||||
self.__state_conf_vector_changed = True
|
||||
|
||||
def __enter_sequence_main_region_orthogonal_r2_t_default(self):
|
||||
"""'default' enter sequence for state T.
|
||||
"""
|
||||
#'default' enter sequence for state T
|
||||
self.__state_vector[1] = self.State.main_region_orthogonal_r2t
|
||||
self.__state_conf_vector_position = 1
|
||||
self.__state_conf_vector_changed = True
|
||||
|
||||
def __enter_sequence_main_region_default(self):
|
||||
"""'default' enter sequence for region main region.
|
||||
"""
|
||||
#'default' enter sequence for region main region
|
||||
self.__react_main_region__entry_default()
|
||||
|
||||
def __enter_sequence_main_region_orthogonal_r1_default(self):
|
||||
"""'default' enter sequence for region r1.
|
||||
"""
|
||||
#'default' enter sequence for region r1
|
||||
self.__react_main_region_orthogonal_r1__entry_default()
|
||||
|
||||
def __enter_sequence_main_region_orthogonal_r2_default(self):
|
||||
"""'default' enter sequence for region r2.
|
||||
"""
|
||||
#'default' enter sequence for region r2
|
||||
self.__react_main_region_orthogonal_r2__entry_default()
|
||||
|
||||
def __exit_sequence_main_region_orthogonal_r1_temp(self):
|
||||
"""Default exit sequence for state Temp.
|
||||
"""
|
||||
#Default exit sequence for state Temp
|
||||
self.__state_vector[0] = self.State.main_region_orthogonal
|
||||
self.__state_conf_vector_position = 0
|
||||
|
||||
def __exit_sequence_main_region_orthogonal_r1_one(self):
|
||||
"""Default exit sequence for state One.
|
||||
"""
|
||||
#Default exit sequence for state One
|
||||
self.__state_vector[0] = self.State.main_region_orthogonal
|
||||
self.__state_conf_vector_position = 0
|
||||
|
||||
def __exit_sequence_main_region_orthogonal_r1_initial(self):
|
||||
"""Default exit sequence for state Initial.
|
||||
"""
|
||||
#Default exit sequence for state Initial
|
||||
self.__state_vector[0] = self.State.main_region_orthogonal
|
||||
self.__state_conf_vector_position = 0
|
||||
self.__exit_action_main_region_orthogonal_r1_initial()
|
||||
|
||||
def __exit_sequence_main_region_orthogonal_r1_zero(self):
|
||||
"""Default exit sequence for state Zero.
|
||||
"""
|
||||
#Default exit sequence for state Zero
|
||||
self.__state_vector[0] = self.State.main_region_orthogonal
|
||||
self.__state_conf_vector_position = 0
|
||||
|
||||
def __exit_sequence_main_region_orthogonal_r2_s(self):
|
||||
"""Default exit sequence for state S.
|
||||
"""
|
||||
#Default exit sequence for state S
|
||||
self.__state_vector[1] = self.State.main_region_orthogonal
|
||||
self.__state_conf_vector_position = 1
|
||||
self.__exit_action_main_region_orthogonal_r2_s()
|
||||
|
||||
def __exit_sequence_main_region_orthogonal_r2_t(self):
|
||||
"""Default exit sequence for state T.
|
||||
"""
|
||||
#Default exit sequence for state T
|
||||
self.__state_vector[1] = self.State.main_region_orthogonal
|
||||
self.__state_conf_vector_position = 1
|
||||
|
||||
def __exit_sequence_main_region(self):
|
||||
"""Default exit sequence for region main region.
|
||||
"""
|
||||
#Default exit sequence for region main region
|
||||
state = self.__state_vector[0]
|
||||
if state == self.State.main_region_orthogonal_r1temp:
|
||||
self.__exit_sequence_main_region_orthogonal_r1_temp()
|
||||
elif state == self.State.main_region_orthogonal_r1one:
|
||||
self.__exit_sequence_main_region_orthogonal_r1_one()
|
||||
elif state == self.State.main_region_orthogonal_r1initial:
|
||||
self.__exit_sequence_main_region_orthogonal_r1_initial()
|
||||
elif state == self.State.main_region_orthogonal_r1zero:
|
||||
self.__exit_sequence_main_region_orthogonal_r1_zero()
|
||||
state = self.__state_vector[1]
|
||||
if state == self.State.main_region_orthogonal_r2s:
|
||||
self.__exit_sequence_main_region_orthogonal_r2_s()
|
||||
elif state == self.State.main_region_orthogonal_r2t:
|
||||
self.__exit_sequence_main_region_orthogonal_r2_t()
|
||||
|
||||
def __react_main_region_orthogonal_r1__entry_default(self):
|
||||
"""Default react sequence for initial entry .
|
||||
"""
|
||||
#Default react sequence for initial entry
|
||||
self.__enter_sequence_main_region_orthogonal_r1_initial_default()
|
||||
|
||||
def __react_main_region_orthogonal_r2__entry_default(self):
|
||||
"""Default react sequence for initial entry .
|
||||
"""
|
||||
#Default react sequence for initial entry
|
||||
self.__enter_sequence_main_region_orthogonal_r2_s_default()
|
||||
|
||||
def __react_main_region__entry_default(self):
|
||||
"""Default react sequence for initial entry .
|
||||
"""
|
||||
#Default react sequence for initial entry
|
||||
self.__enter_sequence_main_region_orthogonal_default()
|
||||
|
||||
def __react(self, transitioned_before):
|
||||
"""Implementation of __react function.
|
||||
"""
|
||||
#State machine reactions.
|
||||
return transitioned_before
|
||||
|
||||
|
||||
def __main_region_orthogonal_react(self, transitioned_before):
|
||||
"""Implementation of __main_region_orthogonal_react function.
|
||||
"""
|
||||
#The reactions of state Orthogonal.
|
||||
transitioned_after = transitioned_before
|
||||
#Always execute local reactions.
|
||||
transitioned_after = self.__react(transitioned_before)
|
||||
return transitioned_after
|
||||
|
||||
|
||||
def __main_region_orthogonal_r1_temp_react(self, transitioned_before):
|
||||
"""Implementation of __main_region_orthogonal_r1_temp_react function.
|
||||
"""
|
||||
#The reactions of state Temp.
|
||||
transitioned_after = transitioned_before
|
||||
if transitioned_after < 0:
|
||||
if self.x == 0:
|
||||
self.__exit_sequence_main_region_orthogonal_r1_temp()
|
||||
self.__enter_sequence_main_region_orthogonal_r1_zero_default()
|
||||
transitioned_after = 0
|
||||
elif self.x == 1:
|
||||
self.__exit_sequence_main_region_orthogonal_r1_temp()
|
||||
self.__enter_sequence_main_region_orthogonal_r1_one_default()
|
||||
transitioned_after = 0
|
||||
return transitioned_after
|
||||
|
||||
|
||||
def __main_region_orthogonal_r1_one_react(self, transitioned_before):
|
||||
"""Implementation of __main_region_orthogonal_r1_one_react function.
|
||||
"""
|
||||
#The reactions of state One.
|
||||
return transitioned_before
|
||||
|
||||
|
||||
def __main_region_orthogonal_r1_initial_react(self, transitioned_before):
|
||||
"""Implementation of __main_region_orthogonal_r1_initial_react function.
|
||||
"""
|
||||
#The reactions of state Initial.
|
||||
transitioned_after = transitioned_before
|
||||
if transitioned_after < 0:
|
||||
if self.__time_events[0]:
|
||||
self.__exit_sequence_main_region_orthogonal_r1_initial()
|
||||
self.x = self.x + 1
|
||||
self.__time_events[0] = False
|
||||
self.__enter_sequence_main_region_orthogonal_r1_temp_default()
|
||||
transitioned_after = 0
|
||||
return transitioned_after
|
||||
|
||||
|
||||
def __main_region_orthogonal_r1_zero_react(self, transitioned_before):
|
||||
"""Implementation of __main_region_orthogonal_r1_zero_react function.
|
||||
"""
|
||||
#The reactions of state Zero.
|
||||
return transitioned_before
|
||||
|
||||
|
||||
def __main_region_orthogonal_r2_s_react(self, transitioned_before):
|
||||
"""Implementation of __main_region_orthogonal_r2_s_react function.
|
||||
"""
|
||||
#The reactions of state S.
|
||||
transitioned_after = transitioned_before
|
||||
if transitioned_after < 1:
|
||||
if self.__time_events[1]:
|
||||
self.__exit_sequence_main_region_orthogonal_r2_s()
|
||||
self.__time_events[1] = False
|
||||
self.__enter_sequence_main_region_orthogonal_r2_t_default()
|
||||
self.__main_region_orthogonal_react(0)
|
||||
transitioned_after = 1
|
||||
#If no transition was taken
|
||||
if transitioned_after == transitioned_before:
|
||||
#then execute local reactions.
|
||||
transitioned_after = self.__main_region_orthogonal_react(transitioned_before)
|
||||
return transitioned_after
|
||||
|
||||
|
||||
def __main_region_orthogonal_r2_t_react(self, transitioned_before):
|
||||
"""Implementation of __main_region_orthogonal_r2_t_react function.
|
||||
"""
|
||||
#The reactions of state T.
|
||||
transitioned_after = transitioned_before
|
||||
#Always execute local reactions.
|
||||
transitioned_after = self.__main_region_orthogonal_react(transitioned_before)
|
||||
return transitioned_after
|
||||
|
||||
|
||||
def __clear_in_events(self):
|
||||
"""Implementation of __clear_in_events function.
|
||||
"""
|
||||
self.__time_events[0] = False
|
||||
self.__time_events[1] = False
|
||||
|
||||
|
||||
def __micro_step(self):
|
||||
"""Implementation of __micro_step function.
|
||||
"""
|
||||
transitioned = -1
|
||||
self.__state_conf_vector_position = 0
|
||||
state = self.__state_vector[0]
|
||||
if state == self.State.main_region_orthogonal_r1temp:
|
||||
transitioned = self.__main_region_orthogonal_r1_temp_react(transitioned)
|
||||
elif state == self.State.main_region_orthogonal_r1one:
|
||||
transitioned = self.__main_region_orthogonal_r1_one_react(transitioned)
|
||||
elif state == self.State.main_region_orthogonal_r1initial:
|
||||
transitioned = self.__main_region_orthogonal_r1_initial_react(transitioned)
|
||||
elif state == self.State.main_region_orthogonal_r1zero:
|
||||
transitioned = self.__main_region_orthogonal_r1_zero_react(transitioned)
|
||||
if self.__state_conf_vector_position < 1:
|
||||
state = self.__state_vector[1]
|
||||
if state == self.State.main_region_orthogonal_r2s:
|
||||
self.__main_region_orthogonal_r2_s_react(transitioned)
|
||||
elif state == self.State.main_region_orthogonal_r2t:
|
||||
self.__main_region_orthogonal_r2_t_react(transitioned)
|
||||
|
||||
|
||||
def run_cycle(self):
|
||||
"""Implementation of run_cycle function.
|
||||
"""
|
||||
#Performs a 'run to completion' step.
|
||||
if self.timer_service is None:
|
||||
raise ValueError('Timer service must be set.')
|
||||
|
||||
if self.__is_executing:
|
||||
return
|
||||
self.__is_executing = True
|
||||
next_event = self.__get_next_event()
|
||||
if next_event is not None:
|
||||
self.__execute_queued_event(next_event)
|
||||
condition_0 = True
|
||||
while condition_0:
|
||||
self.__micro_step()
|
||||
self.__clear_in_events()
|
||||
condition_0 = False
|
||||
next_event = self.__get_next_event()
|
||||
if next_event is not None:
|
||||
self.__execute_queued_event(next_event)
|
||||
condition_0 = True
|
||||
self.__is_executing = False
|
||||
|
||||
|
||||
def enter(self):
|
||||
"""Implementation of enter function.
|
||||
"""
|
||||
#Activates the state machine.
|
||||
if self.timer_service is None:
|
||||
raise ValueError('Timer service must be set.')
|
||||
|
||||
if self.__is_executing:
|
||||
return
|
||||
self.__is_executing = True
|
||||
#Default enter sequence for statechart D
|
||||
self.__enter_sequence_main_region_default()
|
||||
self.__is_executing = False
|
||||
|
||||
|
||||
def exit(self):
|
||||
"""Implementation of exit function.
|
||||
"""
|
||||
#Deactivates the state machine.
|
||||
if self.__is_executing:
|
||||
return
|
||||
self.__is_executing = True
|
||||
#Default exit sequence for statechart D
|
||||
self.__exit_sequence_main_region()
|
||||
self.__state_vector[0] = self.State.null_state
|
||||
self.__state_vector[1] = self.State.null_state
|
||||
self.__state_conf_vector_position = 1
|
||||
self.__is_executing = False
|
||||
|
||||
|
||||
def trigger_without_event(self):
|
||||
"""Implementation of triggerWithoutEvent function.
|
||||
"""
|
||||
self.run_cycle()
|
||||
|
||||
325
StartingPoint/srcgen/e.py
Normal file
325
StartingPoint/srcgen/e.py
Normal file
|
|
@ -0,0 +1,325 @@
|
|||
"""Implementation of statechart e.
|
||||
Generated by itemis CREATE code generator.
|
||||
"""
|
||||
|
||||
import queue
|
||||
import sys, os
|
||||
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '../lib')))
|
||||
from yakindu.rx import Observable
|
||||
|
||||
class E:
|
||||
"""Implementation of the state machine E.
|
||||
"""
|
||||
|
||||
class State:
|
||||
""" State Enum
|
||||
"""
|
||||
(
|
||||
main_region_orthogonal_state,
|
||||
main_region_orthogonal_state_r1state_a,
|
||||
main_region_orthogonal_state_r2state_b,
|
||||
null_state
|
||||
) = range(4)
|
||||
|
||||
|
||||
def __init__(self):
|
||||
""" Declares all necessary variables including list of states, histories etc.
|
||||
"""
|
||||
|
||||
self.v = None
|
||||
self.x = None
|
||||
self.x_observable = Observable()
|
||||
self.y = None
|
||||
self.y_observable = Observable()
|
||||
|
||||
self.in_event_queue = queue.Queue()
|
||||
# enumeration of all states:
|
||||
self.__State = E.State
|
||||
self.__state_conf_vector_changed = None
|
||||
self.__state_vector = [None] * 2
|
||||
for __state_index in range(2):
|
||||
self.__state_vector[__state_index] = self.State.null_state
|
||||
|
||||
# for timed statechart:
|
||||
self.timer_service = None
|
||||
self.__time_events = [None] * 1
|
||||
|
||||
# initializations:
|
||||
#Default init sequence for statechart E
|
||||
self.v = 0
|
||||
self.__is_executing = False
|
||||
self.__state_conf_vector_position = None
|
||||
|
||||
def is_active(self):
|
||||
"""Checks if the state machine is active.
|
||||
"""
|
||||
return self.__state_vector[0] is not self.__State.null_state or self.__state_vector[1] is not self.__State.null_state
|
||||
|
||||
def is_final(self):
|
||||
"""Checks if the statemachine is final.
|
||||
Always returns 'false' since this state machine can never become final.
|
||||
"""
|
||||
return False
|
||||
|
||||
def is_state_active(self, state):
|
||||
"""Checks if the state is currently active.
|
||||
"""
|
||||
s = state
|
||||
if s == self.__State.main_region_orthogonal_state:
|
||||
return (self.__state_vector[0] >= self.__State.main_region_orthogonal_state)\
|
||||
and (self.__state_vector[0] <= self.__State.main_region_orthogonal_state_r2state_b)
|
||||
if s == self.__State.main_region_orthogonal_state_r1state_a:
|
||||
return self.__state_vector[0] == self.__State.main_region_orthogonal_state_r1state_a
|
||||
if s == self.__State.main_region_orthogonal_state_r2state_b:
|
||||
return self.__state_vector[1] == self.__State.main_region_orthogonal_state_r2state_b
|
||||
return False
|
||||
|
||||
def time_elapsed(self, event_id):
|
||||
"""Add time events to in event queue
|
||||
"""
|
||||
if event_id in range(1):
|
||||
self.in_event_queue.put(lambda: self.raise_time_event(event_id))
|
||||
self.run_cycle()
|
||||
|
||||
def raise_time_event(self, event_id):
|
||||
"""Raise timed events using the event_id.
|
||||
"""
|
||||
self.__time_events[event_id] = True
|
||||
|
||||
def __execute_queued_event(self, func):
|
||||
func()
|
||||
|
||||
def __get_next_event(self):
|
||||
if not self.in_event_queue.empty():
|
||||
return self.in_event_queue.get()
|
||||
return None
|
||||
|
||||
def __entry_action_main_region_orthogonal_state_r2_state_b(self):
|
||||
"""Entry action for state 'StateB'..
|
||||
"""
|
||||
#Entry action for state 'StateB'.
|
||||
self.timer_service.set_timer(self, 0, (1 * 1000), False)
|
||||
|
||||
def __exit_action_main_region_orthogonal_state_r2_state_b(self):
|
||||
"""Exit action for state 'StateB'..
|
||||
"""
|
||||
#Exit action for state 'StateB'.
|
||||
self.timer_service.unset_timer(self, 0)
|
||||
|
||||
def __enter_sequence_main_region_orthogonal_state_default(self):
|
||||
"""'default' enter sequence for state OrthogonalState.
|
||||
"""
|
||||
#'default' enter sequence for state OrthogonalState
|
||||
self.__enter_sequence_main_region_orthogonal_state_r1_default()
|
||||
self.__enter_sequence_main_region_orthogonal_state_r2_default()
|
||||
|
||||
def __enter_sequence_main_region_orthogonal_state_r1_state_a_default(self):
|
||||
"""'default' enter sequence for state StateA.
|
||||
"""
|
||||
#'default' enter sequence for state StateA
|
||||
self.__state_vector[0] = self.State.main_region_orthogonal_state_r1state_a
|
||||
self.__state_conf_vector_position = 0
|
||||
self.__state_conf_vector_changed = True
|
||||
|
||||
def __enter_sequence_main_region_orthogonal_state_r2_state_b_default(self):
|
||||
"""'default' enter sequence for state StateB.
|
||||
"""
|
||||
#'default' enter sequence for state StateB
|
||||
self.__entry_action_main_region_orthogonal_state_r2_state_b()
|
||||
self.__state_vector[1] = self.State.main_region_orthogonal_state_r2state_b
|
||||
self.__state_conf_vector_position = 1
|
||||
self.__state_conf_vector_changed = True
|
||||
|
||||
def __enter_sequence_main_region_default(self):
|
||||
"""'default' enter sequence for region main region.
|
||||
"""
|
||||
#'default' enter sequence for region main region
|
||||
self.__react_main_region__entry_default()
|
||||
|
||||
def __enter_sequence_main_region_orthogonal_state_r1_default(self):
|
||||
"""'default' enter sequence for region r1.
|
||||
"""
|
||||
#'default' enter sequence for region r1
|
||||
self.__react_main_region_orthogonal_state_r1__entry_default()
|
||||
|
||||
def __enter_sequence_main_region_orthogonal_state_r2_default(self):
|
||||
"""'default' enter sequence for region r2.
|
||||
"""
|
||||
#'default' enter sequence for region r2
|
||||
self.__react_main_region_orthogonal_state_r2__entry_default()
|
||||
|
||||
def __exit_sequence_main_region_orthogonal_state_r1_state_a(self):
|
||||
"""Default exit sequence for state StateA.
|
||||
"""
|
||||
#Default exit sequence for state StateA
|
||||
self.__state_vector[0] = self.State.main_region_orthogonal_state
|
||||
self.__state_conf_vector_position = 0
|
||||
|
||||
def __exit_sequence_main_region_orthogonal_state_r2_state_b(self):
|
||||
"""Default exit sequence for state StateB.
|
||||
"""
|
||||
#Default exit sequence for state StateB
|
||||
self.__state_vector[1] = self.State.main_region_orthogonal_state
|
||||
self.__state_conf_vector_position = 1
|
||||
self.__exit_action_main_region_orthogonal_state_r2_state_b()
|
||||
|
||||
def __exit_sequence_main_region(self):
|
||||
"""Default exit sequence for region main region.
|
||||
"""
|
||||
#Default exit sequence for region main region
|
||||
state = self.__state_vector[0]
|
||||
if state == self.State.main_region_orthogonal_state_r1state_a:
|
||||
self.__exit_sequence_main_region_orthogonal_state_r1_state_a()
|
||||
state = self.__state_vector[1]
|
||||
if state == self.State.main_region_orthogonal_state_r2state_b:
|
||||
self.__exit_sequence_main_region_orthogonal_state_r2_state_b()
|
||||
|
||||
def __react_main_region_orthogonal_state_r1__entry_default(self):
|
||||
"""Default react sequence for initial entry .
|
||||
"""
|
||||
#Default react sequence for initial entry
|
||||
self.__enter_sequence_main_region_orthogonal_state_r1_state_a_default()
|
||||
|
||||
def __react_main_region_orthogonal_state_r2__entry_default(self):
|
||||
"""Default react sequence for initial entry .
|
||||
"""
|
||||
#Default react sequence for initial entry
|
||||
self.__enter_sequence_main_region_orthogonal_state_r2_state_b_default()
|
||||
|
||||
def __react_main_region__entry_default(self):
|
||||
"""Default react sequence for initial entry .
|
||||
"""
|
||||
#Default react sequence for initial entry
|
||||
self.__enter_sequence_main_region_orthogonal_state_default()
|
||||
|
||||
def __react(self, transitioned_before):
|
||||
"""Implementation of __react function.
|
||||
"""
|
||||
#State machine reactions.
|
||||
return transitioned_before
|
||||
|
||||
|
||||
def __main_region_orthogonal_state_react(self, transitioned_before):
|
||||
"""Implementation of __main_region_orthogonal_state_react function.
|
||||
"""
|
||||
#The reactions of state OrthogonalState.
|
||||
transitioned_after = transitioned_before
|
||||
#Always execute local reactions.
|
||||
transitioned_after = self.__react(transitioned_before)
|
||||
return transitioned_after
|
||||
|
||||
|
||||
def __main_region_orthogonal_state_r1_state_a_react(self, transitioned_before):
|
||||
"""Implementation of __main_region_orthogonal_state_r1_state_a_react function.
|
||||
"""
|
||||
#The reactions of state StateA.
|
||||
transitioned_after = transitioned_before
|
||||
if transitioned_after < 0:
|
||||
if self.v == 0:
|
||||
self.__exit_sequence_main_region_orthogonal_state_r1_state_a()
|
||||
self.x_observable.next()
|
||||
self.__enter_sequence_main_region_orthogonal_state_r1_state_a_default()
|
||||
transitioned_after = 0
|
||||
return transitioned_after
|
||||
|
||||
|
||||
def __main_region_orthogonal_state_r2_state_b_react(self, transitioned_before):
|
||||
"""Implementation of __main_region_orthogonal_state_r2_state_b_react function.
|
||||
"""
|
||||
#The reactions of state StateB.
|
||||
transitioned_after = transitioned_before
|
||||
if transitioned_after < 1:
|
||||
if self.__time_events[0]:
|
||||
self.__exit_sequence_main_region_orthogonal_state_r2_state_b()
|
||||
self.y_observable.next()
|
||||
self.__time_events[0] = False
|
||||
self.__enter_sequence_main_region_orthogonal_state_r2_state_b_default()
|
||||
self.__main_region_orthogonal_state_react(0)
|
||||
transitioned_after = 1
|
||||
#If no transition was taken
|
||||
if transitioned_after == transitioned_before:
|
||||
#then execute local reactions.
|
||||
transitioned_after = self.__main_region_orthogonal_state_react(transitioned_before)
|
||||
return transitioned_after
|
||||
|
||||
|
||||
def __clear_in_events(self):
|
||||
"""Implementation of __clear_in_events function.
|
||||
"""
|
||||
self.__time_events[0] = False
|
||||
|
||||
|
||||
def __micro_step(self):
|
||||
"""Implementation of __micro_step function.
|
||||
"""
|
||||
transitioned = -1
|
||||
self.__state_conf_vector_position = 0
|
||||
state = self.__state_vector[0]
|
||||
if state == self.State.main_region_orthogonal_state_r1state_a:
|
||||
transitioned = self.__main_region_orthogonal_state_r1_state_a_react(transitioned)
|
||||
if self.__state_conf_vector_position < 1:
|
||||
state = self.__state_vector[1]
|
||||
if state == self.State.main_region_orthogonal_state_r2state_b:
|
||||
self.__main_region_orthogonal_state_r2_state_b_react(transitioned)
|
||||
|
||||
|
||||
def run_cycle(self):
|
||||
"""Implementation of run_cycle function.
|
||||
"""
|
||||
#Performs a 'run to completion' step.
|
||||
if self.timer_service is None:
|
||||
raise ValueError('Timer service must be set.')
|
||||
|
||||
if self.__is_executing:
|
||||
return
|
||||
self.__is_executing = True
|
||||
next_event = self.__get_next_event()
|
||||
if next_event is not None:
|
||||
self.__execute_queued_event(next_event)
|
||||
condition_0 = True
|
||||
while condition_0:
|
||||
self.__micro_step()
|
||||
self.__clear_in_events()
|
||||
condition_0 = False
|
||||
next_event = self.__get_next_event()
|
||||
if next_event is not None:
|
||||
self.__execute_queued_event(next_event)
|
||||
condition_0 = True
|
||||
self.__is_executing = False
|
||||
|
||||
|
||||
def enter(self):
|
||||
"""Implementation of enter function.
|
||||
"""
|
||||
#Activates the state machine.
|
||||
if self.timer_service is None:
|
||||
raise ValueError('Timer service must be set.')
|
||||
|
||||
if self.__is_executing:
|
||||
return
|
||||
self.__is_executing = True
|
||||
#Default enter sequence for statechart E
|
||||
self.__enter_sequence_main_region_default()
|
||||
self.__is_executing = False
|
||||
|
||||
|
||||
def exit(self):
|
||||
"""Implementation of exit function.
|
||||
"""
|
||||
#Deactivates the state machine.
|
||||
if self.__is_executing:
|
||||
return
|
||||
self.__is_executing = True
|
||||
#Default exit sequence for statechart E
|
||||
self.__exit_sequence_main_region()
|
||||
self.__state_vector[0] = self.State.null_state
|
||||
self.__state_vector[1] = self.State.null_state
|
||||
self.__state_conf_vector_position = 1
|
||||
self.__is_executing = False
|
||||
|
||||
|
||||
def trigger_without_event(self):
|
||||
"""Implementation of triggerWithoutEvent function.
|
||||
"""
|
||||
self.run_cycle()
|
||||
|
||||
Loading…
Add table
Add a link
Reference in a new issue