CoSApp tutorials
Simulations with discrete-time events¶
Dynamic systems involving time derivatives can be solved using a time driver (see tutorial on time simulations). Such simulations are referred to as continuous time simulations, as all quantities are expected to vary continuously with time.
Discontinuities, however, can be introduced with the occurrence of events, defined within a system.
Events¶
Events are triggerable objects which activate when a certain condition is detected in their owner system. They are defined at system setup with method add_event
:
class BasicEventSystem(System):
def setup(self):
self.add_event('e')
In this basic example, event e
is defined, but no activation condition is prescribed. As long as it is the case, the event will never occur.
Event trigger¶
The activation of an event is driven by its attribute trigger
. In the most simple case, the trigger is defined by a character string describing the activation condition as an evaluable expression. An event trigger can be set at the definition of the event, by specifying argument trigger=...
in add_event
, or interactively for a given system, as in:
basic = BasicEventSystem('basic')
basic.e.trigger = 't == 0.123'
Here, system basic
will react if time ever reaches value 0.123 during a time simulation.
Valid trigger expressions are of the kind "lhs <op> rhs"
, where left- and right-hand sides lhs
and rhs
are expressions evaluable in the context of the owner system, and operator <op>
is one of (==
, <
, >
, <=
, >=
). Even though "lhs <op> rhs"
corresponds to a Boolean expression, the event activation will not be triggered by the Boolean value. Instead, time drivers will check for sign changes of lhs - rhs
at each integration time step. Since lhs and rhs
are assumed to be continuous expressions in time, a sign change of their difference guaranties that there exists a time when lhs equals rhs. This time value, determined numerically by solving a root-finding problem, will be regarded as the occurrence time of the event.
Strict and non-strict inequalities are treated identically, so the actual discriminating cases are ==
, <=
, and >=
:
<=
: expressionlhs - rhs
goes from positive to negative, meaninglhs
becomes smaller thanrhs
;>=
: expressionlhs - rhs
goes from negative to positive, meaninglhs
becomes greater thanrhs
;==
: expressionlhs - rhs
simply changes sign.
These distinctions allow one to define directional events (different events occur when a exceeds b, and when b exceeds a), as well as one-way events. For example,
class BreakableSystem(System):
def setup(self):
self.add_inward('x', 0.0)
self.add_inward('x_max', 1.0)
self.add_event('failure', trigger="x > x_max")
indicates that if failure
ever occurs (in which case we expect the system will change, as we will see later), there is no going back, even is x
later returns below x_max
.
Primitive and derived events¶
Events triggered by a zero-crossing expression, as discussed in previous section, are said to be primitive, as they are self-determined. Alternatively, events may be triggered by other events, in which case they are referred to as secondary, or derived events.
Derived events can be either:
Synchronized events, simply triggered by the occurrence of another event:
event.trigger = other_event
Filtered events, triggered by the occurrence of a distinct event and an additional condition, given as an evaluable Boolean expression. Syntax is:
event = other_event.filter(<condition>)
Merged events, triggered by the occurrence of one among a list of events. Syntax is:
from cosapp.multimode import Event event = Event.merge(event1, event2, ...)
[1]:
from cosapp.base import System
from cosapp.multimode import Event
from math import hypot, cos, exp
class MultiEventSystem(System):
def setup(self):
self.add_inward('x', 1.0)
self.add_inward('y', 0.0)
self.add_outward('r', 0.0)
self.add_outward('z', 0.0)
# Primitive events:
kapow = self.add_event('kapow', trigger="x == z")
boom = self.add_event('boom', trigger="z < 0.25")
# Derived events:
self.add_event('bam', trigger=kapow.filter("r > 1"))
self.add_event('zap', trigger=Event.merge(kapow, boom))
def compute(self):
self.r = r = hypot(self.x, self.y)
self.z = exp(-r / 4) * cos(5 * r)
class BasicEventSystem(System):
def setup(self):
self.add_event('e') # undefined, at this point
batman = MultiEventSystem('batman')
basic = BasicEventSystem('basic')
basic.e.trigger = batman.zap # synchronized events
Final events¶
An event can be defined as final, meaning its occurrence will stop time simulations. Two possibilities are available:
Specify optional argument
final=True
inSystem.add_event
;Set attribute
final
, as inbatman.kapow.final = True
.
Defining what happens when an event occurs¶
The way events impact their owner system is described in method System.transition
. This method can be viewed as the counterpart of compute
for event-based behaviour, whereas compute
describes the system behaviour during continuous time phases. Whenever one or several primitive events are detected between \(t\) and \(t + \Delta t\), their occurrence times are computed, only to retain the earliest, as the occurrence time \(t_{e}\). After a first regular integration
between \(t\) and \(t + t_{e}\), system transition is executed, assuming time is fixed at \(t_{e}\).
The transition logic in each system can be specified, depending on which events actually occurred. This information is known from attribute present
of events. During transition, systems can be reconfigured, by adding sub-systems, declaring new unknowns, new connections, etc.
class SpaceRocket(System):
def setup(self):
self.add_outward('a', np.zeros(3), desc="Acceleration")
self.add_transient('v', der='a')
self.add_transient('x', der='v')
self.add_child(BottomRocketSection('bottom'))
self.add_child(TopRocketSection('top'))
self.add_event('enter_stratosphere', trigger="x[2] > 6000")
def compute(self):
# Define how acceleration is computed
def transition(self):
if self.enter_stratosphere.present:
self.pop_child('bottom')
Mode variables¶
Mode variables are piecewise constant variables which may only change value during system transitions. They are useful to keep information on the current state (or mode) of the system. In particular, their values can be used to devise different courses of action in method compute
.
Like their continuous-time counterparts (inwards, outwards, etc.), mode variables can be either input or output variables. They are declared with methods add_inward_modevar
and add_outward_modevar
at system setup.
For example:
class BreakableSystem(System):
def setup(self):
self.add_inward('x_max', 1.0)
self.add_inward('x', 0.0)
self.add_outward('y', 0.0)
self.add_event('failure', trigger="x > x_max")
self.add_outward_modevar('broken', False)
def compute(self):
self.y = 0.0 if self.broken else self.x
def transition(self):
if self.failure.present:
self.broken = True
Ground rules¶
Just like inwards or input port variables should not be modified inside method compute
, input mode variables should not be modified inside transition
. More importantly, mode variables (either input or output) must never be modified in method compute
, to ensure that they are indeed constant in each continuous time phase. The reciprocal is not true, though, as continuous time variables can be modified during transitions.
Since outward mode variables can be used in method compute
but may only be changed later in transition
, their initial value may be of importance. In such cases, the value to be used at the beginning of time simulations (that is during the first continuous time phase) can be specified by argument init
in add_outward_modevar
. The value can be a constant, or an evaluable expression. For example, in BreakableSystem
above, we may want to start with a value of broken
consistent
with the actual initial values of x
and x_max
, and thus declare:
class BreakableSystem(System):
def setup(self):
self.add_inward('x_max', 1.0)
self.add_inward('x', 0.0)
self.add_outward('y', 0.0)
self.add_event('failure', trigger="x > x_max")
self.add_outward_modevar('broken', False, init="x > x_max")
The second argument of add_outward_modevar
corresponds to the value assigned at the creation of the variable (False
, here). When init
is specified, this value can be omitted. If both arguments are given, though, the expression of init
must be consistent with the type of value
.
Connecting mode variables¶
Mode variables can be passed on from one system to another, with the same rules as inwards and outwards. An output mode variable may be connected to an input continuous time variable (either inward or input port variable). However, input mode variables cannot be connected to continuous time variables.
[2]:
from cosapp.base import System, Port
import numpy as np
import logging
class AbPort(Port):
def setup(self):
self.add_variable('a', 1.0)
self.add_variable('b', np.zeros(3))
class ContinuousTimeSystem(System):
def setup(self):
self.add_input(AbPort, 'p_in')
self.add_output(AbPort, 'p_out')
self.add_inward('x_in', 1.0)
self.add_outward('x_out', 0.0)
class BasicMultimodeSystem(System):
def setup(self):
self.add_inward_modevar("m_in", 3.14)
self.add_outward_modevar("m_out", 0.1, desc="System state")
self.add_event("beep", trigger=None)
top = System('top')
c1 = top.add_child(ContinuousTimeSystem('c1'))
c2 = top.add_child(ContinuousTimeSystem('c2'))
d1 = top.add_child(BasicMultimodeSystem('d1'))
d2 = top.add_child(BasicMultimodeSystem('d2'))
top.connect(c1, d1, {'x_in': 'm_out'}) # discrete.m_out -> continuous.x_in
top.connect(d1, d2, {'m_out': 'm_in'}) # discrete.m_out -> discrete.m_in
try:
# continuous.x_out -> discrete.m_in - not allowed
top.connect(c2, d1, {'x_out': 'm_in'})
except Exception as error:
logging.error(error)
ERROR:root:Input mode variables cannot be connected to continuous time variables (c2.outwards -> d1).
Time drivers and events¶
Time drivers keep a record of all encountered events, accessible at the end of simulations with property recorded_events
. This attribute contains a list of named tuples EventRecord(time, events)
, each containing the occurrence time, and the list of cascading events at that time, starting by the primitive event.
If the time driver performing the simulation has a recorder, the latter will record a double entry at each event occurrence, capturing the system state before and after transition. Furthermore, the driver also creates a secondary recorder event_data
. Recorder event_data
will track all the fields requested in the user-defined recorder, but will contain only time entries corresponding to events, for easier post-processing.