### Install Regelum

Source: https://aidagroup.github.io/regelum/latest

Install from PyPI with uv.

```bash
uv add regelum

```

--------------------------------

### Quick Example

Source: https://aidagroup.github.io/regelum/latest

A quick example demonstrating the usage of Regelum to simulate a temperature sensor, heater controller, and heating cycles.

```python
import regelum as rg


class TemperatureSensor(rg.Node):
    class State(rg.NodeState):
        temperature: float = rg.var(init=19.0)

    def update(self) -> State:
        return self.State(temperature=21.5)


class HeaterController(rg.Node):
    class Inputs(rg.NodeInputs):
        temperature: float = rg.src(TemperatureSensor.State.temperature)

    class State(rg.NodeState):
        heater_on: bool

    def update(self, inputs: Inputs) -> State:
        return self.State(heater_on=inputs.temperature < 22.0)


class HeatingCycles(rg.Node):
    class Inputs(rg.NodeInputs):
        heater_on: bool = rg.src(HeaterController.State.heater_on)

    class State(rg.NodeState):
        count: int = rg.var(init=0)

    def update(self, inputs: Inputs, prev_state: State) -> State:
        return self.State(
            count=prev_state.count + int(inputs.heater_on),
        )


sensor = TemperatureSensor(name="room_sensor")
controller = HeaterController(name="heater_controller")
cycles = HeatingCycles(name="heating_cycles")

system = rg.PhasedReactiveSystem(
    phases=[
        rg.Phase(
            "control",
            nodes=(sensor, controller, cycles),
            transitions=(rg.Goto(rg.terminate),),
            is_initial=True,
        ),
    ],
)

system.step()
print(system.read(controller.State.heater_on))

```

--------------------------------

### Step Records Example

Source: https://aidagroup.github.io/regelum/latest/concepts/compilation

Example output showing the records generated after a 30-tick run of the player system.

```python
for record in records:
    print(record.phase, record.node, record.state)
# measure Network {'bandwidth_kbps': 600.0}
# decide  QualityPolicy {'stalling': False}
# play    Decoder {'fetched_seconds': 0.278}
# play    MediaSession {'buffer_seconds': 9.11}
# play    Logger {'history': [...]} 

```

--------------------------------

### Phase and Transition Examples

Source: https://aidagroup.github.io/regelum/latest/concepts/phases

Examples demonstrating phase declarations with unconditional Goto transitions to other phases or termination.

```python
rg.Phase(
    "measure",
    nodes=(network,),
    transitions=(rg.Goto("decide"),),  # 
    is_initial=True,
)

rg.Phase(
    "play",
    nodes=(decoder, session, logger),
    transitions=(rg.Goto(rg.terminate),),  # 
)


```

--------------------------------

### Branch Chain Example

Source: https://aidagroup.github.io/regelum/latest/concepts/phases

An example demonstrating the use of If and Else for defining policy branches.

```python
rg.If(rg.V(policy.State.stalling), "drop_quality", name="stalling"),
    rg.Else("play", name="healthy")
```

--------------------------------

### Video player example with If and Else

Source: https://aidagroup.github.io/regelum/latest/concepts/phases

An example of a phase definition for a video player using If and Else for transitions based on a policy's state.

```python
rg.Phase(
    "decide",
    nodes=(policy,),
    transitions=(
        rg.If(rg.V(policy.State.stalling), "drop_quality", name="stalling"),
        rg.Else("play", name="healthy"),
    ),
)
```

--------------------------------

### Run Standalone Example

Source: https://aidagroup.github.io/regelum/latest/examples/controlled_pendulum

Command to run the standalone simulation example.

```bash
uv run python examples/controlled_pendulum/standalone.py
```

--------------------------------

### Decoder Node Example

Source: https://aidagroup.github.io/regelum/latest/concepts/phases

Example of a Decoder node with its State and update method.

```python
class Decoder(rg.Node):
    """The plant.  Buffer fills with newly fetched video, drains with playback."""

    class State(rg.NodeState):
        fetched_seconds: float

    def update(self, inputs: Inputs) -> State:
        bitrate = max(inputs.bitrate_kbps, 1)
        return self.State(fetched_seconds=inputs.bandwidth_kbps / bitrate * TICK_DT_SECONDS)
```

--------------------------------

### Build the runtime system

Source: https://aidagroup.github.io/regelum/latest/concepts/compilation

Example of building a phased reactive system with phases, nodes, and transitions.

