Nivel 105: demo_level105.py

Este es el nivel 105 del tour de aprendizaje.

CĂłdigo Fuente

"""
DEMO LEVEL 105: States Analysis
-----------------------------
Adds: AnĂĄlisis de estados.
Continues: L104.

DIAGRAM:
analysis.get_states_analysis()
"""

from wpipe import Pipeline, step

@step(name="start")
def start(data: dict) -> None:

    """Start step.

    Args:

        data: Input data for the step.

    Returns:

        dict: Result of the step.

    """
    return {"estado": "iniciado"}

@step(name="verificar")
def verificar(data: dict) -> None:

    """Verificar step.

    Args:

        data: Input data for the step.

    Returns:

        dict: Result of the step.

    """
    return {"verificado": True}

if __name__ == "__main__":
    print(">>> AnĂĄlisis de estados...")

    pipe = Pipeline(
        pipeline_name="viaje_l105_states",
        verbose=True,
        tracking_db="output/states105.db",
    )
    pipe.set_steps([start, verificar])
    pipe.run({})

    states = pipe.tracker.analysis.get_states_analysis()
    print(f"\n📊 Estados: {states}")

Resultado de EjecuciĂłn


>>> AnĂĄlisis de estados...
[PIPELINE STATUS] Registered: PIPE-D79831AF
viaje_l105_states ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 100% 0:00:00
[PIPELINE STATUS] PIPE-D79831AF: COMPLETED

📊 Estados: {‘total_states’: 2, ‘total_executions’: 4, ‘total_errors’: 0, ‘most_used’: [{‘state_name’: ‘start’, ‘execution_count’: 2, ‘total_ms’: 0, ‘error_count’: 0, ‘avg_duration_ms’: 0.0}, {‘state_name’: ‘verificar’, ‘execution_count’: 2, ‘total_ms’: 0, ‘error_count’: 0, ‘avg_duration_ms’: 0.0}], ‘slowest’: [{‘state_name’: ‘start’, ‘execution_count’: 2, ‘total_ms’: 0, ‘error_count’: 0, ‘avg_duration_ms’: 0.0}, {‘state_name’: ‘verificar’, ‘execution_count’: 2, ‘total_ms’: 0, ‘error_count’: 0, ‘avg_duration_ms’: 0.0}], ‘most_errors’: []}