thal 0.0.1

Reactive semantic runtime — molecules, reactions, and effect actors for building LLM-backed applications as dataflow programs.
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

//! Spinner Effect actor backed by `indicatif`.
//!
//! Reactions emit `Spinner { label: ..., running: true }` to start an
//! animation, `Spinner { running: false }` to stop. Singleton primary key —
//! the latest emit replaces the previous state via the molecule's merge
//! clause.

use super::EffectActor;
use crate::runtime::{Molecule, ReactorHandle};
use crate::value::Value;
use crate::Error;
use async_trait::async_trait;
use indicatif::{ProgressBar, ProgressStyle};
use parking_lot::Mutex;
use std::sync::Arc;
use std::time::Duration;

pub struct SpinnerActor {
    state: Arc<Mutex<Option<ProgressBar>>>,
}

impl SpinnerActor {
    pub fn new() -> Self {
        Self {
            state: Arc::new(Mutex::new(None)),
        }
    }
}

impl Default for SpinnerActor {
    fn default() -> Self {
        Self::new()
    }
}

#[async_trait]
impl EffectActor for SpinnerActor {
    fn kind_name(&self) -> &'static str {
        "Spinner"
    }

    async fn run(&self, request: Molecule, handle: ReactorHandle) -> Result<(), Error> {
        let running = request
            .fields
            .get("running")
            .and_then(|v| if let Value::Bool(b) = v { Some(*b) } else { None })
            .unwrap_or(false);
        let label = request
            .fields
            .get("label")
            .and_then(|v| if let Value::String(s) = v { Some(s.clone()) } else { None })
            .unwrap_or_default();

        {
            let mut state = self.state.lock();
            if running {
                if let Some(pb) = state.as_ref() {
                    pb.set_message(label);
                } else {
                    let pb = ProgressBar::new_spinner();
                    let style = ProgressStyle::with_template("{spinner:.cyan} {msg}")
                        .unwrap_or_else(|_| ProgressStyle::default_spinner());
                    pb.set_style(style.tick_strings(&[
                        "⠋", "⠙", "⠹", "⠸", "⠼", "⠴", "⠦", "⠧", "⠇", "⠏",
                    ]));
                    pb.set_message(label);
                    pb.enable_steady_tick(Duration::from_millis(80));
                    *state = Some(pb);
                }
            } else if let Some(pb) = state.take() {
                pb.finish_and_clear();
            }
        }

        let mut updated = request.clone();
        updated
            .fields
            .insert("status".into(), Value::String("Done".into()));
        handle.emit(updated).await
    }
}