```python
import regelum as rg


def build_system() -> rg.PhasedReactiveSystem:
    network = Network()
    policy = QualityPolicy()
    controller = BitrateController()
    decoder = Decoder()
    session = MediaSession()
    logger = Logger()

    return rg.PhasedReactiveSystem(
        phases=[
            rg.Phase(
                "measure",
                nodes=(network,),
                transitions=(rg.Goto("decide"),),
                is_initial=True,
            ),
            rg.Phase(
                "decide",
                nodes=(policy,),
                transitions=(
                    rg.If(
                        rg.V(policy.State.stalling),
                        "drop_quality",
                        name="stalling",
                    ),
                    rg.Else("play", name="healthy"),
                ),
            ),
            rg.Phase(
                "drop_quality",
                nodes=(controller,),
                transitions=(rg.Goto("play"),),
            ),
            rg.Phase(
                "play",
                nodes=(decoder, session, logger),
                transitions=(rg.Goto(rg.terminate),),
            ),
        ],
    )

```

--------------------------------

### Video Player Example

Source: https://aidagroup.github.io/regelum/latest/concepts/nodes

The full Python code for the adaptive-bitrate video player example, illustrating the use of various Regelum nodes to create a feedback control loop for video streaming.

```python
"""Adaptive-bitrate video player as a feedback control loop.

Each tick the player asks one question: do I have enough buffered video to
keep playing at the current quality?  If yes, the short branch just plays the
next chunk.  If no, the long branch first lowers the target bitrate, then
plays.  Buffer and bitrate persist across ticks, so the branching pattern is
driven by the closed loop between the network, the buffer, and the policy.

Phase graph::

    measure (init) -> decide -+-[healthy]--> play -> bottom
                              |
                              +-[stalling]-> drop_quality -> play -> bottom
"""

from __future__ import annotations

import regelum as rg

TICK_DT_SECONDS = 1.0
BITRATE_LADDER_KBPS = (240, 480, 720, 1080, 2160)
TOP_BITRATE_KBPS = BITRATE_LADDER_KBPS[-1]
STALL_HORIZON_SECONDS = 4.0


class Network(rg.Node):
    """Stochastic-looking but deterministic bandwidth model.

    Drops to a slow link in the middle of the run so the policy has to react,
    then recovers so the buffer can refill.  The schedule is fixed to keep the
    example reproducible.
    """

    class Inputs(rg.NodeInputs):
        tick: int = rg.src(rg.Clock.tick)

    class State(rg.NodeState):
        bandwidth_kbps: float = rg.var(init=float(TOP_BITRATE_KBPS))

    def update(self, inputs: Inputs) -> State:
        if inputs.tick < 6:
            value = 2400.0
        elif inputs.tick < 14:
            value = 600.0
        elif inputs.tick < 22:
            value = 1100.0
        else:
            value = 2400.0
        return self.State(bandwidth_kbps=value)


class QualityPolicy(rg.Node):
    """Decides whether the player is about to stall.

    The estimated drain rate is ``1 - bandwidth / bitrate`` seconds of video
    lost per wall-second.  If buffered seconds will not survive
    ``STALL_HORIZON_SECONDS`` at that drain rate, mark the tick as stalling.
    """

    class Inputs(rg.NodeInputs):
        buffer_seconds: float = rg.src(lambda: MediaSession.State.buffer_seconds)
        bitrate_kbps: int = rg.src(lambda: BitrateController.State.value)
        bandwidth_kbps: float = rg.src(Network.State.bandwidth_kbps)

    class State(rg.NodeState):
        stalling: bool = rg.var(init=False)

    def update(self, inputs: Inputs) -> State:
        bitrate = max(inputs.bitrate_kbps, 1)
        drain = max(0.0, 1.0 - inputs.bandwidth_kbps / bitrate)
        if drain <= 0.0:
            return self.State(stalling=False)
        time_to_empty = inputs.buffer_seconds / drain
        return self.State(stalling=time_to_empty < STALL_HORIZON_SECONDS)


class BitrateController(rg.Node):
    """Owns the current target bitrate.  Drops one rung when invoked."""

    class Inputs(rg.NodeInputs):
        current: int = rg.src(lambda: BitrateController.State.value)

    class State(rg.NodeState):
        value: int = rg.var(init=TOP_BITRATE_KBPS)

    def update(self, inputs: Inputs) -> State:
        try:
            index = BITRATE_LADDER_KBPS.index(inputs.current)
        except ValueError:
            index = len(BITRATE_LADDER_KBPS) - 1
        next_index = max(0, index - 1)
        return self.State(value=BITRATE_LADDER_KBPS[next_index])


class Decoder(rg.Node):
    """Models how many seconds of video can be downloaded in one tick.

    With ``bandwidth_kbps`` of throughput and a video encoded at
    ``bitrate_kbps``, one wall-second of downloading produces
    ``bandwidth / bitrate`` seconds of playable content.
    """

    class Inputs(rg.NodeInputs):
        bandwidth_kbps: float = rg.src(Network.State.bandwidth_kbps)
        bitrate_kbps: int = rg.src(lambda: BitrateController.State.value)

    class State(rg.NodeState):
        fetched_seconds: float

    def update(self, inputs: Inputs) -> State:
        bitrate = max(inputs.bitrate_kbps, 1)
        return self.State(fetched_seconds=inputs.bandwidth_kbps / bitrate * TICK_DT_SECONDS)


class MediaSession(rg.Node):
    """The plant.  Buffer fills with newly fetched video, drains with playback."""

    class Inputs(rg.NodeInputs):
        previous: float = rg.src(lambda: MediaSession.State.buffer_seconds)
        fetched: float = rg.src(Decoder.State.fetched_seconds)

    class State(rg.NodeState):
        buffer_seconds: float = rg.var(init=10.0)

    def update(self, inputs: Inputs) -> State:
        next_buffer = inputs.previous + inputs.fetched - TICK_DT_SECONDS
        return self.State(buffer_seconds=max(0.0, next_buffer))


class Logger(rg.Node):

```

