oxigdal-workflow 0.1.5

DAG-based workflow engine for complex geospatial processing pipelines
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120

//! Satellite imagery processing workflow example.

use oxigdal_workflow::{
    WorkflowDefinition,
    dag::{ResourceRequirements, RetryPolicy, TaskEdge, TaskNode, WorkflowDag},
    scheduler::{ScheduleType, Scheduler, SchedulerConfig},
};
use std::collections::HashMap;

fn create_task(id: &str, name: &str) -> TaskNode {
    TaskNode {
        id: id.to_string(),
        name: name.to_string(),
        description: None,
        config: serde_json::json!({}),
        retry: RetryPolicy::default(),
        timeout_secs: Some(60),
        resources: ResourceRequirements::default(),
        metadata: HashMap::new(),
    }
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    println!("Satellite Imagery Processing Workflow Example");
    println!("=============================================\n");

    // Create a workflow DAG
    let mut dag = WorkflowDag::new();

    // Add tasks
    dag.add_task(create_task("download_imagery", "Download Imagery"))?;
    dag.add_task(create_task(
        "atmospheric_correction",
        "Atmospheric Correction",
    ))?;
    dag.add_task(create_task("cloud_masking", "Cloud Masking"))?;
    dag.add_task(create_task("calculate_ndvi", "Calculate NDVI"))?;
    dag.add_task(create_task("export_results", "Export Results"))?;

    // Add dependencies
    dag.add_dependency(
        "download_imagery",
        "atmospheric_correction",
        TaskEdge::default(),
    )?;
    dag.add_dependency(
        "atmospheric_correction",
        "cloud_masking",
        TaskEdge::default(),
    )?;
    dag.add_dependency("cloud_masking", "calculate_ndvi", TaskEdge::default())?;
    dag.add_dependency("calculate_ndvi", "export_results", TaskEdge::default())?;

    // Create a workflow definition
    let workflow = WorkflowDefinition {
        id: "satellite-processing".to_string(),
        name: "Satellite Imagery Processing".to_string(),
        description: Some("Process Sentinel-2 satellite imagery".to_string()),
        version: "1.0.0".to_string(),
        dag,
    };

    println!("Created workflow: {}", workflow.name);
    println!(
        "Tasks: {:?}",
        workflow
            .dag
            .tasks()
            .iter()
            .map(|t| &t.id)
            .collect::<Vec<_>>()
    );
    println!();

    // Create scheduler
    let config = SchedulerConfig {
        max_concurrent_executions: 10,
        handle_missed_executions: true,
        max_missed_executions: 5,
        execution_timeout_secs: 3600,
        enable_persistence: false,
        persistence_path: None,
        tick_interval_ms: 100,
        timezone: "UTC".to_string(),
    };

    let scheduler = Scheduler::new(config);

    // Schedule the workflow to run daily
    let schedule_id = scheduler
        .add_schedule(
            workflow.clone(),
            ScheduleType::Cron {
                expression: "0 0 * * *".to_string(), // Daily at midnight
            },
        )
        .await?;

    println!("Scheduled workflow with ID: {}", schedule_id);

    // Trigger a manual execution
    let execution_id = scheduler.trigger_manual(&schedule_id).await?;
    println!("Triggered manual execution: {}", execution_id);

    // Get schedule information
    if let Some(schedule) = scheduler.get_schedule(&schedule_id) {
        println!("\nSchedule Information:");
        println!("  ID: {}", schedule.schedule_id);
        println!("  Enabled: {}", schedule.enabled);
        println!("  Type: {:?}", schedule.schedule_type);
        if let Some(next) = schedule.next_execution {
            println!("  Next execution: {}", next);
        }
    }

    println!("\nWorkflow scheduled successfully!");

    Ok(())
}