Lua CEP API

Subjects Subject API – adc_main_an3_v

– adc_main_an4_v
mvb_var Subscribe to a variable on the MVB bus. – adc_main_an5_v
mvb_var(name)
can_var Subscribe to a variable on the CAN bus. – adc_main_an6_v
j1708_var Subscribe to a variable on the J1708 bus. – adc_main_an7_v
• Description: Subscribe to a variable on the MVB bus.
rn_var Subscribe to a Railster variable. – adc_main_an8_v
• Argument: name (string) – The MVB variable name.
adc_var Subscribe to an ADC variable. – adc_main_an9_v
• Return: Observable – An Observable that produces a new
value anytime the variable is seen on the bus.
Observables Observable API
rename() Rename an observable and optionally set its value. rename(new_name, value=None)
can_var(name)
scale() Produce values scaled by a fixed coefficient.
mapval() Like :map, but only acts on the value of the variable. • Description: Rename an observable and optionally set its
• Description: Subscribe to a variable on the CAN bus.
filterval() Like :filter, but only acts on the value of the variable. value.
• Argument: name (string) – The CAN variable name.
scanval() Like :scan, but only acts on the value of the variable. • Arguments:
dt() Produces the difference in seconds between samples. • Return: Observable – An Observable that produces a new
value anytime the variable is seen on the bus. – new_name (string) – The new name for the observable.
take_when() Produces values when a predicate on ‘other‘ is true. – value (optional, any type) – A new value to set for the ob-
binary() Transform the values to 1 or 0, based on predicate. servable.
count() Count the number of samples received. j1708_var(name) • Return: Observable – A new Observable with the specified
time_count() Count how long values have satisfied a predicate. name and value.
diff() Produces the difference between subsequent values. • Description: Subscribe to a variable on the J1708 bus.
time_diff() Derivative of a variable over time. • Argument: name (string) – The J1708 variable name. scale(coefficient)
integrate() Integral of a variable over time. • Return: Observable – An Observable that produces a new
agg() Apply an aggregation function on subsequent values. value anytime the variable is seen on the bus. • Description: Produce values scaled by a fixed coefficient.
is() Filter values that are exactly a given value. • Argument: coefficient (numeric) – The scaling coefficient.
above() Filter values above a given value. • Return: Observable – A new Observable that produces scaled
rn_var(name) values.
below() Filter values below a given value.
s_trig() Schmidt trigger: returns an Observable that produces
boolean values when its input - goes above an upper • Description: Subscribe to a Railster variable.
mapval(function)
limit (produces true) - goes below a lower limit (pro- • Argument: name (string) – The Railster variable name.
duces false).
• Return: Observable – An Observable that produces a new • Description: Like :map, but only acts on the value of the vari-
ttl_group() Return a new Observable that combine values pro- value anytime the variable is seen on the bus.
duced by self and the arguments, if their latest value able.
has been produced within a time-to-live timespan. • Argument: function (function) – The function to apply on the
dedup() Return a new Observable that yields its input only variable value.
adc_var(name)
when it differs from its previous value. • Return: Observable – A new Observable with the mapped val-
run- Exponential smoothing estimation of the variable ues.
ning_ mean() mean over a timespan. • Description: Subscribe to an ADC variable.
• Argument: name (string) – The ADC variable name. filterval(function)
Observers • Return: Observable – An Observable that produces a new
value anytime the variable is seen on the bus.
:alert() Emit a named alert every time a sample is given to it. • Description: Like :filter, but only acts on the value of the
• Remark: The valid ADC names are: variable.
logger() Log the observables at a specified period.
– adc_main_an0_v • Argument: function (function) – The function to apply on the
counter() Send an accumulated variable. variable value.
live() Send a live variable for near realtime dashboarding. – adc_main_an1_v
• Return: Observable – A new Observable with the filtered val-
live_debug() Send a debug variable for non-validated variables. – adc_main_an2_v ues.
scanval(function, seed) – predicate (function) – The function that takes a value and s_trig(lower, upper)
returns a boolean.

• Description: Like :scan, but only acts on the value of the vari- – interval (float) – The time interval over which to count • Description: Schmidt trigger: returns an Observable that pro-
able. values (in seconds). duces boolean values when its input goes above an upper limit
• Arguments: (produces true) or goes below a lower limit (produces false).
diff() • Arguments:
– function (function) – The function to apply on the vari-
able value. – lower – The lower limit.
– seed (any type) – The initial value for the accumulator. • Description: Produces the difference between subsequent
– upper – The upper limit.
values.
• Return: Observable – A new Observable with the accumulated
values.
time_diff() binary(predicate)

dt()
• Description: Derivative of a variable over time. • Description: Transform the values to 1 or 0, based on a given
predicate.
• Description: Produces the difference in seconds between • Arguments:
samples. integrate()
• Return: Observable – A new Observable that produces time – predicate (function) – The function that takes a value and
differences between samples. returns a boolean.
• Description: Integral of a variable over time.

take_when(predicate, other) count()

agg(aggregator)