--------------------------------

### System Step Execution

Source: https://aidagroup.github.io/regelum/latest/concepts/compilation

Example of executing a single step of the Regelum system.

```python
records = system.step()

```

--------------------------------

### Node Constructor Example

Source: https://aidagroup.github.io/regelum/latest/concepts/nodes

Example of a custom node constructor that forwards the 'name' argument to the base class.

```python
    def __init__(self, *, name: str | None = None) -> None:
        super().__init__(name=name)
```

--------------------------------

### Main Execution Example

Source: https://aidagroup.github.io/regelum/latest/concepts/nodes

This main function builds the system, runs it for a number of ticks, and prints the state snapshots.

```python
def main() -> None:
    system = build_system()
    print(f"compile_ok = {system.compile_report.ok}")
    print(
        "phase schedules: "
        + " | ".join(
            f"{name}={schedule}" for name, schedule in system.compile_report.phase_schedules.items()
        )
    )
    print()
    print("tick | bw(kbps) | bitrate | buffer(s) | stall? | path")
    print("-----+----------+---------+-----------+--------+----------------------")
    for _ in range(30):
        records = system.step()
        path_phases: list[str] = []
        for record in records:
            if not path_phases or path_phases[-1] != record.type:
                path_phases.append(record.type)
        path = " -> ".join(path_phases)
        snapshot = system.snapshot()
        print(
            f"{system.read(rg.Clock.tick):4d} | "
            f"{snapshot['Network.bandwidth_kbps']:8.0f} | "
            f"{snapshot['BitrateController.value']:7d} | "
            f"{snapshot['MediaSession.buffer_seconds']:9.2f} | "
            f"{str(snapshot['QualityPolicy.stalling']):>6} | "
            f"{path}"
        )


if __name__ == "__main__":
    main()
```

--------------------------------

### Instantiate Observer, Logger, and PhasedReactiveSystem

Source: https://aidagroup.github.io/regelum/latest/examples/free_pendulum

Sets up the observer, logger, and the main phased reactive system with initial and observation phases.

```python
observer = Observer()
logger = Logger()

system = rg.PhasedReactiveSystem(
    phases=[
        rg.Phase(
            "plant",
            nodes=(plant,),
            transitions=(rg.Goto("observe"),),
            is_initial=True,
        ),
        rg.Phase(
            "observe",
            nodes=(observer, logger),
            transitions=(rg.Goto(rg.terminate),),
        ),
    ],
    base_dt=BASE_DT,
)
```

--------------------------------

### System Initialization with Continuous Dynamics

Source: https://aidagroup.github.io/regelum/latest/concepts/continuous

Example demonstrating the initialization of a PhasedReactiveSystem with continuous dynamics, specifying base_dt and integrating an ODESystem.

```python
controller = Controller(dt="0.01")
electrical = rg.ODESystem(nodes=(plant,), dt="0.001")

system = rg.PhasedReactiveSystem(
    phases=phases,
    base_dt="auto",  # Fraction(1, 1000)
)
```

--------------------------------

### Video Player Compilation Report

Source: https://aidagroup.github.io/regelum/latest/concepts/compilation

Example of how to build the video player system and print its compilation report.

```python
from examples.video_player.video_player import build_system

system = build_system()
print(system.compile_report.format())

```

--------------------------------

### Running Multiple Ticks

Source: https://aidagroup.github.io/regelum/latest/concepts/compilation

Example of running multiple ticks (steps) of the Regelum system.

```python
system.run(steps=30)

```

--------------------------------

### Instantiate Plant and ODESystem

Source: https://aidagroup.github.io/regelum/latest/examples/free_pendulum

