rules¶

expression¶

Expression parser for Rules Model conditions.

Parses condition expressions including compound boolean expressions like:

"<simtime> == 34h"
"<clocktime> == 12:00"
"drinking_water.level >= 23"
"drinking_water.level <= 21 && drinking_water.level >= 23"
"(level < 10 || level > 90) && status == true"
"a > b" (attribute vs attribute)
"23 >= level" (literal vs attribute)

Both sides of a comparison can be an attribute name, a literal value, or a time variable.

Uses pyparsing with infix_notation for proper operator precedence.

class BooleanExpression(operator: str, operands: list[BooleanExpression | Comparison] = <factory>)¶

Bases: object

A compound boolean expression with AND/OR/NOT operators.

evaluate(simtime: float | None = None, clocktime: float | None = None, attributes: dict[str, Any] | None = None) → bool¶

Evaluate this boolean expression.

Parameters:

simtime – Simulation time in seconds since start
clocktime – Clock time in seconds since midnight
attributes – Dict mapping attribute names to values

Returns:

True if expression is satisfied

Return type:

bool

get_attribute_names() → set[str]¶

Return set of all attribute names referenced in this expression.

Returns:: Set of attribute names
Return type:: set[str]

get_time_thresholds() → list[tuple[ExpressionType, float]]¶

Return time thresholds from all operands recursively.

Returns:: List of (ExpressionType, threshold_value) tuples
Return type:: list[tuple[ExpressionType, float]]

operands: list[BooleanExpression | Comparison]¶

operator: str¶

class Comparison(left: ComparisonSide, operator: ComparisonOperator, right: ComparisonSide)¶

Bases: object

A single comparison expression (e.g., level >= 23).

evaluate(simtime: float | None = None, clocktime: float | None = None, attributes: dict[str, Any] | None = None) → bool¶

Evaluate this comparison.

Parameters:

simtime – Simulation time in seconds since start
clocktime – Clock time in seconds since midnight
attributes – Dict mapping attribute names to values

Returns:

True if comparison is satisfied

Return type:

bool

get_attribute_names() → set[str]¶

Return set of attribute names referenced by this comparison.

Returns:: Set of attribute names (empty if simtime/clocktime/literal only)
Return type:: set[str]

get_time_thresholds() → list[tuple[ExpressionType, float]]¶

Return time thresholds from this comparison.

If one side is SIMTIME/CLOCKTIME and the other is LITERAL, returns the literal value tagged with the time type.

Returns:: List of (ExpressionType, threshold_value) tuples
Return type:: list[tuple[ExpressionType, float]]

left: ComparisonSide¶

operator: ComparisonOperator¶

right: ComparisonSide¶

class ComparisonOperator(*values)¶

Bases: Enum

Comparison operators with their evaluation functions.

EQ = '=='¶

GE = '>='¶

GT = '>'¶

LE = '<='¶

LT = '<'¶

NE = '!='¶

evaluate(left: Any, right: Any) → bool¶

Evaluate the comparison between left and right values.

Parameters:

left – Left-hand side value
right – Right-hand side value

Returns:

Result of the comparison

Return type:

bool

class ComparisonSide(expr_type: ExpressionType, attribute_name: str | None = None, value: Any = None)¶

Bases: object

One side of a comparison expression.

attribute_name: str | None = None¶

expr_type: ExpressionType¶

value: Any = None¶

class ExpressionType(*values)¶

Bases: Enum

Type of one side of a condition expression.

ATTRIBUTE = 'attribute'¶

CLOCKTIME = 'clocktime'¶

LITERAL = 'literal'¶

SIMTIME = 'simtime'¶

parse_clock_time(expr: str) → float¶

Parse a clock time expression to seconds since midnight.

Supports formats like:

"12:30" → 45000 seconds (12 hours 30 minutes)
"08:15" → 29700 seconds
"23:59:59" → 86399 seconds

Parameters:: expr – Clock time expression string (HH:MM or HH:MM:SS)
Returns:: Seconds since midnight
Return type:: float
Raises:: ValueError – If the expression cannot be parsed

parse_condition(expr_string: str) → Comparison | BooleanExpression¶

Parse a condition expression string.

