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
121
122
123
124
125
126
127

//! Change detection workflow example.

use oxigdal_workflow::{
    WorkflowDefinition,
    dag::{ResourceRequirements, RetryPolicy, TaskEdge, TaskNode, WorkflowDag},
    scheduler::{ScheduleType, Scheduler},
    templates::{TemplateCategory, WorkflowTemplateLibrary},
};
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!("Change Detection Workflow Example");
    println!("==================================\n");

    // Create template library and get change detection template
    let library = WorkflowTemplateLibrary::new();

    println!("Available templates:");
    for template in library.list_templates() {
        println!("  - {} ({})", template.name, template.id);
    }
    println!();

    // Get change detection templates
    let change_templates = library.list_by_category(&TemplateCategory::ChangeDetection);

    if let Some(template) = change_templates.first() {
        println!("Using template: {}", template.name);
        println!("Description: {}", template.description);
        println!("Parameters:");
        for param in &template.parameters {
            println!(
                "  - {} ({}): {}",
                param.name,
                if param.required {
                    "required"
                } else {
                    "optional"
                },
                param.description
            );
        }
        println!();

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

        // Add tasks
        dag.add_task(create_task("load_before_image", "Load Before Image"))?;
        dag.add_task(create_task("load_after_image", "Load After Image"))?;
        dag.add_task(create_task("preprocess_images", "Preprocess Images"))?;
        dag.add_task(create_task("calculate_difference", "Calculate Difference"))?;
        dag.add_task(create_task("apply_threshold", "Apply Threshold"))?;
        dag.add_task(create_task("generate_change_map", "Generate Change Map"))?;

        // Add dependencies
        dag.add_dependency(
            "load_before_image",
            "preprocess_images",
            TaskEdge::default(),
        )?;
        dag.add_dependency("load_after_image", "preprocess_images", TaskEdge::default())?;
        dag.add_dependency(
            "preprocess_images",
            "calculate_difference",
            TaskEdge::default(),
        )?;
        dag.add_dependency(
            "calculate_difference",
            "apply_threshold",
            TaskEdge::default(),
        )?;
        dag.add_dependency(
            "apply_threshold",
            "generate_change_map",
            TaskEdge::default(),
        )?;

        let workflow = WorkflowDefinition {
            id: "change-detection-demo".to_string(),
            name: "Change Detection Demo".to_string(),
            description: Some("Detect changes between two time periods".to_string()),
            version: "1.0.0".to_string(),
            dag,
        };

        println!("Created workflow with {} tasks", workflow.dag.task_count());

        // Schedule the workflow
        let scheduler = Scheduler::with_defaults();

        let schedule_id = scheduler
            .add_schedule(
                workflow,
                ScheduleType::Interval {
                    interval_secs: 86400, // Daily
                },
            )
            .await?;

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

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

        println!("\nChange detection workflow configured successfully!");
    } else {
        println!("No change detection templates found in library");
    }

    Ok(())
}