Instantiates the FreePendulum and wraps it in an ODESystem for integration.

```python
pendulum = FreePendulum()
plant = rg.ODESystem(nodes=(pendulum,), dt=BASE_DT)
```

--------------------------------

### Resetting System State

Source: https://aidagroup.github.io/regelum/latest/concepts/compilation

Examples of resetting the Regelum system state, with and without overrides.

```python
system.reset()
system.reset(initial_state={MediaSession.State.buffer_seconds: 5.0})

```

--------------------------------

### Phase Declaration Example

Source: https://aidagroup.github.io/regelum/latest/concepts/phases

An example of a phase declaration in Regelum, specifying its name, nodes, transitions, and initial status.

```python
rg.Phase(
    "measure",
    nodes=(network,),
    transitions=(rg.Goto("decide"),),
    is_initial=True,
)


```

--------------------------------

### Run Simulation and Read Samples

Source: https://aidagroup.github.io/regelum/latest/examples/free_pendulum

Runs the simulation for a specified duration and reads the logged samples.

```python
system = build_system()
system.run(700)
samples = system.read(Logger.State.samples)
```

--------------------------------

### Reading State Variables

Source: https://aidagroup.github.io/regelum/latest/concepts/compilation

Example of reading specific state variables and clock fields using system.read().

```python
buffer = system.read(session.State.buffer_seconds)
tick = system.read(rg.Clock.tick)
time = system.read(rg.Clock.time)

```

--------------------------------

### Adaptive-bitrate video player example

Source: https://aidagroup.github.io/regelum/latest/concepts/phases

Full code listing for the adaptive-bitrate video player, demonstrating feedback control loop concepts with phases.

```python
"""Adaptive-bitrate video player as a feedback control loop.

Each tick the player asks one question: do I have enough buffered video to
keep playing at the current quality?  If yes, the short branch just plays the
next chunk.  If no, the long branch first lowers the target bitrate, then
plays.  Buffer and bitrate persist across ticks, so the branching pattern is
driven by the closed loop between the network, the buffer, and the policy.

Phase graph::

    measure (init) -> decide -+-[healthy]--> play -> bottom
                              |
                              +-[stalling]-> drop_quality -> play -> bottom
"""

from __future__ import annotations

import regelum as rg

TICK_DT_SECONDS = 1.0
BITRATE_LADDER_KBPS = (240, 480, 720, 1080, 2160)
TOP_BITRATE_KBPS = BITRATE_LADDER_KBPS[-1]
STALL_HORIZON_SECONDS = 4.0


class Network(rg.Node):
    """Stochastic-looking but deterministic bandwidth model.

    Drops to a slow link in the middle of the run so the policy has to react,
    then recovers so the buffer can refill.  The schedule is fixed to keep the
    example reproducible.
    """

    class Inputs(rg.NodeInputs):
        tick: int = rg.src(rg.Clock.tick)

    class State(rg.NodeState):
        bandwidth_kbps: float = rg.var(init=float(TOP_BITRATE_KBPS))

    def update(self, inputs: Inputs) -> State:
        if inputs.tick < 6:
            value = 2400.0
        elif inputs.tick < 14:
            value = 600.0
        elif inputs.tick < 22:
            value = 1100.0
        else:
            value = 2400.0
        return self.State(bandwidth_kbps=value)


class QualityPolicy(rg.Node):
    """Decides whether the player is about to stall.

    The estimated drain rate is ``1 - bandwidth / bitrate`` seconds of video
    lost per wall-second.  If buffered seconds will not survive
    ``STALL_HORIZON_SECONDS`` at that drain rate, mark the tick as stalling.
    """

    class Inputs(rg.NodeInputs):
        buffer_seconds: float = rg.src(lambda: MediaSession.State.buffer_seconds)
        bitrate_kbps: int = rg.src(lambda: BitrateController.State.value)
        bandwidth_kbps: float = rg.src(Network.State.bandwidth_kbps)

    class State(rg.NodeState):
        stalling: bool = rg.var(init=False)

    def update(self, inputs: Inputs) -> State:
        bitrate = max(inputs.bitrate_kbps, 1)
        drain = max(0.0, 1.0 - inputs.bandwidth_kbps / bitrate)
        if drain <= 0.0:
            return self.State(stalling=False)
        time_to_empty = inputs.buffer_seconds / drain
        return self.State(stalling=time_to_empty < STALL_HORIZON_SECONDS)


class BitrateController(rg.Node):
    """Owns the current target bitrate.  Drops one rung when invoked."""

    class Inputs(rg.NodeInputs):
        current: int = rg.src(lambda: BitrateController.State.value)

    class State(rg.NodeState):
        value: int = rg.var(init=TOP_BITRATE_KBPS)

    def update(self, inputs: Inputs) -> State:
        try:
            index = BITRATE_LADDER_KBPS.index(inputs.current)
        except ValueError:
            index = len(BITRATE_LADDER_KBPS) - 1
        next_index = max(0, index - 1)
        return self.State(value=BITRATE_LADDER_KBPS[next_index])


class Decoder(rg.Node):
    """Models how many seconds of video can be downloaded in one tick.

    With ``bandwidth_kbps`` of throughput and a video encoded at
    ``bitrate_kbps``, one wall-second of downloading produces
    ``bandwidth / bitrate`` seconds of playable content.
    """

    class Inputs(rg.NodeInputs):
        bandwidth_kbps: float = rg.src(Network.State.bandwidth_kbps)
        bitrate_kbps: int = rg.src(lambda: BitrateController.State.value)

    class State(rg.NodeState):

```

