Nivel 8: demo_level8.py

Este es el nivel 8 del tour de aprendizaje.

ACTIVAR MODO INTERACTIVO

Código Fuente

"""
DEMO LEVEL 8: Branching (Condition)
-----------------------------------
Adds: Use of Condition() to execute True/False branches.

DIAGRAM:
(process_frame)
      |
      v
Condition(Obstacle detected?)
      |--- [True]  -> (emergency_brake)
      |--- [False] -> (gentle_accelerate)
"""

import random
from typing import Any, Dict, Generator, Tuple
import numpy as np
from wpipe import Condition, For, Pipeline, step, to_obj


def simulate_video() -> Generator[Tuple[int, np.ndarray], None, None]:
    """Simulate a video stream.

    Yields:
        Tuple[int, np.ndarray]: Frame ID and image.
    """
    for i in range(10):
        yield i, np.zeros((100, 100, 3), dtype=np.uint8)


@step(name="start_camera")
def start_camera(_data: Dict[str, Any]) -> Dict[str, Any]:
    """Start camera.

    Args:
        _data (Dict[str, Any]): The current pipeline context data.

    Returns:
        Dict[str, Any]: Stream generator.
    """
    return {"stream": simulate_video()}


@step(name="process_frame")
@to_obj
def process_frame(ctx: Any) -> Dict[str, Any]:
    """Process a frame and detect hazards.

    Args:
        ctx (Any): The context object.

    Returns:
        Dict[str, Any]: Frame ID and obstacle status.
    """
    try:
        frame_id, _ = next(ctx.stream)
        danger = random.random() < 0.3
        print(f"🖼️ Frame {frame_id} | Danger: {danger}")
        return {"current_frame": frame_id, "obstacle": danger}
    except StopIteration:
        return {"error": "End"}


@step(name="brake")
def brake(data: Dict[str, Any]) -> Dict[str, Any]:
    """Apply emergency brakes.

    Args:
        data (Dict[str, Any]): The current pipeline context data.

    Returns:
        Dict[str, Any]: Action performed.
    """
    print(f"🛑 EMERGENCY BRAKE ACTIVATED! Data: {data}")
    return {"action": "Braking"}


@step(name="accelerate")
def accelerate(data: Dict[str, Any]) -> Dict[str, Any]:
    """Accelerate smoothly.

    Args:
        data (Dict[str, Any]): The current pipeline context data.

    Returns:
        Dict[str, Any]: Action performed.
    """
    print(f"🛣️ Clear road. Accelerating... Data: {data}")
    return {"action": "Accelerating"}


if __name__ == "__main__":
    pipeline = Pipeline(pipeline_name="Trip_L8", verbose=True)
    pipeline.set_steps(
        [
            start_camera,
            For(
                iterations=5,
                steps=[
                    process_frame,
                    Condition(
                        expression="obstacle == True",
                        branch_true=[brake],
                        branch_false=[accelerate],
                    ),
                ],
            ),
        ]
    )
    pipeline.run({})

Resultado de Ejecución

---

🖼️ Frame 0 | Danger: True [CONDITION] Evaluating: obstacle == True 🛑 EMERGENCY BRAKE ACTIVATED! Data: {‘_pipeline_start_time’: ‘2026-04-30T13:36:32.231285’, ‘stream’: <generator object simulate_video at 0x72efe6a249e0>, ‘_loop_iteration’: 0, ‘progress_rich’: <rich.progress.Progress object at 0x72efe6a28c20>, ‘current_frame’: 0, ‘obstacle’: True} 🖼️ Frame 1 | Danger: False [CONDITION] Evaluating: obstacle == True 🛣️ Clear road. Accelerating… Data: {‘_pipeline_start_time’: ‘2026-04-30T13:36:32.231285’, ‘stream’: <generator object simulate_video at 0x72efe6a249e0>, ‘_loop_iteration’: 1, ‘progress_rich’: <rich.progress.Progress object at 0x72efe6a28c20>, ‘current_frame’: 1, ‘obstacle’: False, ‘action’: ‘Braking’} 🖼️ Frame 2 | Danger: False [CONDITION] Evaluating: obstacle == True 🛣️ Clear road. Accelerating… Data: {‘_pipeline_start_time’: ‘2026-04-30T13:36:32.231285’, ‘stream’: <generator object simulate_video at 0x72efe6a249e0>, ‘_loop_iteration’: 2, ‘progress_rich’: <rich.progress.Progress object at 0x72efe6a28c20>, ‘current_frame’: 2, ‘obstacle’: False, ‘action’: ‘Accelerating’} 🖼️ Frame 3 | Danger: True [CONDITION] Evaluating: obstacle == True 🛑 EMERGENCY BRAKE ACTIVATED! Data: {‘_pipeline_start_time’: ‘2026-04-30T13:36:32.231285’, ‘stream’: <generator object simulate_video at 0x72efe6a249e0>, ‘_loop_iteration’: 3, ‘progress_rich’: <rich.progress.Progress object at 0x72efe6a28c20>, ‘current_frame’: 3, ‘obstacle’: True, ‘action’: ‘Accelerating’} 🖼️ Frame 4 | Danger: True [CONDITION] Evaluating: obstacle == True 🛑 EMERGENCY BRAKE ACTIVATED! Data: {‘_pipeline_start_time’: ‘2026-04-30T13:36:32.231285’, ‘stream’: <generator object simulate_video at 0x72efe6a249e0>, ‘_loop_iteration’: 4, ‘progress_rich’: <rich.progress.Progress object at 0x72efe6a28c20>, ‘current_frame’: 4, ‘obstacle’: True, ‘action’: ‘Braking’} Trip_L8 ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 100% 0:00:00