kernel/fixture/
fixture_engine.rs

1use std::collections::hash_map::Entry;
2use std::collections::HashMap;
3use std::hash::{DefaultHasher, Hash, Hasher};
4use std::sync::{mpsc, OnceLock};
5use std::sync::mpsc::{Receiver, Sender};
6use std::thread;
7use common::fixture::{ChannelIndex, ChannelValue, Fixture, FixtureError, PropertyType, MAX_CHANNEL};
8use common::fixture::ChannelError::{ChannelAlreadyInUse, UniverseOutOfRange};
9use crate::fixture::fixture_command::FixtureCommand;
10use common::fixture::FixtureError::{DmxStateDesync, InvalidFixture};
11use common::logging::LogLevel::{Error, Info};
12use common::networking::subscription_objects::{DMXConfigForClientState, DMXConfigurationForClient};
13use common::{r_debug_log, r_log};
14use crate::fixture::fixture_engine::ChannelReservation::{Empty, Pending, Reserved};
15use crate::networking::on_dmx_config_update;
16
17/// Represents the reservation state of a single DMX-Channel.
18///
19/// * **Empty** – Channel is not in use.
20/// * **Pending(String)** – Channel has been claimed by a fixture but not yet finalized.
21/// * **Reserved(String, PropertyType, usize)** – Channel is fully reserved by a fixture with an associated property.
22#[derive(Clone, Debug)]
23enum ChannelReservation {
24    Empty,
25    Pending(String),
26    Reserved(String, PropertyType, usize),
27}
28
29/// The core engine managing fixture instances, thread synchronization, and DMX channel reservations.
30pub struct FixtureEngine {
31    fixtures: HashMap<String, Fixture>,
32    dmx_config: Vec<[ChannelReservation; MAX_CHANNEL as usize]>,
33    receiver: Receiver<FixtureCommand>,
34}
35
36/// Global static sender channel used to dispatch asynchronous commands to the running `FixtureEngine`.
37static FIXTURE_ACTION_SENDER: OnceLock<Sender<FixtureCommand>> = OnceLock::new();
38
39impl FixtureEngine {
40
41    /// Spawns the fixture engine actor thread and returns an interface receiver channel for updates.
42    ///
43    /// # Errors
44    ///
45    /// Returns an error if the engine has already been started (preventing multiple instances).
46    pub fn spawn() -> Result<Receiver<(usize,Vec<Fixture>)>, &'static str > {
47
48        if FIXTURE_ACTION_SENDER.get().is_some() {
49            return Err("Critical Error: The fixture engine has already been started!");
50        }
51
52        let (tx, rx) = mpsc::channel();
53
54        let mut engine = Self {
55            fixtures: HashMap::new(),
56            dmx_config: Vec::new(),
57            receiver: rx,
58        };
59
60        if FIXTURE_ACTION_SENDER.set(tx).is_err() {
61            return Err("Race condition: Engine was started in parallel!");
62        }
63
64        let (interface_sender, interface_receiver) = mpsc::channel();
65
66        thread::spawn(move || {
67            engine.run(interface_sender)
68        });
69
70        r_debug_log!(Info, "Fixture Engine Actor thread started successfully.");
71
72        Ok(interface_receiver)
73    }
74
75    /// Main event loop processing incoming commands and triggering state notifications.
76    ///
77    /// # Arguments
78    ///
79    /// * `interface_sender` - Channel used to broadcast updated fixture lists and universe counts to the runtime.
80    fn run(&mut self, interface_sender: Sender<(usize,Vec<Fixture>)>) {
81        while let Ok(command) = self.receiver.recv() {
82            let mut dmx_config_changed = false;
83            let mut dmx_values_changed = false;
84            match command {
85                FixtureCommand::SpawnFixture {name, fixture_type_name, start_channel, start_universe, reply_to } => {
86                    let result = self.new_fixture(name, fixture_type_name, start_channel, start_universe);
87                    reply_to.send(result.clone()).unwrap();
88                    if result.is_ok() {
89                        dmx_config_changed = true;
90                    }
91                }
92
93                FixtureCommand::MoveFixture {name, new_channel, new_universe, reply_to} => {
94                    let result = self.move_fixture(name, new_channel, new_universe);
95                    reply_to.send(result.clone()).unwrap();
96                    if result.is_ok() {
97                        dmx_config_changed = true;
98                    }
99                }
100
101                FixtureCommand::RemoveFixture {name, reply_to} => {
102                    let result = self.remove_fixture(name);
103                    reply_to.send(result.clone()).unwrap();
104                    if result.is_ok() {
105                        dmx_config_changed = true;
106                    }
107                }
108
109                FixtureCommand::SetProperty {fixture_name, property, value, reply_to} => {
110                    let result = self.set_property(fixture_name, property, value);
111                    reply_to.send(result.clone()).unwrap();
112                    if result.is_ok() {
113                        dmx_values_changed = true;
114                    }
115                }
116
117                FixtureCommand::GetType {fixture_name, reply_to} => {
118                    reply_to.send(self.get_fixture_type_from_string(fixture_name)).unwrap();
119                }
120            }
121
122            if dmx_config_changed {
123                dmx_values_changed = true;
124                on_dmx_config_update(self.get_dmx_config_for_client());
125            }
126
127            if dmx_values_changed {
128                let universe_count = self.dmx_config.len();
129                let fixtures: Vec<Fixture> = self.fixtures.values().cloned().collect();
130                interface_sender.send((universe_count, fixtures)).unwrap();
131            }
132
133        }
134    }
135
136    /// Creates and registers a new fixture instance, verifying and reserving its DMX channels.
137    ///
138    /// # Arguments
139    ///
140    /// * `name`              - Unique name for the fixture instance
141    /// * `fixture_type_name` - Template name of the registered fixture type
142    /// * `start_channel`     - DMX channel offset within the universe
143    /// * `start_universe`    - Target DMX universe index (0-based)
144    ///
145    /// # Errors
146    ///
147    /// * [`FixtureNameAlreadyInUse`](FixtureError::FixtureNameAlreadyInUse) – if the name is already taken
148    /// * [`InvalidFixtureType`](FixtureError::InvalidFixtureType) - if `fixture_type_name` is not registered
149    /// * [`ChannelAlreadyInUse`] – if required channels overlap with existing
150    /// * [`ChannelOutOfRange`](ChannelError::ChannelOutOfRange) - if any required channel is out of bounds
151    /// ([MAX_CHANNEL]).
152    /// * [`UniverseOutOfRange`] - if the fixture is in a non-existent universe
153    /// * [`DmxStateDesync`] - if the DMX-State and the fixture registry are out of sync
154    fn new_fixture(
155        &mut self, name: String, fixture_type_name: String, start_channel: ChannelIndex, start_universe: usize
156    ) -> Result<(), FixtureError> {
157        let fixture = Fixture::new(
158            fixture_type_name, start_channel,start_universe, name.clone()
159        )?;
160
161        let max_universe = fixture
162            .iter_over_properties()
163            .flat_map(|(_, channel)| channel.get_channel_indices())
164            .map(|(_, universe_index)| universe_index)
165            .max()
166            .unwrap_or(start_universe);
167
168        self.ensure_universe_size(max_universe + 1);
169
170        //Pending Reservation
171        fixture.iter_over_properties().try_for_each(|(_, channel)| {
172            for (channel_index, universe_index) in channel.get_channel_indices() {
173                let universe = self.dmx_config.get_mut(universe_index)
174                    .ok_or(UniverseOutOfRange)?;
175                if let Reserved(existing, property, _) = &universe[channel_index as usize] {
176                    return Err(ChannelAlreadyInUse(format!("{}, {}", existing, property)));
177                }
178
179                universe[channel_index as usize] = Pending(name.clone())
180            }
181
182            Ok(())
183        }).map_err(FixtureError::from)?;
184
185        // Insert into fixture list
186        if let Entry::Vacant(entry) = self.fixtures.entry(name.clone()) {
187            entry.insert(fixture.clone());
188        } else {
189            return Err(FixtureError::FixtureNameAlreadyInUse(name.clone()))
190        }
191
192        //Finalize Reservation
193        fixture.iter_over_properties().try_for_each(|(property, channel)| {
194            let mut fine_degree = 0;
195
196            for (channel_index, universe_index) in channel.get_channel_indices() {
197                let universe = self.dmx_config.get_mut(universe_index)
198                    .ok_or(UniverseOutOfRange)?;
199
200                let ch_index = channel_index as usize;
201
202                match &universe[ch_index] {
203                    Pending(existing) if *existing == name => {
204                        universe[ch_index] = Reserved(existing.clone(), property.clone(), fine_degree);
205                        fine_degree += 1;
206                    }
207
208                    _ => {
209                        return Err(DmxStateDesync);
210                    }
211                }
212            }
213
214            Ok(())
215        }).map_err(FixtureError::from)?;
216
217        Ok(())
218    }
219
220    /// Relocates an existing fixture to a new channel and/or universe.
221    ///
222    /// # Arguments
223    ///
224    /// * `name`          - Name of the fixture to move
225    /// * `new_channel`   - New starting DMX channel index
226    /// * `new_universe`  - New target DMX universe index (0-based)
227    ///
228    /// # Errors
229    ///
230    /// * [`InvalidFixture`] – if the fixture name is not found
231    /// * [`ChannelAlreadyInUse`] – if the new target channels are blocked
232    /// * [`ChannelOutOfRange`](ChannelError::ChannelOutOfRange) - if the new channel range is out of bounds
233    /// * [`UniverseOutOfRange`] - if the fixture is in a non-existent universe
234    /// * [`DmxStateDesync`] - if the DMX-State and the fixture registry are out of sync
235    fn move_fixture(
236        &mut self, name: String, new_channel: ChannelIndex, new_universe: usize
237    ) -> Result<(), FixtureError> {
238        let mut fixture_original = self.fixtures.get_mut(&name).ok_or(InvalidFixture(name.clone()))?
239            .clone();
240
241
242        let mut fixture_clone = fixture_original.clone();
243        fixture_clone.move_to_channel(new_channel, new_universe)?;
244
245        let max_universe = fixture_clone
246            .iter_over_properties()
247            .flat_map(|(_, channel)| channel.get_channel_indices())
248            .map(|(_, uni_idx)| uni_idx)
249            .max()
250            .unwrap_or(new_universe);
251
252        self.ensure_universe_size(max_universe + 1);
253
254        // Reserve Pending
255        fixture_clone.iter_over_properties().try_for_each(|(_, channel)| {
256
257            for (channel_index, universe_index) in channel.get_channel_indices() {
258                let universe = self.dmx_config.get_mut(universe_index)
259                    .ok_or(UniverseOutOfRange)?;
260
261                match &universe[channel_index as usize] {
262                    Reserved(existing, _, _) if existing == &name => {}
263                    Reserved(existing, property, _) => {
264                        return Err(ChannelAlreadyInUse(format!("{}, {}", existing, property)));
265                    }
266                    _ => {
267                        universe[channel_index as usize] = Pending(name.clone())
268                    }
269                }
270            }
271
272            Ok(())
273        }).map_err(FixtureError::from)?;
274
275        // Remove old Reservations
276        self.remove_reservations(&mut fixture_original)?;
277
278        // Reserve final
279        fixture_clone.iter_over_properties().try_for_each(|(property, channel)| {
280            let mut fine_degree = 0;
281
282            for (channel_index, universe_index) in channel.get_channel_indices() {
283                let universe = self.dmx_config.get_mut(universe_index)
284                    .ok_or(UniverseOutOfRange)?;
285
286                let ch_index = channel_index as usize;
287
288                match &universe[ch_index] {
289                    Pending(existing) if *existing == name => {
290                        universe[ch_index] = Reserved(name.clone(), property.clone(), fine_degree);
291                        fine_degree += 1;
292                    }
293
294                    Empty => {
295                        universe[ch_index] = Reserved(name.clone(), property.clone(), fine_degree);
296                        fine_degree += 1;
297                    }
298
299                    _ => {
300                        return Err(DmxStateDesync)
301                    }
302                }
303            }
304
305            Ok(())
306        }).map_err(FixtureError::from)?;
307
308        self.fixtures.insert(name, fixture_clone);
309
310        Ok(())
311    }
312
313    /// Removes a fixture instance and frees its associated DMX channel reservations.
314    ///
315    /// # Arguments
316    ///
317    /// * `name` - Name of the fixture to remove
318    ///
319    /// # Errors
320    ///
321    /// * [`InvalidFixture`] – if the fixture does not exist
322    /// * [`UniverseOutOfRange`] - if the fixture is in a non-existent universe
323    /// * [`DmxStateDesync`] - if the DMX-State and the fixture registry are out of sync
324    fn remove_fixture(&mut self, name: String) -> Result<(), FixtureError> {
325        let mut fixture = self.fixtures.remove(&name).ok_or(InvalidFixture(name.clone()))?;
326
327        if let Err(e) = self.remove_reservations(&mut fixture) {
328
329            //Rollback
330            self.fixtures.insert(name, fixture);
331            return Err(e);
332        }
333
334        Ok(())
335    }
336
337    /// Updates a specific property value on an active fixture.
338    ///
339    /// # Arguments
340    ///
341    /// * `name`     - Name of the target fixture
342    /// * `property` - Property type to update
343    /// * `value`    - New raw channel value
344    ///
345    /// # Errors
346    ///
347    /// * [`InvalidFixture`] – if the fixture is not found
348    /// * [`MissingProperty`](FixtureError::MissingProperty) – if the fixture lacks the specified property
349    fn set_property(&mut self, name: String, property: PropertyType, value: ChannelValue) -> Result<(), FixtureError> {
350
351        let fixture = self.fixtures.get_mut(&name).ok_or(InvalidFixture(name.clone()))?;
352
353        fixture.set(property, value)?;
354
355        Ok(())
356    }
357
358    /// Retrieves the fixture type name for a given fixture instance string.
359    ///
360    /// # Arguments
361    ///
362    /// * `name` - Name of the fixture instance
363    ///
364    /// # Errors
365    ///
366    /// * [`InvalidFixture`] – if no fixture with this name exists
367    fn get_fixture_type_from_string(&self, name: String) -> Result<String, FixtureError> {
368        match self.fixtures.get(&name) {
369            None => Err(InvalidFixture(name)),
370            Some(fixture) => Ok(fixture.get_fixture_type()),
371        }
372    }
373
374    /// Ensures that the internal DMX configuration vector has at least the given size (universes).
375    ///
376    /// # Arguments
377    ///
378    /// * `size` - Required minimum number of universes
379    fn ensure_universe_size(&mut self, size: usize) {
380        if size > self.dmx_config.len() {
381            self.dmx_config.resize_with(size, || std::array::from_fn(|_| Empty));
382        }
383    }
384
385    /// Clears all DMX channel reservations for a specific fixture.
386    ///
387    /// # Arguments
388    ///
389    /// * `fixture` - Reference to the fixture instance
390    ///
391    /// # Errors
392    ///
393    /// * [`UniverseOutOfRange`] - if the fixture is in a non-existent universe
394    /// * [`DmxStateDesync`] - if the DMX-State and the fixture registry are out of sync
395    fn remove_reservations(&mut self, fixture: &mut Fixture) -> Result<(), FixtureError> {
396        let name = fixture.get_name();
397
398        fixture.iter_over_properties().try_for_each(|(_, channel)| {
399            for (channel_index, universe_index) in channel.get_channel_indices() {
400                let universe = self.dmx_config.get_mut(universe_index)
401                    .ok_or(UniverseOutOfRange)?;
402
403                match &universe[channel_index as usize] {
404                    Reserved(existing, _, _) if *existing == *name => {}
405                    _ => {
406                        return Err(DmxStateDesync);
407                    }
408                }
409                universe[channel_index as usize] = Empty;
410            }
411
412            Ok(())
413        }).map_err(FixtureError::from)?;
414        Ok(())
415    }
416
417    /// Generates a snapshot of the current DMX configuration mapped for client subscription updates.
418    fn get_dmx_config_for_client(&self) -> DMXConfigForClientState{
419        self.dmx_config.iter().map(|universe| {
420            universe.iter().map(|channel| {
421                match channel {
422                    Reserved(fixture, property, fine_degree) => {
423                        let fixture_type = match self.fixtures.get(&fixture.clone()) {
424                            Some(fixture_object) => fixture_object.get_fixture_type(),
425                            None => {
426                                r_log!(Error,"Fixture {} is saved in DMXConfiguration, but not in  FixtureList.",
427                                        fixture
428                                    );
429                                return DMXConfigurationForClient::Empty;
430                            }
431                        };
432
433                        let mut hasher = DefaultHasher::new();
434                        fixture_type.hash(&mut hasher);
435                        let full_hash: u64 = hasher.finish();
436                        let fixture_type_hash = (full_hash % 256) as u8;
437
438                        DMXConfigurationForClient::Reserved {
439                            fixture_name: fixture.clone(),
440                            property_type: property.clone(),
441                            fine_degree: *fine_degree,
442                            fixture_type_hash
443                        }
444                    }
445                    _ => DMXConfigurationForClient::Empty,
446                }
447            }).collect()
448        }).collect()
449    }
450}
451
452/// Creates and registers a new fixture instance, verifying and reserving its DMX channels.
453///
454/// # Arguments
455///
456/// * `name`              - Unique name for the fixture instance
457/// * `fixture_type_name` - Template name of the registered fixture type
458/// * `start_channel`     - DMX channel offset within the universe
459/// * `start_universe`    - Target DMX universe index (0-based)
460///
461/// # Errors
462///
463/// * [`FixtureNameAlreadyInUse`](FixtureError::FixtureNameAlreadyInUse) – if the name is already taken
464/// * [`InvalidFixtureType`](FixtureError::InvalidFixtureType) - if `fixture_type_name` is not registered
465/// * [`ChannelAlreadyInUse`] – if required channels overlap with existing
466/// * [`ChannelOutOfRange`](ChannelError::ChannelOutOfRange) - if any required channel is out of bounds
467/// ([MAX_CHANNEL]).
468/// * [`UniverseOutOfRange`] - if the fixture is in a non-existent universe
469/// * [`DmxStateDesync`] - if the DMX-State and the fixture registry are out of sync
470
471pub fn new_fixture(
472    name: String, fixture_type_name: String, start_channel: ChannelIndex, start_universe: usize
473) -> Result<(), FixtureError> {
474    let (reply_tx, reply_rx) = mpsc::channel();
475
476    let cmd = FixtureCommand::SpawnFixture {
477        name,
478        fixture_type_name,
479        start_channel,
480        start_universe,
481        reply_to: reply_tx,
482    };
483
484    let engine_tx = FIXTURE_ACTION_SENDER.get().expect("CRITICAL: Engine not running!");
485    engine_tx.send(cmd).unwrap();
486
487    reply_rx.recv().unwrap()
488}
489
490/// Relocates an existing fixture to a new channel and/or universe.
491///
492/// # Arguments
493///
494/// * `name`          - Name of the fixture to move
495/// * `new_channel`   - New starting DMX channel index
496/// * `new_universe`  - New target DMX universe index (0-based)
497///
498/// # Errors
499///
500/// * [`InvalidFixture`] – if the fixture name is not found
501/// * [`ChannelAlreadyInUse`] – if the new target channels are blocked
502/// * [`ChannelOutOfRange`](ChannelError::ChannelOutOfRange) - if the new channel range is out of bounds
503/// * [`UniverseOutOfRange`] - if the fixture is in a non-existent universe
504/// * [`DmxStateDesync`] - if the DMX-State and the fixture registry are out of sync
505pub fn move_fixture(name: String, new_channel: ChannelIndex, new_universe: usize) -> Result<(), FixtureError> {
506    let (reply_tx, reply_rx) = mpsc::channel();
507
508    let cmd = FixtureCommand::MoveFixture {
509        name,
510        new_channel,
511        new_universe,
512        reply_to: reply_tx,
513    };
514
515    let engine_tx = FIXTURE_ACTION_SENDER.get().expect("CRITICAL: Engine not running!");
516    engine_tx.send(cmd).unwrap();
517
518    reply_rx.recv().unwrap()
519}
520
521/// Removes a fixture instance and frees its associated DMX channel reservations.
522///
523/// # Arguments
524///
525/// * `name` - Name of the fixture to remove
526///
527/// # Errors
528///
529/// * [`InvalidFixture`] – if the fixture does not exist
530/// * [`UniverseOutOfRange`] - if the fixture is in a non-existent universe
531/// * [`DmxStateDesync`] - if the DMX-State and the fixture registry are out of sync
532pub fn remove_fixture(name: String) -> Result<(), FixtureError> {
533    let (reply_tx, reply_rx) = mpsc::channel();
534
535    let cmd = FixtureCommand::RemoveFixture {
536        name,
537        reply_to: reply_tx,
538    };
539
540    let engine_tx = FIXTURE_ACTION_SENDER.get().expect("CRITICAL: Engine not running!");
541    engine_tx.send(cmd).unwrap();
542
543    reply_rx.recv().unwrap()
544}
545
546/// Updates a specific property value on an active fixture.
547///
548/// # Arguments
549///
550/// * `fixture_name` - Name of the target fixture
551/// * `property` - Property type to update
552/// * `value` - New raw channel value
553///
554/// # Errors
555///
556/// * [`InvalidFixture`] – if the fixture is not found
557/// * [`MissingProperty`](FixtureError::MissingProperty) – if the fixture lacks the specified property
558pub fn set_property(fixture_name: String, property: PropertyType, value: ChannelValue) -> Result<(), FixtureError> {
559    let (reply_tx, reply_rx) = mpsc::channel();
560
561    let cmd = FixtureCommand::SetProperty {
562        fixture_name,
563        property,
564        value,
565        reply_to: reply_tx
566    };
567
568    let engine_tx = FIXTURE_ACTION_SENDER.get().expect("CRITICAL: Engine not running!");
569    engine_tx.send(cmd).unwrap();
570
571    reply_rx.recv().unwrap()
572}
573
574/// Retrieves the fixture type name for a given fixture instance string.
575///
576/// # Arguments
577///
578/// * `name` - Name of the fixture instance
579///
580/// # Errors
581///
582/// * [`InvalidFixture`] – if no fixture with this name exists
583pub fn get_fixture_type(fixture_name: String) -> Result<String, FixtureError> {
584    let (reply_tx, reply_rx) = mpsc::channel();
585
586    let cmd = FixtureCommand::GetType {
587        fixture_name,
588        reply_to: reply_tx
589    };
590
591    let engine_tx = FIXTURE_ACTION_SENDER.get().expect("CRITICAL: Engine not running!");
592    engine_tx.send(cmd).unwrap();
593
594    reply_rx.recv().unwrap()
595}
kernel/fixture/fixture_engine.rs

kernel/fixture/
fixture_engine.rs