--------------------------------

### MediaSession Node Example

Source: https://aidagroup.github.io/regelum/latest/concepts/nodes

Demonstrates a basic node with inputs and state variables using nested classes.

```python
class MediaSession(rg.Node):
    class Inputs(rg.NodeInputs):
        fetched: float = rg.src(Decoder.State.fetched_seconds)

    class State(rg.NodeState):
        buffer_seconds: float = rg.var(init=10.0)

```

--------------------------------

### Network and Decoder Node Example

Source: https://aidagroup.github.io/regelum/latest/concepts/nodes

Shows how inputs can reference state variables from other nodes.

```python
class Network(rg.Node):
    ...


class Decoder(rg.Node):
    class Inputs(rg.NodeInputs):
        bandwidth_kbps: float = rg.src(Network.State.bandwidth_kbps)

```

--------------------------------

### Logger Node Example

Source: https://aidagroup.github.io/regelum/latest/concepts/continuous

An example of a Logger node that sources tick, time, and integrator state, demonstrating how to access system clock and state variables within a continuous phase context.

```python
class Logger(rg.Node):
    class Inputs(rg.NodeInputs):
        tick: int = rg.src(rg.Clock.tick)
        time: float = rg.src(rg.Clock.time)
        x: float = rg.src(Integrator.State.x)
```

--------------------------------

### State Snapshot Access

Source: https://aidagroup.github.io/regelum/latest/concepts/compilation

Example of accessing the current state of the Regelum system using snapshot().

```python
snapshot = system.snapshot()
print(snapshot["MediaSession.buffer_seconds"])
print(snapshot["BitrateController.value"])

```

--------------------------------

### Extended chain with If, Elif, and Else

Source: https://aidagroup.github.io/regelum/latest/concepts/phases

An example showing a transition tuple with multiple conditions using If, Elif, and Else.

```python
transitions = (
    rg.If(rg.V(policy.State.stalling), "drop_quality"),
    rg.Elif(rg.V(policy.State.healthy_steady), "upgrade_quality"),
    rg.Else("play"),
)
```

--------------------------------

### Node Execution Example

Source: https://aidagroup.github.io/regelum/latest/concepts/nodes

This Python code demonstrates how a node might be executed within the system, printing snapshot information.

```python
path = " -> ".join(path_phases)
snapshot = system.snapshot()
print(
    f"{system.read(rg.Clock.tick):4d} | "
    f"{snapshot['Network.bandwidth_kbps']:8.0f} | "
    f"{snapshot['BitrateController.value']:7d} | "
    f"{snapshot['MediaSession.buffer_seconds']:9.2f} | "
    f"{str(snapshot['QualityPolicy.stalling']):>6} | "
    f"{path}"
)


if __name__ == "__main__":
    main()
```

--------------------------------

### Instance-Bound Reference Example

Source: https://aidagroup.github.io/regelum/latest/concepts/nodes

Illustrates using a specific instance reference for inputs when multiple producers of the same node class exist.

```python
class Network(rg.Node):
    ...


network_main = Network(name="main")
network_backup = Network(name="backup")


class Decoder(rg.Node):
    class Inputs(rg.NodeInputs):
        bandwidth_kbps: float = rg.src(network_main.State.bandwidth_kbps)

```

--------------------------------

### Mutating Inputs Example

Source: https://aidagroup.github.io/regelum/latest/concepts/nodes

Example demonstrating how a Logger node can mutate its own history input.

```Python
class Logger(rg.Node):
    class Inputs(rg.NodeInputs):
        history: list[Logger.Sample] = rg.src(lambda: Logger.State.history)

    class State(rg.NodeState):
        history: list[Logger.Sample] = rg.var(init=lambda: [])

    def update(self, inputs: Inputs) -> State:
        inputs.history.append(record)
        return self.State(history=inputs.history)

```

