streamduck_core/thread/
mod.rs

1//! Device Thread
2//!
3//! A separate thread for processing, rendering images on streamdeck and reading buttons
4
5use std::collections::HashMap;
6use std::io::Cursor;
7use std::ops::Deref;
8use std::sync::{Arc};
9use std::sync::mpsc::{channel, Sender, TryRecvError};
10use std::thread::spawn;
11use std::time::{Duration, Instant};
12use image::{DynamicImage, ImageFormat};
13use streamdeck::{Colour, DeviceImage, ImageMode, StreamDeck};
14use tokio::runtime::Builder;
15use tokio::sync::mpsc::UnboundedSender;
16use tokio::sync::RwLock;
17use rendering::RendererComponent;
18use crate::core::{CoreHandle, SDCore};
19use crate::core::button::{Component, parse_unique_button_to_component};
20use crate::images::SDImage;
21use crate::modules::core_module::CoreSettings;
22use crate::modules::UniqueSDModule;
23
24/// Rendering utilities
25pub mod util;
26pub mod rendering;
27
28/// Collection of images
29pub type ImageCollection = Arc<RwLock<HashMap<String, SDImage>>>;
30
31/// Handle for contacting renderer thread
32pub struct DeviceThreadHandle {
33    tx: Sender<Vec<DeviceThreadCommunication>>
34}
35
36impl DeviceThreadHandle {
37    /// Sends commands to device thread
38    pub fn send(&self, commands: Vec<DeviceThreadCommunication>) {
39        self.tx.send(commands).ok();
40    }
41}
42
43/// Enum of various operations that can be sent to device thread
44#[allow(dead_code)]
45pub enum DeviceThreadCommunication {
46    /// Tells renderer that screen should be updated
47    RefreshScreen,
48
49    /// Sets streamdeck brightness to provided value
50    SetBrightness(u8),
51
52    /// Sets button image to specified image
53    SetButtonImage(u8, DynamicImage),
54
55    /// Sets button image to raw buffer of image
56    SetButtonImageRaw(u8, Arc<DeviceImage>),
57
58    /// Clears button and sets it to black color
59    ClearButtonImage(u8),
60}
61
62/// Spawns device thread from a core reference
63pub fn spawn_device_thread(core: Arc<SDCore>, streamdeck: StreamDeck, key_tx: UnboundedSender<(u8, bool)>) -> DeviceThreadHandle {
64    let (tx, rx) = channel::<Vec<DeviceThreadCommunication>>();
65
66    spawn(move || {
67        let runtime = Builder::new_current_thread()
68            .enable_all()
69            .build()
70            .unwrap();
71
72        runtime.block_on(async {
73            let core = CoreHandle::wrap(core.clone());
74            let mut streamdeck = streamdeck;
75            let mut last_buttons = Vec::new();
76
77            streamdeck.set_blocking(false).ok();
78
79            let missing = rendering::draw_missing_texture(core.core.image_size);
80
81            let mut animation_counters = HashMap::new();
82            let mut last_iter = Instant::now();
83            let mut renderer_map = HashMap::new();
84            let mut animation_cache: HashMap<u64, (Arc<DeviceImage>, u64)> = HashMap::new();
85            let mut previous_state: HashMap<u8, u64> = HashMap::new();
86            let mut time = 0;
87            let mut last_time = time;
88            loop {
89                if core.core.is_closed().await {
90                    break;
91                }
92
93                // Reading commands
94                match rx.try_recv() {
95                    Ok(com) => {
96                        for com in com {
97                            match com {
98                                DeviceThreadCommunication::SetBrightness(brightness) => {
99                                    streamdeck.set_brightness(brightness).ok();
100                                }
101
102                                DeviceThreadCommunication::SetButtonImage(key, image) => {
103                                    let mut buffer = vec![];
104
105                                    image.write_to(&mut Cursor::new(&mut buffer), match streamdeck.kind().image_mode() {
106                                        ImageMode::Bmp => ImageFormat::Bmp,
107                                        ImageMode::Jpeg => ImageFormat::Jpeg,
108                                    }).ok();
109
110                                    streamdeck.write_button_image(key, &DeviceImage::from(buffer)).ok();
111                                }
112
113                                DeviceThreadCommunication::SetButtonImageRaw(key, image) => {
114                                    streamdeck.write_button_image(key, image.deref()).ok();
115                                }
116
117                                DeviceThreadCommunication::ClearButtonImage(key) => {
118                                    streamdeck.set_button_rgb(key, &Colour {
119                                        r: 0,
120                                        g: 0,
121                                        b: 0
122                                    }).ok();
123                                }
124
125                                DeviceThreadCommunication::RefreshScreen => {
126                                    let current_screen = core.get_current_screen().await;
127
128                                    if current_screen.is_none() {
129                                        return;
130                                    }
131
132                                    let current_screen = current_screen.unwrap();
133                                    let screen_handle = current_screen.read().await;
134                                    let current_screen = screen_handle.buttons.clone();
135                                    drop(screen_handle);
136
137                                    let core_settings: CoreSettings = core.config().get_plugin_settings().await.unwrap_or_default();
138
139                                    renderer_map.clear();
140
141                                    for (key, button) in current_screen {
142                                        let unwrapped_button = button.read().await;
143                                        if unwrapped_button.0.contains_key(RendererComponent::NAME) {
144                                            let names = unwrapped_button.component_names();
145                                            let mut modules = core.module_manager().get_modules_for_rendering(&names).await;
146                                            drop(unwrapped_button);
147
148                                            let component = parse_unique_button_to_component::<RendererComponent>(&button).await.unwrap();
149
150                                            modules.retain(|x, _| !component.plugin_blacklist.contains(x));
151                                            modules.retain(|x, _| !core_settings.renderer.plugin_blacklist.contains(x));
152
153                                            renderer_map.insert(key, (component, button, modules.into_values().collect::<Vec<UniqueSDModule>>()));
154                                        }
155                                    }
156
157                                    for (_, renderer) in core.core.render_manager.read_renderers().await.iter() {
158                                        renderer.refresh(&core).await;
159                                    }
160                                }
161                            }
162                        }
163                    }
164                    Err(err) => {
165                        match err {
166                            TryRecvError::Empty => {}
167                            TryRecvError::Disconnected => break,
168                        }
169                    }
170                }
171
172                rendering::process_frame(&core, &mut streamdeck, &mut animation_cache, &mut animation_counters, &mut renderer_map, &mut previous_state, &missing, time).await;
173                time += 1;
174
175                // Occasionally cleaning cache
176                if time % 3000 == 0 && time != last_time {
177                    animation_cache.retain(|_, (_, t)| *t > time);
178                }
179
180                last_time = time;
181
182                // Rate limiter
183                let rate = 1.0 / core.core.frame_rate as f32;
184                let time_since_last = last_iter.elapsed().as_secs_f32();
185                let to_wait = match rate - time_since_last {
186                    n if n < 0.0 => None,
187                    n => Some(Duration::from_secs_f32(n)),
188                };
189
190                // Reading buttons
191                match streamdeck.read_buttons(to_wait) {
192                    Ok(buttons) => {
193                        for (key, value) in buttons.iter().enumerate() {
194                            if let Some(last_value) = last_buttons.get(key) {
195                                if last_value != value {
196                                    if key_tx.send((key as u8, *last_value == 0)).is_err() {
197                                        log::error!("Key Handler task crashed, killing connection...");
198                                        core.core.close().await;
199                                    }
200                                }
201                            } else {
202                                if *value > 0 {
203                                    if key_tx.send((key as u8, true)).is_err() {
204                                        log::error!("Key Handler task crashed, killing connection...");
205                                        core.core.close().await;
206                                    }
207                                }
208                            }
209                        }
210                        last_buttons = buttons;
211                    }
212                    Err(err) => {
213                        match err {
214                            streamdeck::Error::NoData => {}
215                            streamdeck::Error::Hid(_) => {
216                                log::trace!("hid connection failed");
217                                core.core.close().await
218                            }
219                            _ => {
220                                panic!("Error on streamdeck thread: {:?}", err);
221                            }
222                        }
223                    }
224                }
225
226                last_iter = Instant::now();
227            }
228
229            log::trace!("rendering closed");
230        });
231    });
232
233    DeviceThreadHandle {
234        tx
235    }
236}
streamduck_core/thread/mod.rs

streamduck_core/thread/
mod.rs