Supports simple comparisons and compound boolean expressions. Both sides of a comparison can be an attribute, literal, or time variable:

"<simtime> == 34h"
"level >= 23"
"a > b" (attribute vs attribute)
"23 >= level" (literal vs attribute)
"level <= 21 && level >= 23"
"(level < 10 || level > 90) && status == true"
"NOT status == true"

Parameters:: expr_string – Condition expression string
Returns:: Parsed condition (Comparison or BooleanExpression)
Return type:: ParsedCondition
Raises:: ValueError – If the expression cannot be parsed

parse_duration(expr: str) → float¶

Parse a duration expression to seconds.

Supports formats like:

"34h" → 122400 seconds
"5m" → 300 seconds
"1d5h30s" → 104430 seconds
"12h30m" → 45000 seconds

Parameters:: expr – Duration expression string (e.g., “1d5h30m”)
Returns:: Duration in seconds
Return type:: float
Raises:: ValueError – If the expression cannot be parsed

model¶

Rules Model for conditional attribute updates.

This model updates entity attributes based on conditions defined in rules datasets or model configuration.

class DatasetReferenceIndex(ids: 'dict[str, dict[int, int]]', references: 'dict[str, dict[str, int]]')¶

Bases: object

get_entity_idx_by_id(entity_id: int) → tuple[str, int] | tuple[None, None]¶

Retrieve location of an entity id :param entity_id: an entity id

Returns:: A tuple (entity group name, entity_idx)

get_entity_idx_by_reference(ref: str) → tuple[str, int] | tuple[None, None]¶

Retrieve location of an entity by reference :param ref: an entity reference

Returns:: A tuple (entity group name, entity_idx)

ids: dict[str, dict[int, int]]¶

references: dict[str, dict[str, int]]¶

class Model(config: dict)¶

Bases: TrackedModel

Model that applies conditional rules to update entity attributes.

Rules can be defined in the model config or in a separate rules dataset. Both sources are merged if both are provided.

Each rule specifies:

A condition (time-based or attribute-based)
A target entity (by id or reference)
An output attribute and value to set when the condition is true
Optionally, an else_value when the condition is false

rules: list[Rule]¶

setup(state: TrackedState, schema: AttributeSchema, settings: Settings, init_data_handler: InitDataHandler, logger: Logger, **__: Any) → None¶

Set up the model with state and schema.

Parameters:

state – TrackedState instance for managing entity data
schema – AttributeSchema for attribute registration
settings – Global settings
init_data_handler – Handler for initial data loading
logger – Logger instance

timeline_info: t.Optional[t.Any]¶

update(state: TrackedState, moment: Moment) → Moment | None¶

Update entity attributes based on rules.

Parameters:

state – TrackedState instance
moment – Current simulation moment

Returns:

Next moment when an update is needed, or None

Return type:

Optional[Moment]

Bases: object

A single rule that updates a target attribute based on a condition.

condition: Comparison | BooleanExpression¶

else_value: bool | int | float | str | None¶

from_entity_idx: int | None¶

classmethod from_spec(spec: RuleSpec, reference_indices: dict[str, DatasetReferenceIndex], state: TrackedState, schema: AttributeSchema)¶

output: str¶

output_array: UniformAttribute¶

source_attributes: dict[str, UniformAttribute]¶

to_entity_idx: int¶

value: bool | int | float | str¶

class RuleSpec(condition: 'str', output: 'str', value: 'bool | int | float | str | None', to_dataset: 't.Optional[str]' = None, from_dataset: 't.Optional[str]' = None, from_id: 't.Optional[int]' = None, from_reference: 't.Optional[str]' = None, to_id: 't.Optional[int]' = None, to_reference: 't.Optional[str]' = None, else_value: 't.Any' = None)¶

Bases: object

condition: str¶

else_value: Any = None¶

from_dataset: str | None = None¶

from_id: int | None = None¶

from_reference: str | None = None¶

output: str¶

to_dataset: str | None = None¶

to_id: int | None = None¶

to_reference: str | None = None¶

value: bool | int | float | str | None¶

exception RuleValidationError¶

Bases: ValueError

Raised when a rule specification is invalid.