--------------------------------

### Vector State Example

Source: https://aidagroup.github.io/regelum/latest/concepts/continuous

Example of defining an ODENode with a vector-valued state using np.ndarray.

```python
class Filter(rg.ODENode):
    class Inputs(rg.NodeInputs):
        voltage: np.ndarray = rg.src(Inverter.State.phase_v)

    class State(rg.NodeState):
        current: np.ndarray = rg.var(init=lambda: np.zeros(3))

    def dstate(self, inputs: Inputs, state: State) -> State:
        return self.State(current=(inputs.voltage - R * state.current) / L)

```

--------------------------------

### Adaptive-bitrate video player

Source: https://aidagroup.github.io/regelum/latest/concepts/compilation

Full code listing for the adaptive-bitrate video player, demonstrating a feedback control loop.

```python
"""Adaptive-bitrate video player as a feedback control loop.

Each tick the player asks one question: do I have enough buffered video to
keep playing at the current quality?  If yes, the short branch just plays the
next chunk.  If no, the long branch first lowers the target bitrate, then
plays.  Buffer and bitrate persist across ticks, so the branching pattern is
driven by the closed loop between the network, the buffer, and the policy.

Phase graph::

    measure (init) -> decide -+-[healthy]--> play -> bottom
                              |
                              +-[stalling]-> drop_quality -> play -> bottom
"""

from __future__ import annotations

import regelum as rg

TICK_DT_SECONDS = 1.0
BITRATE_LADDER_KBPS = (240, 480, 720, 1080, 2160)
TOP_BITRATE_KBPS = BITRATE_LADDER_KBPS[-1]
STALL_HORIZON_SECONDS = 4.0


class Network(rg.Node):
    """Stochastic-looking but deterministic bandwidth model.

    Drops to a slow link in the middle of the run so the policy has to react,
    then recovers so the buffer can refill.  The schedule is fixed to keep the
    example reproducible.
    """

    class Inputs(rg.NodeInputs):
        tick: int = rg.src(rg.Clock.tick)

    class State(rg.NodeState):
        bandwidth_kbps: float = rg.var(init=float(TOP_BITRATE_KBPS))

    def update(self, inputs: Inputs) -> State:
        if inputs.tick < 6:
            value = 2400.0
        elif inputs.tick < 14:
            value = 600.0
        elif inputs.tick < 22:
            value = 1100.0
        else:
            value = 2400.0
        return self.State(bandwidth_kbps=value)


class QualityPolicy(rg.Node):
    """Decides whether the player is about to stall.

    The estimated drain rate is ``1 - bandwidth / bitrate`` seconds of video
    lost per wall-second.  If buffered seconds will not survive
    ``STALL_HORIZON_SECONDS`` at that drain rate, mark the tick as stalling.
    """

    class Inputs(rg.NodeInputs):
        buffer_seconds: float = rg.src(lambda: MediaSession.State.buffer_seconds)
        bitrate_kbps: int = rg.src(lambda: BitrateController.State.value)
        bandwidth_kbps: float = rg.src(Network.State.bandwidth_kbps)

    class State(rg.NodeState):
        stalling: bool = rg.var(init=False)

    def update(self, inputs: Inputs) -> State:
        bitrate = max(inputs.bitrate_kbps, 1)
        drain = max(0.0, 1.0 - inputs.bandwidth_kbps / bitrate)
        if drain <= 0.0:
            return self.State(stalling=False)
        time_to_empty = inputs.buffer_seconds / drain
        return self.State(stalling=time_to_empty < STALL_HORIZON_SECONDS)


class BitrateController(rg.Node):
    """Owns the current target bitrate.  Drops one rung when invoked."

    class Inputs(rg.NodeInputs):
        current: int = rg.src(lambda: BitrateController.State.value)

    class State(rg.NodeState):
        value: int = rg.var(init=TOP_BITRATE_KBPS)

    def update(self, inputs: Inputs) -> State:
        try:
            index = BITRATE_LADDER_KBPS.index(inputs.current)
        except ValueError:
            index = len(BITRATE_LADDER_KBPS) - 1
        next_index = max(0, index - 1)
        return self.State(value=BITRATE_LADDER_KBPS[next_index])


class Decoder(rg.Node):
    """Models how many seconds of video can be downloaded in one tick.

    With ``bandwidth_kbps`` of throughput and a video encoded at
    ``bitrate_kbps``, one wall-second of downloading produces
    ``bandwidth / bitrate`` seconds of playable content.
    """

    class Inputs(rg.NodeInputs):
        bandwidth_kbps: float = rg.src(Network.State.bandwidth_kbps)

```

--------------------------------

### ODENode Example

Source: https://aidagroup.github.io/regelum/latest/concepts/continuous

