API Reference
This page contains the documentation for the mstlo Python interface.
Core Functions
mstlo_python.parse_formula(formula_str)
Parse an STL formula from a string using the same DSL syntax as Rust's stl! macro.
This allows you to write formulas using the same syntax as Rust, making it easy to port formulas between Python and Rust code.
Syntax:
Predicates:
signal > value- Signal greater than valuesignal < value- Signal less than valuesignal >= value- Signal greater than or equalsignal <= value- Signal less than or equal
Variable Predicates (for runtime-updateable thresholds):
signal > $variable- Signal greater than variablesignal < $variable- Signal less than variablesignal >= $variable- Signal greater than or equal to variablesignal <= $variable- Signal less than or equal to variable
Boolean Constants:
true- Always truefalse- Always false
Unary Operators:
!(sub)ornot(sub)- NegationG[start, end](sub)orglobally[start, end](sub)- Globally (always)F[start, end](sub)oreventually[start, end](sub)- Eventually (finally)
Binary Operators:
left && rightorleft and right- Conjunctionleft || rightorleft or right- Disjunctionleft -> rightorleft implies right- Implicationleft U[start, end] rightorleft until[start, end] right- Until
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
formula_str
|
str
|
A string containing an STL formula |
required |
Returns:
| Type | Description |
|---|---|
Formula
|
The parsed Formula object |
Raises:
| Type | Description |
|---|---|
ValueError
|
If the formula string cannot be parsed |
Examples:
>>> # Simple predicate
>>> f = parse_formula("x > 5")
>>> # Globally operator
>>> f = parse_formula("G[0, 10](x > 5)")
>>> # Complex formula
>>> f = parse_formula("G[0, 10](x > 5) && F[0, 5](y < 3)")
>>> # Using keyword syntax
>>> f = parse_formula("globally[0, 10](x > 5) and eventually[0, 5](y < 3)")
>>> # Using variables for runtime-updateable thresholds
>>> f = parse_formula("G[0, 10](x > $threshold)")
>>> f = parse_formula("x > $min && x < $max")
Core Classes
mstlo_python.Monitor
Online STL monitor.
Monitors signal traces against an STL formula and produces verdicts. Supports multiple semantics (DelayedQualitative, EagerQualitative, DelayedQuantitative, Rosi), algorithms (Incremental, Naive), and signal interpolations (ZeroOrderHold, Linear).
The monitor processes signals incrementally and produces verdicts when sufficient information is available.
Examples:
>>> phi = Formula.always(0, 5, Formula.gt('x', 0.5))
>>> monitor = Monitor(phi, semantics='Rosi')
>>> for t in range(10):
... result = monitor.update('x', 0.8, float(t))
... print(result)
update(signal, value, timestamp)
Update the monitor with a new signal value.
Processes the new data point and produces verdicts when sufficient information is available. Each update may trigger multiple evaluations, one for each synchronized step (which may include interpolated values).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
signal
|
str
|
Name of the signal being updated |
required |
value
|
float
|
Signal value at this timestamp |
required |
timestamp
|
float
|
Timestamp in seconds (must be monotonically increasing) |
required |
Returns:
| Type | Description |
|---|---|
MonitorOutput
|
MonitorOutput object containing: |
MonitorOutput
|
|
MonitorOutput
|
|
MonitorOutput
|
|
MonitorOutput
|
|
Note
- Evaluations list may be empty if data is being buffered
- For multi-signal formulas, the synchronizer may produce interpolated steps
- Each evaluation corresponds to a specific synchronized step
- Multiple outputs may be produced for each synchronized step
Examples:
>>> output = monitor.update('x', 0.8, 1.0)
>>> # Use Rust-style Display formatting
>>> print(output)
>>> # Access properties
>>> print(f"Input: {output.input_signal} at t={output.input_timestamp}")
>>> # Get verdicts
>>> for ts, val in output.verdicts():
... print(f"Verdict at {ts}: {val}")
>>> # Access as dictionary
>>> d = output.to_dict()
>>> for eval in d['evaluations']:
... for out in eval['outputs']:
... print(f" t={out['timestamp']}: {out['value']}")
get_signal_identifiers()
Get the set of signal identifiers used in the monitor's formula.
Returns:
| Type | Description |
|---|---|
set[str]
|
Set[str]: A set of signal names (identifiers) used in the formula. |
Examples:
>>> phi = parse_formula("G[0,5](x > 5) && F[0,3](y < 10)")
>>> monitor = Monitor(phi)
>>> signals = monitor.get_signal_identifiers()
>>> print(signals) # {'x', 'y'}
get_variables()
Get the Variables object associated with this monitor.
This allows you to update variable values at runtime, which will affect future evaluations of the formula.
Returns:
| Type | Description |
|---|---|
Variables
|
The Variables object containing runtime variable values. |
Examples:
>>> vars = Variables()
>>> vars.set("threshold", 5.0)
>>> monitor = Monitor(formula, variables=vars)
>>> # Update a variable at runtime
>>> monitor.get_variables().set("threshold", 10.0)
get_specification()
Get the STL specification as a string representation.
Returns:
| Type | Description |
|---|---|
str
|
String representation of the STL formula being monitored. |
Examples:
>>> phi = parse_formula("G[0,5](x > 5)")
>>> monitor = Monitor(phi)
>>> print(monitor.get_specification()) # "G[0s,5s](x > 5)"
get_algorithm()
Get the algorithm used by this monitor.
Returns:
| Type | Description |
|---|---|
str
|
"Incremental" or "Naive" |
Examples:
>>> monitor = Monitor(phi, algorithm="Incremental")
>>> print(monitor.get_algorithm()) # "Incremental"
get_semantics()
Get the semantics used by this monitor.
Returns:
| Type | Description |
|---|---|
str
|
One of: "DelayedQualitative", "EagerQualitative", "DelayedQuantitative", or "Rosi" |
Examples:
>>> monitor = Monitor(phi, semantics="DelayedQuantitative")
>>> print(monitor.get_semantics()) # "DelayedQuantitative"
get_synchronization_strategy()
Deprecated, use get_signal_interpolation.
Emits a DeprecationWarning and reports the interpolation in force, so it answers "ZeroOrderHold" or "Linear" and never "None".
Returns:
| Type | Description |
|---|---|
str
|
One of: "ZeroOrderHold" or "Linear" |
Examples:
>>> monitor = Monitor(phi, signal_interpolation="Linear")
>>> print(monitor.get_signal_interpolation()) # "Linear"
get_signal_interpolation()
Get the signal interpolation used by this monitor.
Returns:
| Type | Description |
|---|---|
str
|
One of: "ZeroOrderHold" or "Linear" |
Examples:
>>> monitor = Monitor(phi, semantics="DelayedQualitative",
... signal_interpolation="Linear")
>>> print(monitor.get_signal_interpolation()) # "Linear"
get_temporal_depth()
Get the maximum lookahead required by the formula (temporal depth).
This corresponds to the maximum time interval in any temporal operator.
Returns:
| Type | Description |
|---|---|
float
|
The temporal depth in seconds. |
Examples:
>>> phi = parse_formula("G[0,5](x > 5) && F[0,3](y < 10)")
>>> monitor = Monitor(phi)
>>> print(monitor.get_temporal_depth()) # 5.0
reset()
Reset the monitor to its initial state.
Clears all internal caches and evaluation buffers. The formula, semantics, algorithm, synchronization strategy, and variables are preserved.
Use this to reuse a monitor across multiple independent traces without rebuilding it from scratch.
Examples:
>>> monitor = Monitor(formula)
>>> for signal, value, ts in trace_1:
... monitor.update(signal, value, ts)
>>> monitor.reset()
>>> for signal, value, ts in trace_2:
... monitor.update(signal, value, ts)
update_batch(steps)
Update the monitor with a batch of steps.
The batch may be given in either of two forms:
- signal-major - a mapping from each signal name to its
(value, timestamp)samples, when you have one trace per signal; - flat - an iterable of
(signal, value, timestamp)tuples, when your samples are interleaved.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
steps
|
Union[Mapping[str, Iterable[Tuple[float, float]]], Iterable[Tuple[str, float, float]]]
|
The batch, in either form. Examples:
|
required |
Returns:
| Type | Description |
|---|---|
MonitorOutput
|
A single MonitorOutput holding every evaluation produced by the |
MonitorOutput
|
batch. Its input metadata is the last step processed, or None if |
MonitorOutput
|
the batch was empty. |
Raises:
| Type | Description |
|---|---|
ValueError
|
If the batch is not one of the two accepted forms. |
Note
Steps are sorted by timestamp before evaluation, because the Incremental algorithm expects a chronological stream. The sort is stable, so steps sharing a timestamp are processed in the order given: the result of a batch is fully determined by its input.
Examples:
>>> steps = {
... "temperature": [(25.0, 1.0), (26.0, 2.0)],
... "pressure": [(101.3, 1.5)]
... }
>>> output = monitor.update_batch(steps)
>>> print(output) # Display all finalized verdicts
>>> output = monitor.update_batch([
... ("temperature", 25.0, 1.0),
... ("pressure", 101.3, 1.5),
... ("temperature", 26.0, 2.0),
... ])
mstlo_python.Variables
Container for runtime variable values.
Variables allow you to define STL formulas with dynamic thresholds that
can be updated at runtime. Use the $variable syntax in formulas to
reference variables.
Examples:
>>> # Create variables and set initial values
>>> vars = Variables()
>>> vars.set("threshold", 5.0)
>>> vars.set("limit", 10.0)
>>>
>>> # Parse a formula using variables
>>> formula = parse_formula("x > $threshold && y < $limit")
>>>
>>> # Create a monitor with variables
>>> monitor = Monitor(formula, variables=vars)
>>>
>>> # Update a variable value at runtime
>>> vars.set("threshold", 7.0) # Affects future evaluations
Note
Variable predicates require the Incremental algorithm. Using variables with the Naive algorithm will raise an error.
clear()
Remove all variables.
Examples:
>>> vars = Variables()
>>> vars.set("x", 5.0)
>>> vars.clear()
>>> vars.names() # []
contains(name)
Check if a variable exists.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
The variable name (without $ prefix) |
required |
Returns:
| Type | Description |
|---|---|
bool
|
True if the variable exists, False otherwise. |
Examples:
>>> vars = Variables()
>>> vars.set("x", 5.0)
>>> vars.contains("x") # True
>>> vars.contains("y") # False
get(name)
Get a variable's value.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
The variable name (without $ prefix) |
required |
Returns:
| Type | Description |
|---|---|
Union[float, None]
|
The variable's value, or None if not set. |
Examples:
>>> vars = Variables()
>>> vars.set("x", 5.0)
>>> print(vars.get("x")) # prints 5.0
>>> print(vars.get("y")) # prints None
names()
Get a list of all variable names.
Returns:
| Type | Description |
|---|---|
List[str]
|
A list of variable names. |
Examples:
>>> vars = Variables()
>>> vars.set("a", 1.0)
>>> vars.set("b", 2.0)
>>> vars.names() # ['a', 'b']
remove(name)
Remove a variable.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
The variable name to remove (without $ prefix) |
required |
Returns:
| Type | Description |
|---|---|
Union[float, None]
|
The variable's previous value, or None if it didn't exist. |
Examples:
>>> vars = Variables()
>>> vars.set("x", 5.0)
>>> vars.remove("x") # returns 5.0
>>> vars.remove("x") # returns None
set(name, value)
Set a variable to a value.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
The variable name (without $ prefix) |
required |
value
|
float
|
The variable value |
required |
Examples:
>>> vars = Variables()
>>> vars.set("threshold", 5.0)
mstlo_python.Formula
Signal Temporal Logic (STL) formula.
Use static methods to construct formulas. Formulas can be composed using boolean and temporal operators.
Examples:
>>> # G[0,5](x > 0.5)
>>> phi = Formula.always(0, 5, Formula.gt('x', 0.5))
>>> # (x > 0.3) AND F[0,3](y < 0.8)
>>> phi2 = Formula.and_(
... Formula.gt('x', 0.3),
... Formula.eventually(0, 3, Formula.lt('y', 0.8))
... )
>>> # Using variables for dynamic thresholds
>>> phi3 = Formula.gt_var('x', 'threshold')
always(start, end, child)
staticmethod
Create a globally (always) temporal formula.
The formula must hold at all time points in the interval [start, end] relative to the current time.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
start
|
float
|
Start of time interval (seconds) |
required |
end
|
float
|
End of time interval (seconds) |
required |
child
|
Formula
|
Formula that must hold throughout the interval |
required |
Returns:
| Type | Description |
|---|---|
Formula
|
Formula representing: |
Examples:
>>> # For the next 5 seconds, x must be > 0.5
>>> Formula.always(0, 5, Formula.gt('x', 0.5))
and_(left, right)
staticmethod
eventually(start, end, child)
staticmethod
Create an eventually (finally) temporal formula.
The formula must hold at some time point in the interval [start, end] relative to the current time.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
start
|
float
|
Start of time interval (seconds) |
required |
end
|
float
|
End of time interval (seconds) |
required |
child
|
Formula
|
Formula that must hold at some point in the interval |
required |
Returns:
| Type | Description |
|---|---|
Formula
|
Formula representing: |
Examples:
>>> # Within 3 seconds, y must drop below 0.8
>>> Formula.eventually(0, 3, Formula.lt('y', 0.8))
false_()
staticmethod
Create a constant false formula.
Returns:
| Type | Description |
|---|---|
Formula
|
Formula that is never satisfied (⊥) |
Examples:
>>> Formula.false_()
gt(signal, value)
staticmethod
Create a greater-than atomic predicate.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
signal
|
str
|
Name of the signal to compare |
required |
value
|
float
|
Threshold value to compare against |
required |
Returns:
| Type | Description |
|---|---|
Formula
|
Formula representing: |
Examples:
>>> Formula.gt('x', 0.5) # x > 0.5
gt_var(signal, variable)
staticmethod
Create a greater-than atomic predicate with a variable threshold.
The variable value is looked up at runtime from a Variables object, allowing dynamic thresholds that can be updated on-the-fly.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
signal
|
str
|
Name of the signal to compare |
required |
variable
|
str
|
Name of the variable (without $ prefix) |
required |
Returns:
| Type | Description |
|---|---|
Formula
|
Formula representing: |
Note
Requires Incremental algorithm. The Naive algorithm does not support variables.
Examples:
>>> Formula.gt_var('x', 'threshold') # x > $threshold
implies(left, right)
staticmethod
Create an implication between two formulas.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
left
|
Formula
|
Antecedent (condition) |
required |
right
|
Formula
|
Consequent (result) |
required |
Returns:
| Type | Description |
|---|---|
Formula
|
Formula representing: |
Examples:
>>> # If x > 2.0, then y < 1.0
>>> Formula.implies(Formula.gt('x', 2.0), Formula.lt('y', 1.0))
lt(signal, value)
staticmethod
Create a less-than atomic predicate.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
signal
|
str
|
Name of the signal to compare |
required |
value
|
float
|
Threshold value to compare against |
required |
Returns:
| Type | Description |
|---|---|
Formula
|
Formula representing: |
Examples:
>>> Formula.lt('x', 0.5) # x < 0.5
lt_var(signal, variable)
staticmethod
Create a less-than atomic predicate with a variable threshold.
The variable value is looked up at runtime from a Variables object, allowing dynamic thresholds that can be updated on-the-fly.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
signal
|
str
|
Name of the signal to compare |
required |
variable
|
str
|
Name of the variable (without $ prefix) |
required |
Returns:
| Type | Description |
|---|---|
Formula
|
Formula representing: |
Note
Requires Incremental algorithm. The Naive algorithm does not support variables.
Examples:
>>> Formula.lt_var('y', 'limit') # y < $limit
not_(child)
staticmethod
or_(left, right)
staticmethod
true_()
staticmethod
Create a constant true formula.
Returns:
| Type | Description |
|---|---|
Formula
|
Formula that is always satisfied (⊤) |
Examples:
>>> Formula.true_()
until(start, end, left, right)
staticmethod
Create an until temporal formula.
The left formula must hold until the right formula becomes true, and the right formula must become true within the interval [start, end].
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
start
|
float
|
Start of time interval (seconds) |
required |
end
|
float
|
End of time interval (seconds) |
required |
left
|
Formula
|
Formula that must hold until right becomes true |
required |
right
|
Formula
|
Formula that must eventually become true |
required |
Returns:
| Type | Description |
|---|---|
Formula
|
Formula representing: |
Examples:
>>> # x > 0 must hold until y < 0.5 (within 5 seconds)
>>> Formula.until(0, 5, Formula.gt('x', 0), Formula.lt('y', 0.5))
Data Structures
mstlo_python.MonitorOutputDict
Bases: TypedDict
Output from a monitor update operation.
evaluations
instance-attribute
List of evaluations, one for each evaluated step. May be empty if data is buffered.
input_signal
instance-attribute
The signal name that was updated. None for an empty batch.
input_timestamp
instance-attribute
The timestamp of the input that triggered this update. None for an empty batch.
mstlo_python.EvaluationDict
Bases: TypedDict
Result of evaluating a single input step.
outputs
instance-attribute
List of output verdicts produced by evaluating this step.
sync_step_signal
instance-attribute
Signal name of the evaluated step.
sync_step_timestamp
instance-attribute
Timestamp of the evaluated step.
sync_step_value
instance-attribute
Value of the evaluated step.