• Description: Produces values when a predicate on other is • Description: Count the number of samples received.
• Description: Apply an aggregation function on subsequent
true. values.
• Arguments: • Arguments: time_count(predicate, interval)
– predicate (function) – The function that takes the value – aggregator (function) – The aggregation function to apply
of other and returns a boolean. on values. • Description: Count how long values have satisfied a given
– other (Observable) – The Observable to apply the predi- predicate.
cate on.
is(value) • Arguments:
– predicate (function) – The function that takes a value and
binary(predicate) returns a boolean.
• Description: Filter values that are exactly a given value.
• Arguments: – interval (float) – The time interval over which to count
• Description: Transform the values to 1 or 0, based on a given values (in seconds).
predicate. – value – The value to compare against.
• Arguments:
diff()
– predicate (function) – The function that takes a value and above(value)
returns a boolean.
• Description: Produces the difference between subsequent
• Description: Filter values above a given value. values.
count() • Arguments:
– value – The value to compare against. time_diff()
• Description: Count the number of samples received.

below(value) • Description: Derivative of a variable over time.

time_count(predicate, interval)
• Description: Filter values below a given value.
• Description: Count how long values have satisfied a given integrate()
predicate. • Arguments:
• Arguments: – value – The value to compare against. • Description: Integral of a variable over time.
agg(aggregator) dedup(keepalive_period) live_debug()

• Description: Apply an aggregation function on subsequent • Description: Returns a new Observable that yields its input • Description: Send a debug variable for non-validated vari-
values. only when it differs from its previous value. ables.
• Arguments: • Arguments: • Arguments: None.
– aggregator (function) – The aggregation function to apply
– keepalive_period (optional number) – The period of time
on values.
in seconds during which the input must differ from its Examples
previous value in order to be emitted.
is(value)
Simple alert
running_mean(timespan)
• Description: Filter values that are exactly a given value.
Subscribe to the speed variable on the MVB bus, filter values above
• Arguments: 100, and emit a "HighSpeedAlert" when this condition is met.
• Description: Exponential smoothing estimation of the vari-
– value – The value to compare against. able mean over a timespan. At time t, produces a value S(t)
from a value X(t) such that S(t) = p*X(t) + (1-p)*S(t-1), where p is mvb_var("speed"):above(100):alert("HighSpeedAlert")
the elapsed time between X(t) and X(t-1) divided by the times-
above(value) pan.
Alert with mapval
• Arguments:
• Description: Filter values above a given value.
– timespan (number) – The time span duration, in seconds. Subscribe to the pressure variable from the MVB bus, calculate the
• Arguments:
difference between subsequent values, take the absolute value of the
– value – The value to compare against. difference, and emit a "PressureChangeAlert" when the difference is
Observers
greater than 5.

below(value) :alert(name) mvb_var("pressure")

:diff()
• Description: Filter values below a given value. :mapval(function (val) [Link](x) end)
• Description: Emit a named alert every time a sample is given
:above(5)
• Arguments: to it.
:alert("PressureChangeAlert")
– value – The value to compare against. • Arguments:
– name (string) – The name of the alert to be emitted. Logger
s_trig(lower, upper)
logger(period) Subscribe to the engine RPM variable from the J1708 bus, calculate
• Description: Schmidt trigger: returns an Observable that pro- the derivative of the variable over time, and log (every 1s) the values
duces boolean values when its input goes above an upper limit when the derivative is greater than 1000.
(produces true) or goes below a lower limit (produces false). • Description: Log the observables at a specified period.
• Arguments: • Arguments: logger(1,
j1708_var("engine_rpm")
– lower – The lower limit. – period (number) – The logging period in seconds. :time_diff()
– upper – The upper limit. :above(1000)
)
counter()
ttl_group(ttl, ...)
Live
• Description: Send an accumulated variable.
• Description: Returns a new Observable that combines val-
ues produced by the original Observable and the arguments, • Arguments: None. Subscribe to the battery voltage variable from the CAN bus, calculate
if their latest value has been produced within a time-to-live the running-mean over a 60-second timespan, rename the variable
timespan. and send the values every 30s to the server.
• Arguments: live()
live(
– ttl (number) – The time to live for each variable in this can_var("battery_voltage")
group, in seconds. • Description: Send a live variable for near realtime dashboard-
:running_mean(60)
ing.
– ... (Observables) – List of Observables to group with the :rename("BatteryVoltageMean")
original Observable. • Arguments: None. )
 Scaling

Subscribe to the brake pressure variable from MVB, scale the values
by a factor of 0.145, filter values above 100, and emit a "HighBrake-
PressureAlert" when this condition is met.

mvb_var("brake_pressure")
:scale(0.145)
:above(100)
:alert("HighBrakePressureAlert")

Counter and integration

Subscribe to the fuel level variable from the CAN bus, calculate the
integral of the variable over time, and send the accumulated value.

counter(can_var("fuel_level"):integrate())

Counting occurrences

Subscribe to the throttle position variable from the J1708 bus, count
how long the values have satisfied the condition of being greater than
80 over a 60-second period, and send the live variable for near real-
time dashboarding.

live(
j1708_var("throttle_position")
:time_count(function (val)
return val > 80
end, true)
:rename("HighThrottleDuration")
)