An example of defining an ODENode, including its Inputs, State, and dstate method. The dstate method defines the right-hand side of an ODE system.

```python
import casadi as ca
import numpy as np
import regelum as rg


class Integrator(rg.ODENode):
    class Inputs(rg.NodeInputs):
        u: np.ndarray = rg.src(Controller.State.u)

    class State(rg.NodeState):
        x: np.ndarray = rg.var(init=lambda: np.zeros(3))

    def dstate(self, inputs: Inputs, state: State, *, time: object) -> State:
        return self.State(x=A @ state.x + inputs.u + ca.sin(time))
```

--------------------------------

### Instantiate PhasedReactiveSystem

Source: https://aidagroup.github.io/regelum/latest/concepts/compilation

Basic instantiation of PhasedReactiveSystem.

```python
system = rg.PhasedReactiveSystem(phases=phases)

```

--------------------------------

### State Variable Declaration

Source: https://aidagroup.github.io/regelum/latest/concepts/nodes

Example of declaring a state variable with an initial value.

```python
class State(rg.NodeState):
    buffer_seconds: float = rg.var(init=10.0)

```

--------------------------------

### Handling missing initial state variables

Source: https://aidagroup.github.io/regelum/latest/concepts/compilation

Identifying missing initial state variables and providing runtime overrides.

```python
missing = system.compile_report.required_initial_state_vars
print(missing)

system.reset(
    initial_state={
        MediaSession.State.buffer_seconds: 5.0,
        BitrateController.State.value: 720,
    }
)

```

--------------------------------

### Enum state variable comparison

Source: https://aidagroup.github.io/regelum/latest/concepts/phases

Example of comparing an enum state variable within a symbolic predicate.

```python
from enum import Enum


class PlaybackMode(Enum):
    PLAYING = "playing"
    PAUSED = "paused"


rg.If(rg.V(state.State.mode) == PlaybackMode.PLAYING, "play")
```

--------------------------------

### Post-instantiation connections

Source: https://aidagroup.github.io/regelum/latest/concepts/nodes

Demonstrates connecting input ports to state variables after node instantiation.

```python
session_main = MediaSession(name="main")
session_pip = MediaSession(name="pip")
policy = QualityPolicy()

rg.port(policy.Inputs.buffer_seconds).connect(session_main.State.buffer_seconds)
```

--------------------------------

### Initial value for state variable

Source: https://aidagroup.github.io/regelum/latest/concepts/nodes

Example of defining an initial value for a state variable within a NodeState class.

```python
class MediaSession(rg.Node):
    class State(rg.NodeState):
        buffer_seconds: float = rg.var(init=10.0)
```

--------------------------------

### Callable predicates as an escape hatch

Source: https://aidagroup.github.io/regelum/latest/concepts/phases

Example of using a Python callable (lambda) as a predicate when a symbolic guard is not feasible.

```python
rg.If(lambda state: state.buffer_seconds < 2.0, "buffer_warning")
```

--------------------------------

### System Execution and Output

Source: https://aidagroup.github.io/regelum/latest/concepts/compilation

The `main` function compiles the system, prints compilation status and phase schedules, and then steps through the system for a fixed number of ticks, printing a snapshot of key states at each tick.

```python
def main() -> None:
    system = build_system()
    print(f"compile_ok = {system.compile_report.ok}")
    print(
        "phase schedules: "
        + " | ".join(
            f"{name}={schedule}" for name, schedule in system.compile_report.phase_schedules.items()
        )
    )
    print()
    print("tick | bw(kbps) | bitrate | buffer(s) | stall? | path")
    print("-----+----------+---------+-----------+--------+----------------------")
    for _ in range(30):
        records = system.step()
        path_phases: list[str] = []
        for record in records:
            if not path_phases or path_phases[-1] != record.phase:
                path_phases.append(record.phase)
        path = " -> ".join(path_phases)
        snapshot = system.snapshot()
        print(
            f"{system.read(rg.Clock.tick):4d} | "
            f"{snapshot['Network.bandwidth_kbps']:8.0f} | "
            f"{snapshot['BitrateController.value']:7d} | "
            f"{snapshot['MediaSession.buffer_seconds']:9.2f} | "
            f"{str(snapshot['QualityPolicy.stalling']):>6} | "
            f"{path}"
        )


if __name__ == "__main__":
    main()

```

--------------------------------

### ODENode with Lazy Source Reference

Source: https://aidagroup.github.io/regelum/latest/concepts/continuous

An example showing a lazy reference for an input source in the dstate method.

```python
def dstate(
    self,
    state: State,
    load: float = rg.src(lambda: Load.State.current),
) -> State:
    ...
```

--------------------------------

### Compile report with dependency edges

Source: https://aidagroup.github.io/regelum/latest/concepts/compilation

Formatting the compile report to include dependency edges.

```python
print(system.compile_report.format())

```

--------------------------------

### System Clock Inputs

Source: https://aidagroup.github.io/regelum/latest/concepts/nodes

Example of how to access the system clock's tick and time within a Node's Inputs.

```Python
class Network(rg.Node):
    class Inputs(rg.NodeInputs):
        tick: int = rg.src(rg.Clock.tick)
        time: float = rg.src(rg.Clock.time)

```

--------------------------------

### Compact update method with inputs

Source: https://aidagroup.github.io/regelum/latest/concepts/nodes

An example of a compact update method that declares inputs directly, suitable for small nodes.

```python
class TickCounter(rg.Node):
    class State(rg.NodeState):
        tick: int = rg.var(init=0)

    def update(
        self,
        tick: int = rg.src(lambda: TickCounter.State.tick),
    ) -> State:
        return self.State(tick=tick + 1)
```

--------------------------------

### Node Definitions and System Compilation

Source: https://aidagroup.github.io/regelum/latest/concepts/compilation

Defines various nodes (BitrateController, MediaSession, Logger) and their states, inputs, and update logic. It also includes the `build_system` function which constructs the phased reactive system with defined phases and transitions.

```python
bitrate_kbps: int = rg.src(lambda: BitrateController.State.value)

    class State(rg.NodeState):
        fetched_seconds: float

    def update(self, inputs: Inputs) -> State:
        bitrate = max(inputs.bitrate_kbps, 1)
        return self.State(fetched_seconds=inputs.bandwidth_kbps / bitrate * TICK_DT_SECONDS)


class MediaSession(rg.Node):
    """The plant.  Buffer fills with newly fetched video, drains with playback."""

    class Inputs(rg.NodeInputs):
        previous: float = rg.src(lambda: MediaSession.State.buffer_seconds)
        fetched: float = rg.src(Decoder.State.fetched_seconds)

    class State(rg.NodeState):
        buffer_seconds: float = rg.var(init=10.0)

    def update(self, inputs: Inputs) -> State:
        next_buffer = inputs.previous + inputs.fetched - TICK_DT_SECONDS
        return self.State(buffer_seconds=max(0.0, next_buffer))


class Logger(rg.Node):
    """Appends a per-tick record so the trajectory is visible after the run."""

    Sample = tuple[int, float, int, float, bool]

    class Inputs(rg.NodeInputs):
        tick: int = rg.src(rg.Clock.tick)
        bandwidth_kbps: float = rg.src(Network.State.bandwidth_kbps)
        bitrate_kbps: int = rg.src(lambda: BitrateController.State.value)
        buffer_seconds: float = rg.src(lambda: MediaSession.State.buffer_seconds)
        stalling: bool = rg.src(QualityPolicy.State.stalling)
        history: list["Logger.Sample"] = rg.src(lambda: Logger.State.history)

    class State(rg.NodeState):
        history: list["Logger.Sample"] = rg.var(init=lambda: [])

    def update(self, inputs: Inputs) -> State:
        record: Logger.Sample = (
            inputs.tick,
            inputs.bandwidth_kbps,
            inputs.bitrate_kbps,
            inputs.buffer_seconds,
            inputs.stalling,
        )
        inputs.history.append(record)
        return self.State(history=inputs.history)


def build_system() -> rg.PhasedReactiveSystem:
    network = Network()
    policy = QualityPolicy()
    controller = BitrateController()
    decoder = Decoder()
    session = MediaSession()
    logger = Logger()

    return rg.PhasedReactiveSystem(
        phases=[
            rg.Phase(
                "measure",
                nodes=(network,),
                transitions=(rg.Goto("decide"),),
                is_initial=True,
            ),
            rg.Phase(
                "decide",
                nodes=(policy,),
                transitions=(
                    rg.If(rg.V(policy.State.stalling), "drop_quality", name="stalling"),
                    rg.Else("play", name="healthy"),
                ),
            ),
            rg.Phase(
                "drop_quality",
                nodes=(controller,),
                transitions=(rg.Goto("play"),),
            ),
            rg.Phase(
                "play",
                nodes=(decoder, session, logger),
                transitions=(rg.Goto(rg.terminate),),
            ),
        ],
    )

```

--------------------------------

### Lazy Reference Example

Source: https://aidagroup.github.io/regelum/latest/concepts/nodes

Demonstrates using a lambda for lazy references, useful for self-referential inputs or when producers are defined later.

```python
class MediaSession(rg.Node):
    class Inputs(rg.NodeInputs):
        previous: float = rg.src(lambda: MediaSession.State.buffer_seconds)

```