Merge branch feat/tally into main

atem-connection-rs/src/atem.rs

@@ -1,20 +1,26 @@
 use std::{
     collections::{HashMap, VecDeque},
     net::SocketAddr,
+    ops::DerefMut,
     sync::Arc,
+    time::Duration,
 };
 
 use tokio::{select, sync::Semaphore};
 use tokio_util::sync::CancellationToken;
 
 use crate::{
-    atem_lib::atem_socket::{AtemEvent, AtemSocketCommand, AtemSocketMessage, TrackingId},
+    atem_lib::atem_socket::{
+        AtemSocket, AtemSocketCommand, AtemSocketEvent, AtemSocketMessage, TrackingId,
+    },
     commands::{
         command_base::{BasicWritableCommand, DeserializedCommand},
-        device_profile::DESERIALIZE_VERSION_RAW_NAME,
+        device_profile::version::DESERIALIZE_VERSION_RAW_NAME,
         init_complete::DESERIALIZE_INIT_COMPLETE_RAW_NAME,
+        parse_commands::deserialize_commands,
         time::DESERIALIZE_TIME_RAW_NAME,
     },
+    enums::ProtocolVersion,
     state::AtemState,
 };
 
@@ -27,13 +33,24 @@ pub enum AtemConnectionStatus {
 }
 
 pub struct Atem {
+    protocol_version: tokio::sync::RwLock<ProtocolVersion>,
+
+    socket: tokio::sync::RwLock<AtemSocket>,
+
     waiting_semaphores: tokio::sync::RwLock<HashMap<TrackingId, Arc<Semaphore>>>,
     socket_message_tx: tokio::sync::mpsc::Sender<AtemSocketMessage>,
 }
 
 impl Atem {
-    pub fn new(socket_message_tx: tokio::sync::mpsc::Sender<AtemSocketMessage>) -> Self {
+    pub fn new(
+        socket: AtemSocket,
+        socket_message_tx: tokio::sync::mpsc::Sender<AtemSocketMessage>,
+    ) -> Self {
         Self {
+            protocol_version: tokio::sync::RwLock::new(ProtocolVersion::V7_2),
+
+            socket: tokio::sync::RwLock::new(socket),
+
             waiting_semaphores: tokio::sync::RwLock::new(HashMap::new()),
             socket_message_tx,
         }
@@ -54,25 +71,30 @@ impl Atem {
 
     pub async fn run(
         &self,
-        mut atem_event_rx: tokio::sync::mpsc::UnboundedReceiver<AtemEvent>,
+        mut atem_event_rx: tokio::sync::mpsc::UnboundedReceiver<AtemSocketEvent>,
         cancel: CancellationToken,
     ) {
         let mut status = AtemConnectionStatus::default();
         let mut state = AtemState::default();
 
+        let mut poll_interval = tokio::time::interval(Duration::from_millis(5));
+
         while !cancel.is_cancelled() {
+            let tick = poll_interval.tick();
             select! {
                 _ = cancel.cancelled() => {},
+                _ = tick => {},
                 message = atem_event_rx.recv() => match message {
                     Some(event) => match event {
-                        AtemEvent::Connected => {
+                        AtemSocketEvent::Connected => {
                             log::info!("Atem connected");
                         }
-                        AtemEvent::Disconnected => todo!(),
+                        AtemSocketEvent::Disconnected => todo!("Disconnected"),
-                        AtemEvent::ReceivedCommands(commands) => {
+                        AtemSocketEvent::ReceivedCommands(payload) => {
+                            let commands = deserialize_commands(&payload, self.protocol_version.write().await.deref_mut());
                             self.mutate_state(&mut state, &mut status, commands).await
                         }
-                        AtemEvent::AckedCommand(tracking_id) => {
+                        AtemSocketEvent::AckedCommand(tracking_id) => {
                             log::debug!("Received tracking Id {tracking_id}");
                             if let Some(semaphore) =
                                 self.waiting_semaphores.read().await.get(&tracking_id)
@@ -89,18 +111,19 @@ impl Atem {
                     }
                 }
             }
+
+            self.socket.write().await.poll().await;
         }
     }
 
     pub async fn send_commands(&self, commands: Vec<Box<dyn BasicWritableCommand>>) {
+        let protocol_version = { self.protocol_version.read().await.clone() };
         let (callback_tx, callback_rx) = tokio::sync::oneshot::channel();
         self.socket_message_tx
             .send(AtemSocketMessage::SendCommands {
                 commands: commands
                     .iter()
-                    .map(|command| {
+                    .map(|command| AtemSocketCommand::new(command, &protocol_version))
-                        AtemSocketCommand::new(command, &crate::enums::ProtocolVersion::Unknown)
-                    })
                     .collect(),
                 tracking_ids_callback: callback_tx,
             })
@@ -142,14 +165,19 @@ impl Atem {
         for command in commands {
             match command.raw_name() {
                 DESERIALIZE_VERSION_RAW_NAME => {
+                    log::debug!("Received version response");
                     *state = AtemState::default();
                     *status = AtemConnectionStatus::Connecting
                 }
-                DESERIALIZE_INIT_COMPLETE_RAW_NAME => *status = AtemConnectionStatus::Connected,
+                DESERIALIZE_INIT_COMPLETE_RAW_NAME => {
+                    log::debug!("Received init complete from ATEM");
+                    *status = AtemConnectionStatus::Connected
+                }
                 DESERIALIZE_TIME_RAW_NAME => {
                     todo!("Time command")
                 }
                 _ => {
+                    log::debug!("Applying {} to state", command.raw_name());
                     command.apply_to_state(state);
                 }
             }

atem-connection-rs/src/atem_lib/atem_socket.rs

@@ -54,10 +54,10 @@ pub struct TrackingIdsCallback {
 }
 
 #[derive(Clone)]
-pub enum AtemEvent {
+pub enum AtemSocketEvent {
     Connected,
     Disconnected,
-    ReceivedCommands(VecDeque<Arc<dyn DeserializedCommand>>),
+    ReceivedCommands(Vec<u8>),
     AckedCommand(TrackingId),
 }
 
@@ -82,7 +82,7 @@ pub struct AtemSocketCommand {
 }
 
 impl AtemSocketCommand {
-    pub fn new(command: &Box<dyn BasicWritableCommand>, version: &ProtocolVersion) -> Self {
+    pub fn new<C: BasicWritableCommand>(command: &C, version: &ProtocolVersion) -> Self {
         Self {
             payload: command.payload(version),
             raw_name: command.get_raw_name().to_string(),
@@ -103,14 +103,19 @@ pub struct AtemSocket {
     socket: Option<UdpSocket>,
     address: SocketAddr,
 
+    protocol_version: ProtocolVersion,
+
     last_received_at: SystemTime,
     last_received_packed_id: u16,
     in_flight: Vec<InFlightPacket>,
     ack_timer: Option<SystemTime>,
     received_without_ack: u16,
 
-    atem_event_tx: tokio::sync::mpsc::UnboundedSender<AtemEvent>,
+    atem_message_rx: tokio::sync::mpsc::Receiver<AtemSocketMessage>,
+    atem_event_tx: tokio::sync::mpsc::UnboundedSender<AtemSocketEvent>,
     connected_callbacks: Mutex<Vec<tokio::sync::oneshot::Sender<bool>>>,
+
+    tick_interval: tokio::time::Interval,
 }
 
 #[derive(PartialEq, Clone)]
@@ -145,7 +150,11 @@ enum AtemSocketReceiveError {
 }
 
 impl AtemSocket {
-    pub fn new(atem_event_tx: tokio::sync::mpsc::UnboundedSender<AtemEvent>) -> Self {
+    pub fn new(
+        atem_message_rx: tokio::sync::mpsc::Receiver<AtemSocketMessage>,
+        atem_event_tx: tokio::sync::mpsc::UnboundedSender<AtemSocketEvent>,
+    ) -> Self {
+        let tick_interval = tokio::time::interval(Duration::from_millis(5));
         Self {
             connection_state: ConnectionState::Closed,
             reconnect_timer: None,
@@ -159,80 +168,80 @@ impl AtemSocket {
             socket: None,
             address: "0.0.0.0:0".parse().unwrap(),
 
+            protocol_version: ProtocolVersion::V7_2,
+
             last_received_at: SystemTime::now(),
             last_received_packed_id: 0,
             in_flight: vec![],
             ack_timer: None,
             received_without_ack: 0,
 
+            atem_message_rx,
             atem_event_tx,
             connected_callbacks: Mutex::default(),
+
+            tick_interval,
         }
     }
 
-    pub async fn run(
+    pub async fn poll(&mut self) {
-        &mut self,
+        let tick = self.tick_interval.tick();
-        mut atem_message_rx: tokio::sync::mpsc::Receiver<AtemSocketMessage>,
+        select! {
-        cancel: tokio_util::sync::CancellationToken,
+            _ = tick => {},
-    ) {
+            message = self.atem_message_rx.recv() => {
-        while !cancel.is_cancelled() {
+                match message {
-            select! {
+                    Some(AtemSocketMessage::Connect {
-                _ = cancel.cancelled() => {},
+                        address,
-                message = atem_message_rx.recv() => {
+                        result_callback,
-                    match message {
+                    }) => {
-                        Some(AtemSocketMessage::Connect {
+                        {
-                            address,
+                            let mut connected_callbacks = self.connected_callbacks.lock().await;
-                            result_callback,
+                            connected_callbacks.push(result_callback);
-                        }) => {
-                            {
-                                let mut connected_callbacks = self.connected_callbacks.lock().await;
-                                connected_callbacks.push(result_callback);
-                            }
-                            if self.connect(address).await.is_err() {
-                                let mut connected_callbacks = self.connected_callbacks.lock().await;
-                                for callback in connected_callbacks.drain(0..) {
-                                    let _ = callback.send(false);
-                                }
-                            }
                         }
-                        Some(AtemSocketMessage::Disconnect) => self.disconnect(),
+                        if self.connect(address).await.is_err() {
-                        Some(AtemSocketMessage::SendCommands {
+                            log::debug!("Connect failed");
-                            commands,
+                            let mut connected_callbacks = self.connected_callbacks.lock().await;
-                            tracking_ids_callback,
+                            for callback in connected_callbacks.drain(0..) {
-                        }) => {
+                                let _ = callback.send(false);
-                            let barrier = Arc::new(Barrier::new(2));
+                            }
-                            tracking_ids_callback
-                                .send(TrackingIdsCallback {
-                                    tracking_ids: self.send_commands(commands).await,
-                                    barrier: barrier.clone(),
-                                })
-                                .ok();
-
-                            // Let's play the game "Synchronisation Shenanigans"!
-                            // So, we are sending tracking Ids to the sender of this message, the sender will then wait
-                            // for each of these tracking Ids to be ACK'd by the ATEM. However, the sender will need to
-                            // do ✨ some form of shenanigans ✨ in order to be ready to receive tracking Ids. So we send
-                            // them a barrier as part of the callback so that they can tell us that they are ready for
-                            // us to continue with ATEM communication, at which point we may immediately inform them of a
-                            // received tracking Id matching one included in this callback.
-                            //
-                            // Now, if we were being 🚩 Real Proper Software Developers 🚩 we'd probably expect the receiver
-                            // of the callback to do clever things so that if a tracking Id is received immediately, they
-                            // then wait for something that wants that tracking Id on their side, rather than blocking this
-                            // task so that the caller can do ✨ shenanigans ✨. However, that sounds far too clever and too
-                            // much like 🚩 Real Actual Work 🚩 so instead we've chosen to do this and hope that whichever
-                            // actor we're waiting on doesn't take _too_ long to do ✨ shenanigans ✨ before signalling that
-                            // they are ready. If they do, I suggest finding whoever wrote that code and bonking them 🔨.
-                            barrier.wait().await;
-                        },
-                        None => {
-                            log::info!("ATEM message channel has closed, exiting event loop.");
-                            cancel.cancel();
                         }
                     }
+                    Some(AtemSocketMessage::Disconnect) => self.disconnect(),
+                    Some(AtemSocketMessage::SendCommands {
+                        commands,
+                        tracking_ids_callback,
+                    }) => {
+                        let barrier = Arc::new(Barrier::new(2));
+                        tracking_ids_callback
+                            .send(TrackingIdsCallback {
+                                tracking_ids: self.send_commands(commands).await,
+                                barrier: barrier.clone(),
+                            })
+                            .ok();
+
+                        // Let's play the game "Synchronisation Shenanigans"!
+                        // So, we are sending tracking Ids to the sender of this message, the sender will then wait
+                        // for each of these tracking Ids to be ACK'd by the ATEM. However, the sender will need to
+                        // do ✨ some form of shenanigans ✨ in order to be ready to receive tracking Ids. So we send
+                        // them a barrier as part of the callback so that they can tell us that they are ready for
+                        // us to continue with ATEM communication, at which point we may immediately inform them of a
+                        // received tracking Id matching one included in this callback.
+                        //
+                        // Now, if we were being 🚩 Real Proper Software Developers 🚩 we'd probably expect the receiver
+                        // of the callback to do clever things so that if a tracking Id is received immediately, they
+                        // then wait for something that wants that tracking Id on their side, rather than blocking this
+                        // task so that the caller can do ✨ shenanigans ✨. However, that sounds far too clever and too
+                        // much like 🚩 Real Actual Work 🚩 so instead we've chosen to do this and hope that whichever
+                        // actor we're waiting on doesn't take _too_ long to do ✨ shenanigans ✨ before signalling that
+                        // they are ready. If they do, I suggest finding whoever wrote that code and bonking them 🔨.
+                        barrier.wait().await;
+                    },
+                    None => {
+                        log::info!("ATEM message channel has closed.");
+                    }
                 }
-            };
+            }
-        }
+        };
 
         self.tick().await;
     }
@@ -393,7 +402,7 @@ impl AtemSocket {
             self.connection_state = ConnectionState::Established;
             self.last_received_packed_id = remote_packet_id;
             self.send_ack(remote_packet_id).await;
-            self.on_connect();
+            self.on_connect().await;
             return;
         }
 
@@ -528,34 +537,33 @@ impl AtemSocket {
     }
 
     fn on_commands_received(&mut self, payload: &[u8]) {
-        let commands = deserialize_commands(payload);
-
         let _ = self
             .atem_event_tx
-            .send(AtemEvent::ReceivedCommands(commands));
+            .send(AtemSocketEvent::ReceivedCommands(payload.to_vec()));
     }
 
     fn on_command_acknowledged(&mut self, packets: Vec<AckedPacket>) {
         for ack in packets {
             let _ = self
                 .atem_event_tx
-                .send(AtemEvent::AckedCommand(TrackingId(ack.tracking_id)));
+                .send(AtemSocketEvent::AckedCommand(TrackingId(ack.tracking_id)));
         }
     }
 
-    fn on_connect(&mut self) {
+    async fn on_connect(&mut self) {
-        let _ = self.atem_event_tx.send(AtemEvent::Connected);
+        let _ = self.atem_event_tx.send(AtemSocketEvent::Connected);
-        let mut connected_callbacks = self.connected_callbacks.blocking_lock();
+        let mut connected_callbacks = self.connected_callbacks.lock().await;
         for callback in connected_callbacks.drain(0..) {
             let _ = callback.send(false);
         }
     }
 
     fn on_disconnect(&mut self) {
-        let _ = self.atem_event_tx.send(AtemEvent::Disconnected);
+        let _ = self.atem_event_tx.send(AtemSocketEvent::Disconnected);
     }
 
     fn start_timers(&mut self) {
+        log::debug!("Starting timers");
         self.start_reconnect_timer();
         self.start_retransmit_timer();
     }

atem-connection-rs/src/commands/command_base.rs

@@ -1,4 +1,4 @@
-use std::{collections::HashMap, fmt::Debug, sync::Arc};
+use std::{collections::HashMap, fmt::Debug, process::Command, sync::Arc};
 
 use crate::{enums::ProtocolVersion, state::AtemState};
 
@@ -8,18 +8,51 @@ pub trait DeserializedCommand: Send + Sync + Debug {
 }
 
 pub trait CommandDeserializer: Send + Sync {
-    fn deserialize(&self, buffer: &[u8]) -> Arc<dyn DeserializedCommand>;
+    fn deserialize(&self, buffer: &[u8], version: &ProtocolVersion)
+        -> Arc<dyn DeserializedCommand>;
 }
 
-pub trait SerializableCommand {
+pub trait SerializableCommand: Send + Sync {
     fn payload(&self, version: &ProtocolVersion) -> Vec<u8>;
 }
 
-pub trait BasicWritableCommand: SerializableCommand {
+impl<C: SerializableCommand + ?Sized> SerializableCommand for Box<C> {
+    fn payload(&self, version: &ProtocolVersion) -> Vec<u8> {
+        (**self).payload(version)
+    }
+}
+
+impl<C: SerializableCommand + ?Sized> SerializableCommand for &'_ Box<C> {
+    fn payload(&self, version: &ProtocolVersion) -> Vec<u8> {
+        (**self).payload(version)
+    }
+}
+
+pub trait BasicWritableCommand: SerializableCommand + Send + Sync {
     fn get_raw_name(&self) -> &'static str;
     fn get_minimum_version(&self) -> ProtocolVersion;
 }
 
+impl<C: BasicWritableCommand + ?Sized> BasicWritableCommand for Box<C> {
+    fn get_raw_name(&self) -> &'static str {
+        (**self).get_raw_name()
+    }
+
+    fn get_minimum_version(&self) -> ProtocolVersion {
+        (**self).get_minimum_version()
+    }
+}
+
+impl<C: BasicWritableCommand + ?Sized> BasicWritableCommand for &'_ Box<C> {
+    fn get_raw_name(&self) -> &'static str {
+        (**self).get_raw_name()
+    }
+
+    fn get_minimum_version(&self) -> ProtocolVersion {
+        (**self).get_minimum_version()
+    }
+}
+
 pub trait WritableCommand: BasicWritableCommand {
     fn get_mask_flag(&self) -> HashMap<String, f64>;
     fn get_flag(&self) -> f64;

atem-connection-rs/src/commands/device_profile.rs

@@ -1,34 +1,7 @@
-use std::sync::Arc;
+pub mod audio_mixer_config;
-
+pub mod media_pool_config;
-use crate::enums::ProtocolVersion;
+pub mod mix_effect_block_config;
-
+pub mod multiviewer_config;
-use super::command_base::{CommandDeserializer, DeserializedCommand};
+pub mod product_identifier;
-
+pub mod topology;
-pub const DESERIALIZE_VERSION_RAW_NAME: &str = "_ver";
+pub mod version;
-
-#[derive(Debug)]
-pub struct VersionCommand {
-    pub version: ProtocolVersion,
-}
-
-impl DeserializedCommand for VersionCommand {
-    fn raw_name(&self) -> &'static str {
-        DESERIALIZE_VERSION_RAW_NAME
-    }
-
-    fn apply_to_state(&self, state: &mut crate::state::AtemState) {
-        state.info.api_version = self.version;
-    }
-}
-
-#[derive(Default)]
-pub struct VersionCommandDeserializer {}
-
-impl CommandDeserializer for VersionCommandDeserializer {
-    fn deserialize(&self, buffer: &[u8]) -> std::sync::Arc<dyn DeserializedCommand> {
-        let version = u32::from_be_bytes([buffer[0], buffer[1], buffer[2], buffer[3]]);
-        let version: ProtocolVersion = version.try_into().expect("Invalid protocol version");
-
-        Arc::new(VersionCommand { version })
-    }
-}

atem-connection-rs/src/commands/device_profile/audio_mixer_config.rs

@@ -0,0 +1,47 @@
+use std::sync::Arc;
+
+use crate::{
+    commands::command_base::{CommandDeserializer, DeserializedCommand},
+    state::{audio::AtemClassicAudioState, info::AudioMixerInfo},
+};
+
+pub const DESERIALIZE_AUDIO_MIXER_CONFIG_NAME: &str = "_AMC";
+
+#[derive(Debug)]
+pub struct AudioMixerConfig {
+    inputs: u8,
+    monitors: u8,
+    headphones: u8,
+}
+
+impl DeserializedCommand for AudioMixerConfig {
+    fn raw_name(&self) -> &'static str {
+        DESERIALIZE_AUDIO_MIXER_CONFIG_NAME
+    }
+
+    fn apply_to_state(&self, state: &mut crate::state::AtemState) {
+        state.info.audio_mixer = Some(AudioMixerInfo::new(
+            self.inputs,
+            self.monitors,
+            self.headphones,
+        ));
+        state.audio = Some(AtemClassicAudioState::new(self.inputs, self.monitors != 0))
+    }
+}
+
+#[derive(Default)]
+pub struct AudioMixerConfigDeserializer {}
+
+impl CommandDeserializer for AudioMixerConfigDeserializer {
+    fn deserialize(
+        &self,
+        buffer: &[u8],
+        version: &crate::enums::ProtocolVersion,
+    ) -> std::sync::Arc<dyn DeserializedCommand> {
+        Arc::new(AudioMixerConfig {
+            inputs: buffer[0],
+            monitors: buffer[1],
+            headphones: buffer[2],
+        })
+    }
+}

atem-connection-rs/src/commands/device_profile/media_pool_config.rs

@@ -0,0 +1,40 @@
+use std::sync::Arc;
+
+use crate::{
+    commands::command_base::{CommandDeserializer, DeserializedCommand},
+    state::info::MediaPoolInfo,
+};
+
+pub const DESERIALIZE_MEDIA_POOL_CONFIG_NAME: &str = "_mpl";
+
+#[derive(Debug)]
+pub struct MediaPoolConfig {
+    still_count: u8,
+    clip_count: u8,
+}
+
+impl DeserializedCommand for MediaPoolConfig {
+    fn raw_name(&self) -> &'static str {
+        DESERIALIZE_MEDIA_POOL_CONFIG_NAME
+    }
+
+    fn apply_to_state(&self, state: &mut crate::state::AtemState) {
+        state.info.media_pool = Some(MediaPoolInfo::new(self.still_count, self.clip_count))
+    }
+}
+
+#[derive(Default)]
+pub struct MediaPoolConfigDeserializer {}
+
+impl CommandDeserializer for MediaPoolConfigDeserializer {
+    fn deserialize(
+        &self,
+        buffer: &[u8],
+        _version: &crate::enums::ProtocolVersion,
+    ) -> std::sync::Arc<dyn DeserializedCommand> {
+        Arc::new(MediaPoolConfig {
+            still_count: buffer[0],
+            clip_count: buffer[1],
+        })
+    }
+}

atem-connection-rs/src/commands/device_profile/mix_effect_block_config.rs

@@ -0,0 +1,40 @@
+use std::sync::Arc;
+
+use crate::{
+    commands::command_base::{CommandDeserializer, DeserializedCommand},
+    state::info::MixEffectInfo,
+};
+
+pub const DESERIALIZE_MIX_EFFECT_BLOCK_CONFIG_NAME: &str = "_MeC";
+
+#[derive(Debug)]
+pub struct MixEffectBlockConfig {
+    mix_effect: u8,
+    key_count: u8,
+}
+
+impl DeserializedCommand for MixEffectBlockConfig {
+    fn raw_name(&self) -> &'static str {
+        DESERIALIZE_MIX_EFFECT_BLOCK_CONFIG_NAME
+    }
+
+    fn apply_to_state(&self, state: &mut crate::state::AtemState) {
+        state.info.mix_effects[self.mix_effect as usize] = Some(MixEffectInfo::new(self.key_count));
+    }
+}
+
+#[derive(Default)]
+pub struct MixEffectBlockConfigDeserializer {}
+
+impl CommandDeserializer for MixEffectBlockConfigDeserializer {
+    fn deserialize(
+        &self,
+        buffer: &[u8],
+        _version: &crate::enums::ProtocolVersion,
+    ) -> std::sync::Arc<dyn DeserializedCommand> {
+        Arc::new(MixEffectBlockConfig {
+            mix_effect: buffer[0],
+            key_count: buffer[1],
+        })
+    }
+}

atem-connection-rs/src/commands/device_profile/multiviewer_config.rs

@@ -0,0 +1,58 @@
+use std::sync::Arc;
+
+use crate::{
+    commands::command_base::{CommandDeserializer, DeserializedCommand},
+    enums::ProtocolVersion,
+    state::info::MultiviewerInfo,
+};
+
+pub const DESERIALIZE_MULTIVIEWER_NAME: &str = "_MvC";
+
+#[derive(Debug)]
+pub struct MultiviewerConfig {
+    count: Option<u8>,
+    window_count: u8,
+}
+
+impl DeserializedCommand for MultiviewerConfig {
+    fn raw_name(&self) -> &'static str {
+        DESERIALIZE_MULTIVIEWER_NAME
+    }
+
+    fn apply_to_state(&self, state: &mut crate::state::AtemState) {
+        // TODO: This can't be right...
+
+        let existing_count = match &state.info.multiviewer {
+            Some(multiviewer) => multiviewer.count().as_ref().copied(),
+            None => None,
+        };
+        let count = match self.count {
+            Some(count) => Some(count),
+            None => existing_count,
+        };
+        state.info.multiviewer = Some(MultiviewerInfo::new(count, self.window_count));
+    }
+}
+
+#[derive(Default)]
+pub struct MultiviewerConfigDeserializer {}
+
+impl CommandDeserializer for MultiviewerConfigDeserializer {
+    fn deserialize(
+        &self,
+        buffer: &[u8],
+        version: &crate::enums::ProtocolVersion,
+    ) -> std::sync::Arc<dyn DeserializedCommand> {
+        if *version >= ProtocolVersion::V8_1_1 {
+            Arc::new(MultiviewerConfig {
+                count: None,
+                window_count: buffer[1],
+            })
+        } else {
+            Arc::new(MultiviewerConfig {
+                count: Some(buffer[0]),
+                window_count: buffer[1],
+            })
+        }
+    }
+}

atem-connection-rs/src/commands/device_profile/product_identifier.rs

@@ -0,0 +1,68 @@
+use std::{ffi::CString, sync::Arc};
+
+use crate::{
+    commands::command_base::{CommandDeserializer, DeserializedCommand},
+    enums::{Model, ProtocolVersion},
+};
+
+pub const DESERIALIZE_PRODUCT_IDENTIFIER_RAW_NAME: &str = "_pin";
+
+#[derive(Debug)]
+pub struct ProductIdentifier {
+    pub product_identifier: String,
+    pub model: Model,
+}
+
+impl DeserializedCommand for ProductIdentifier {
+    fn raw_name(&self) -> &'static str {
+        DESERIALIZE_PRODUCT_IDENTIFIER_RAW_NAME
+    }
+
+    fn apply_to_state(&self, state: &mut crate::state::AtemState) {
+        state.info.product_identifier = Some(self.product_identifier.clone());
+        state.info.model = self.model.clone();
+
+        match state.info.model {
+            Model::TwoME
+            | Model::TwoME4K
+            | Model::TwoMEBS4K
+            | Model::Constellation
+            | Model::Constellation8K
+            | Model::ConstellationHD4ME
+            | Model::Constellation4K4ME => {
+                state.info.power = vec![false, false];
+            }
+            _ => {
+                state.info.power = vec![false];
+            }
+        }
+    }
+}
+
+#[derive(Default)]
+pub struct ProductIdentifierDeserializer {}
+
+impl CommandDeserializer for ProductIdentifierDeserializer {
+    fn deserialize(
+        &self,
+        buffer: &[u8],
+        version: &ProtocolVersion,
+    ) -> Arc<dyn DeserializedCommand> {
+        let null_byte_index = buffer
+            .iter()
+            .position(|byte| *byte == b'\0')
+            .expect("No null byte");
+        let product_identifier =
+            CString::from_vec_with_nul(buffer[..(null_byte_index + 1)].to_vec())
+                .expect("Malformed string");
+        let model = buffer[40];
+
+        Arc::new(ProductIdentifier {
+            product_identifier: product_identifier
+                .to_str()
+                .expect("Invalid rust string")
+                .to_string(),
+            model: model.into(),
+        })
+    }
+}

atem-connection-rs/src/commands/device_profile/topology.rs

@@ -0,0 +1,117 @@
+use std::sync::Arc;
+
+use crate::{
+    commands::command_base::{CommandDeserializer, DeserializedCommand},
+    enums::ProtocolVersion,
+    state::info::{AtemCapabilites, MultiviewerInfo},
+};
+
+pub const DESERIALIZE_TOPOLOGY_RAW_NAME: &str = "_top";
+
+#[derive(Debug)]
+pub struct Topology {
+    mix_effects: u8,
+    sources: u8,
+    auxilliaries: u8,
+    mix_minus_outputs: u8,
+    media_players: u8,
+    multiviewers: Option<u8>,
+    serial_ports: u8,
+    max_hyperdecks: u8,
+    dves: u8,
+    stingers: u8,
+    super_sources: u8,
+    talkback_channels: u8,
+    downstream_keyers: u8,
+    camera_control: bool,
+    advanced_chroma_keyers: bool,
+    only_configurable_outputs: bool,
+}
+
+impl DeserializedCommand for Topology {
+    fn raw_name(&self) -> &'static str {
+        DESERIALIZE_TOPOLOGY_RAW_NAME
+    }
+
+    fn apply_to_state(&self, state: &mut crate::state::AtemState) {
+        state.info.capabilities = Some(AtemCapabilites::new(
+            self.mix_effects,
+            self.sources,
+            self.auxilliaries,
+            self.mix_minus_outputs,
+            self.media_players,
+            self.serial_ports,
+            self.max_hyperdecks,
+            self.dves,
+            self.stingers,
+            self.super_sources,
+            self.talkback_channels,
+            self.downstream_keyers,
+            self.camera_control,
+            self.advanced_chroma_keyers,
+            self.only_configurable_outputs,
+        ));
+
+        let window_count = if let Some(mv) = &state.info.multiviewer {
+            *mv.window_count()
+        } else {
+            10
+        };
+
+        state.info.multiviewer = Some(MultiviewerInfo::new(self.multiviewers, window_count));
+    }
+}
+
+#[derive(Default)]
+pub struct TopologyDeserializer {}
+
+impl CommandDeserializer for TopologyDeserializer {
+    fn deserialize(
+        &self,
+        buffer: &[u8],
+        version: &ProtocolVersion,
+    ) -> Arc<dyn DeserializedCommand> {
+        let v230offset = if *version > ProtocolVersion::V8_0_1 {
+            1
+        } else {
+            0
+        };
+
+        let multiviewers = if v230offset > 0 {
+            Some(buffer[6])
+        } else {
+            None
+        };
+
+        let advanced_chroma_keyers = if buffer.len() > 20 {
+            buffer[21 + v230offset] == 1
+        } else {
+            false
+        };
+
+        let only_configurable_outputs = if buffer.len() > 20 {
+            buffer[22 + v230offset] == 1
+        } else {
+            false
+        };
+
+        Arc::new(Topology {
+            mix_effects: buffer[0],
+            sources: buffer[1],
+            downstream_keyers: buffer[2],
+            auxilliaries: buffer[3],
+            mix_minus_outputs: buffer[4],
+            media_players: buffer[5],
+            multiviewers,
+            serial_ports: buffer[6 + v230offset],
+            max_hyperdecks: buffer[7 + v230offset],
+            dves: buffer[8 + v230offset],
+            stingers: buffer[9 + v230offset],
+            super_sources: buffer[10 + v230offset],
+            talkback_channels: buffer[12 + v230offset],
+            camera_control: buffer[17 + v230offset] == 1,
+            advanced_chroma_keyers,
+            only_configurable_outputs,
+        })
+    }
+}

atem-connection-rs/src/commands/device_profile/version.rs

@@ -0,0 +1,25 @@
+use crate::{commands::command_base::DeserializedCommand, enums::ProtocolVersion};
+
+pub const DESERIALIZE_VERSION_RAW_NAME: &str = "_ver";
+
+#[derive(Debug)]
+pub struct Version {
+    pub version: ProtocolVersion,
+}
+
+impl DeserializedCommand for Version {
+    fn raw_name(&self) -> &'static str {
+        DESERIALIZE_VERSION_RAW_NAME
+    }
+
+    fn apply_to_state(&self, state: &mut crate::state::AtemState) {
+        state.info.api_version = self.version.clone();
+    }
+}
+
+pub fn deserialize_version(buffer: &[u8]) -> Version {
+    let version = u32::from_be_bytes([buffer[0], buffer[1], buffer[2], buffer[3]]);
+    let version: ProtocolVersion = version.try_into().expect("Invalid protocol version");
+
+    Version { version }
+}

atem-connection-rs/src/commands/init_complete.rs

@@ -1,5 +1,7 @@
 use std::sync::Arc;
 
+use crate::enums::ProtocolVersion;
+
 use super::command_base::{CommandDeserializer, DeserializedCommand};
 
 pub const DESERIALIZE_INIT_COMPLETE_RAW_NAME: &str = "InCm";
@@ -19,7 +21,11 @@ impl DeserializedCommand for InitComplete {
 pub struct InitCompleteDeserializer {}
 
 impl CommandDeserializer for InitCompleteDeserializer {
-    fn deserialize(&self, _buffer: &[u8]) -> std::sync::Arc<dyn DeserializedCommand> {
+    fn deserialize(
+        &self,
+        _buffer: &[u8],
+        version: &ProtocolVersion,
+    ) -> Arc<dyn DeserializedCommand> {
         Arc::new(InitComplete {})
     }
 }

atem-connection-rs/src/commands/mix_effects/program_input.rs

@@ -4,6 +4,7 @@ use crate::{
     commands::command_base::{
         BasicWritableCommand, CommandDeserializer, DeserializedCommand, SerializableCommand,
     },
+    enums::ProtocolVersion,
     state::util::get_mix_effect,
 };
 
@@ -58,7 +59,11 @@ impl DeserializedCommand for ProgramInput {
 pub struct ProgramInputDeserializer {}
 
 impl CommandDeserializer for ProgramInputDeserializer {
-    fn deserialize(&self, buffer: &[u8]) -> Arc<dyn DeserializedCommand> {
+    fn deserialize(
+        &self,
+        buffer: &[u8],
+        version: &ProtocolVersion,
+    ) -> Arc<dyn DeserializedCommand> {
         let mix_effect = buffer[0];
         let source = u16::from_be_bytes([buffer[2], buffer[3]]);
 

atem-connection-rs/src/commands/parse_commands.rs

@@ -1,16 +1,35 @@
 use std::{collections::VecDeque, sync::Arc};
 
+use crate::{
+    commands::device_profile::version::{deserialize_version, DESERIALIZE_VERSION_RAW_NAME},
+    enums::ProtocolVersion,
+};
+
 use super::{
     command_base::{CommandDeserializer, DeserializedCommand},
-    device_profile::{VersionCommandDeserializer, DESERIALIZE_VERSION_RAW_NAME},
+    device_profile::{
+        audio_mixer_config::{AudioMixerConfigDeserializer, DESERIALIZE_AUDIO_MIXER_CONFIG_NAME},
+        media_pool_config::{MediaPoolConfigDeserializer, DESERIALIZE_MEDIA_POOL_CONFIG_NAME},
+        mix_effect_block_config::{
+            MixEffectBlockConfigDeserializer, DESERIALIZE_MIX_EFFECT_BLOCK_CONFIG_NAME,
+        },
+        multiviewer_config::{MultiviewerConfigDeserializer, DESERIALIZE_MULTIVIEWER_NAME},
+        product_identifier::{
+            ProductIdentifierDeserializer, DESERIALIZE_PRODUCT_IDENTIFIER_RAW_NAME,
+        },
+        topology::{TopologyDeserializer, DESERIALIZE_TOPOLOGY_RAW_NAME},
+    },
     init_complete::{InitCompleteDeserializer, DESERIALIZE_INIT_COMPLETE_RAW_NAME},
     mix_effects::program_input::{ProgramInputDeserializer, DESERIALIZE_PROGRAM_INPUT_RAW_NAME},
     tally_by_source::{TallyBySourceDeserializer, DESERIALIZE_TALLY_BY_SOURCE_RAW_NAME},
     time::{TimeDeserializer, DESERIALIZE_TIME_RAW_NAME},
 };
 
-pub fn deserialize_commands(payload: &[u8]) -> VecDeque<Arc<dyn DeserializedCommand>> {
+pub fn deserialize_commands(
-    let mut parsed_commands = VecDeque::new();
+    payload: &[u8],
+    version: &mut ProtocolVersion,
+) -> VecDeque<Arc<dyn DeserializedCommand>> {
+    let mut parsed_commands: VecDeque<Arc<dyn DeserializedCommand>> = VecDeque::new();
     let mut head = 0;
 
     while payload.len() > head + 8 {
@@ -25,10 +44,21 @@ pub fn deserialize_commands(payload: &[u8]) -> VecDeque<Arc<dyn DeserializedComm
 
         log::debug!("Received command {} with length {}", name, length);
 
-        if let Some(deserializer) = command_deserializer_from_string(name.as_str()) {
+        let command_buffer = &payload[head + 8..head + length];
-            let deserialized_command = deserializer.deserialize(&payload[head + 8..head + length]);
+
+        if name == DESERIALIZE_VERSION_RAW_NAME {
+            let version_command = deserialize_version(command_buffer);
+            *version = version_command.version.clone();
+            log::info!("Switched to protocol version {}", version);
+            parsed_commands.push_back(Arc::new(version_command));
+        } else if let Some(deserializer) = command_deserializer_from_string(name.as_str()) {
+            let deserialized_command = deserializer.deserialize(command_buffer, version);
             log::debug!("Received {:?}", deserialized_command);
             parsed_commands.push_back(deserialized_command);
+        } else {
+            log::warn!("Received command {name} for which there is no deserializer.");
+            // TODO: Remove!
+            todo!("Write deserializer for {name}.");
         }
 
         head += length;
@@ -39,11 +69,20 @@ pub fn deserialize_commands(payload: &[u8]) -> VecDeque<Arc<dyn DeserializedComm
 
 fn command_deserializer_from_string(command_str: &str) -> Option<Box<dyn CommandDeserializer>> {
     match command_str {
-        DESERIALIZE_VERSION_RAW_NAME => Some(Box::<VersionCommandDeserializer>::default()),
         DESERIALIZE_INIT_COMPLETE_RAW_NAME => Some(Box::<InitCompleteDeserializer>::default()),
         DESERIALIZE_PROGRAM_INPUT_RAW_NAME => Some(Box::<ProgramInputDeserializer>::default()),
         DESERIALIZE_TALLY_BY_SOURCE_RAW_NAME => Some(Box::<TallyBySourceDeserializer>::default()),
         DESERIALIZE_TIME_RAW_NAME => Some(Box::<TimeDeserializer>::default()),
+        DESERIALIZE_TOPOLOGY_RAW_NAME => Some(Box::<TopologyDeserializer>::default()),
+        DESERIALIZE_MIX_EFFECT_BLOCK_CONFIG_NAME => {
+            Some(Box::<MixEffectBlockConfigDeserializer>::default())
+        }
+        DESERIALIZE_PRODUCT_IDENTIFIER_RAW_NAME => {
+            Some(Box::<ProductIdentifierDeserializer>::default())
+        }
+        DESERIALIZE_MEDIA_POOL_CONFIG_NAME => Some(Box::<MediaPoolConfigDeserializer>::default()),
+        DESERIALIZE_MULTIVIEWER_NAME => Some(Box::<MultiviewerConfigDeserializer>::default()),
+        DESERIALIZE_AUDIO_MIXER_CONFIG_NAME => Some(Box::<AudioMixerConfigDeserializer>::default()),
         _ => None,
     }
 }

atem-connection-rs/src/commands/tally_by_source.rs

@@ -1,5 +1,7 @@
 use std::{collections::HashMap, sync::Arc};
 
+use crate::enums::ProtocolVersion;
+
 use super::command_base::{CommandDeserializer, DeserializedCommand};
 
 pub const DESERIALIZE_TALLY_BY_SOURCE_RAW_NAME: &str = "TlSr";
@@ -29,7 +31,11 @@ impl DeserializedCommand for TallyBySource {
 pub struct TallyBySourceDeserializer {}
 
 impl CommandDeserializer for TallyBySourceDeserializer {
-    fn deserialize(&self, buffer: &[u8]) -> Arc<dyn DeserializedCommand> {
+    fn deserialize(
+        &self,
+        buffer: &[u8],
+        version: &ProtocolVersion,
+    ) -> Arc<dyn DeserializedCommand> {
         let source_count = u16::from_be_bytes([buffer[0], buffer[1]]) as usize;
 
         log::debug!("{:?}", buffer);

atem-connection-rs/src/commands/time.rs

@@ -1,5 +1,7 @@
 use std::sync::Arc;
 
+use crate::enums::ProtocolVersion;
+
 use super::command_base::{CommandDeserializer, DeserializedCommand};
 
 pub const DESERIALIZE_TIME_RAW_NAME: &str = "Time";
@@ -33,7 +35,8 @@ impl CommandDeserializer for TimeDeserializer {
     fn deserialize(
         &self,
         buffer: &[u8],
-    ) -> std::sync::Arc<dyn super::command_base::DeserializedCommand> {
+        version: &ProtocolVersion,
+    ) -> Arc<dyn DeserializedCommand> {
         let info = TimeInfo {
             hour: buffer[0],
             minute: buffer[1],

atem-connection-rs/src/enums/mod.rs

@@ -1,4 +1,6 @@
-#[derive(Clone, Default, PartialEq)]
+use std::fmt::Display;
+
+#[derive(Debug, Clone, Default, PartialEq)]
 pub enum Model {
     #[default]
     Unknown = 0x00,
@@ -19,9 +21,59 @@ pub enum Model {
     MiniProISO = 0x0f,
     MiniExtreme = 0x10,
     MiniExtremeISO = 0x11,
+    ConstellationHD1ME = 0x12,
+    ConstellationHD2ME = 0x13,
+    ConstellationHD4ME = 0x14,
+    SDI = 0x15,
+    SDIProISO = 0x16,
+    SDIExtremeISO = 0x17,
+    // 0x18 ??
+    // 0x19 ??
+    TelevisionStudioHD8 = 0x1a,
+    TelevisionStudioHD8ISO = 0x1b,
+    // 0x1c ??
+    // 0x1d ??
+    Constellation4K4ME = 0x1e,
+    // 0x1f ??
+    TelevisionStudio4K8 = 0x20,
 }
 
-#[derive(Debug, Default, Clone, Copy, PartialEq)]
+impl From<u8> for Model {
+    fn from(value: u8) -> Self {
+        match value {
+            0x01 => Model::TVS,
+            0x02 => Model::OneME,
+            0x03 => Model::TwoME,
+            0x04 => Model::PS4K,
+            0x05 => Model::OneME4K,
+            0x06 => Model::TwoME4K,
+            0x07 => Model::TwoMEBS4K,
+            0x08 => Model::TVSHD,
+            0x09 => Model::TVSProHD,
+            0x0a => Model::TVSPro4K,
+            0x0b => Model::Constellation,
+            0x0c => Model::Constellation8K,
+            0x0d => Model::Mini,
+            0x0e => Model::MiniPro,
+            0x0f => Model::MiniProISO,
+            0x10 => Model::MiniExtreme,
+            0x11 => Model::MiniExtremeISO,
+            0x12 => Model::ConstellationHD1ME,
+            0x13 => Model::ConstellationHD2ME,
+            0x14 => Model::ConstellationHD4ME,
+            0x15 => Model::SDI,
+            0x16 => Model::SDIProISO,
+            0x17 => Model::SDIExtremeISO,
+            0x1a => Model::TelevisionStudioHD8,
+            0x1b => Model::TelevisionStudioHD8ISO,
+            0x1e => Model::Constellation4K4ME,
+            0x20 => Model::TelevisionStudio4K8,
+            _ => Model::Unknown,
+        }
+    }
+}
+
+#[derive(Debug, Default, Clone, PartialEq, PartialOrd)]
 pub enum ProtocolVersion {
     #[default]
     Unknown = 0,
@@ -48,6 +100,19 @@ impl TryFrom<u32> for ProtocolVersion {
     }
 }
 
+impl Display for ProtocolVersion {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            ProtocolVersion::Unknown => write!(f, "Unknown"),
+            ProtocolVersion::V7_2 => write!(f, "v7.2"),
+            ProtocolVersion::V7_5_2 => write!(f, "v7.5.2"),
+            ProtocolVersion::V8_0 => write!(f, "v8.0"),
+            ProtocolVersion::V8_0_1 => write!(f, "v8.0.1"),
+            ProtocolVersion::V8_1_1 => write!(f, "v8.1.1"),
+        }
+    }
+}
+
 #[derive(Clone, PartialEq)]
 pub enum TransitionStyle {
     MIX = 0x00,

atem-connection-rs/src/state/audio.rs

@@ -44,14 +44,28 @@ pub struct ClassicAudioHeadphoneOutputChannel {
     pub talkback_gain: f64,
 }
 
-#[derive(Clone, PartialEq, Getters, new)]
+#[derive(Clone, PartialEq, Getters)]
 pub struct AtemClassicAudioState {
-    number_of_channels: Option<f64>,
+    number_of_channels: u8,
-    has_monitor: Option<bool>,
+    has_monitor: bool,
     pub channels: HashMap<u64, ClassicAudioChannel>,
     pub monitor: Option<ClassicAudioMonitorChannel>,
     pub headphones: Option<ClassicAudioHeadphoneOutputChannel>,
     pub master: Option<ClassicAudioMasterChannel>,
 
-    pub audio_follow_video_crossfade_transition_enabled: Option<bool>,
+    pub audio_follow_video_crossfade_transition_enabled: bool,
+}
+
+impl AtemClassicAudioState {
+    pub fn new(number_of_channels: u8, has_monitor: bool) -> Self {
+        Self {
+            number_of_channels,
+            has_monitor,
+            channels: Default::default(),
+            monitor: Default::default(),
+            headphones: Default::default(),
+            master: Default::default(),
+            audio_follow_video_crossfade_transition_enabled: false,
+        }
+    }
 }

atem-connection-rs/src/state/info.rs

@@ -3,17 +3,17 @@ use crate::enums::{Model, ProtocolVersion};
 #[derive(Clone, PartialEq, Getters, new)]
 pub struct AtemCapabilites {
     mix_effects: u8,
-    sources: u64,
+    sources: u8,
-    auxilliaries: u64,
+    auxilliaries: u8,
-    mix_minus_outputs: u64,
+    mix_minus_outputs: u8,
-    media_players: u64,
+    media_players: u8,
-    serial_ports: u64,
+    serial_ports: u8,
-    max_hyperdecks: u64,
+    max_hyperdecks: u8,
-    dves: u64,
+    dves: u8,
-    stingers: u64,
+    stingers: u8,
-    super_sources: u64,
+    super_sources: u8,
-    talkback_channels: u64,
+    talkback_channels: u8,
-    downstream_keyers: u64,
+    downstream_keyers: u8,
     camera_control: bool,
     advanced_chroma_keyers: bool,
     only_configurable_outputs: bool,
@@ -21,7 +21,7 @@ pub struct AtemCapabilites {
 
 #[derive(Clone, PartialEq, Getters, new)]
 pub struct MixEffectInfo {
-    key_count: u64,
+    key_count: u8,
 }
 
 #[derive(Clone, PartialEq, Getters, new)]
@@ -31,9 +31,9 @@ pub struct SuperSourceInfo {
 
 #[derive(Clone, PartialEq, Getters, new)]
 pub struct AudioMixerInfo {
-    inputs: u64,
+    inputs: u8,
-    monitors: u64,
+    monitors: u8,
-    headphones: u64,
+    headphones: u8,
 }
 
 #[derive(Clone, PartialEq, Getters, new)]
@@ -49,14 +49,14 @@ pub struct MacroPoolInfo {
 
 #[derive(Clone, PartialEq, Getters, new)]
 pub struct MediaPoolInfo {
-    still_count: u64,
+    still_count: u8,
-    clip_count: u64,
+    clip_count: u8,
 }
 
 #[derive(Clone, PartialEq, Getters, new)]
 pub struct MultiviewerInfo {
-    count: u64,
+    count: Option<u8>,
-    window_count: u64,
+    window_count: u8,
 }
 
 #[derive(Clone, PartialEq, new)]

atem-test/src/main.rs

@@ -13,7 +13,7 @@ use atem_connection_rs::{
 
 use clap::Parser;
 use color_eyre::Report;
-use tokio::time::sleep;
+use tokio::{select, time::sleep};
 use tokio_util::sync::CancellationToken;
 
 /// ATEM Rust Library Test App
@@ -35,32 +35,35 @@ async fn main() {
         tokio::sync::mpsc::channel::<AtemSocketMessage>(10);
     let (atem_event_tx, atem_event_rx) = tokio::sync::mpsc::unbounded_channel();
     let cancel = CancellationToken::new();
-    let cancel_task = cancel.clone();
 
-    let mut atem_socket = AtemSocket::new(atem_event_tx);
+    let mut atem_socket = AtemSocket::new(socket_message_rx, atem_event_tx);
-    tokio::spawn(async move {
-        atem_socket.run(socket_message_rx, cancel_task).await;
-    });
 
-    let atem = Arc::new(Atem::new(socket_message_tx));
+    let atem = Arc::new(Atem::new(atem_socket, socket_message_tx));
     let atem_thread = atem.clone();
-    tokio::spawn(async move {
+    let atem_run = atem_thread.run(atem_event_rx, cancel);
-        atem_thread.run(atem_event_rx, cancel).await;
+
+    let switch_loop = tokio::spawn(async move {
+        let address = Ipv4Addr::from_str(&args.ip).unwrap();
+        let socket = SocketAddrV4::new(address, 9910);
+        atem.connect(socket.into()).await;
+
+        loop {
+            sleep(Duration::from_millis(5000)).await;
+            log::info!("Switch to source 1");
+            atem.send_commands(vec![Box::new(ProgramInput::new(0, 1))])
+                .await;
+            log::info!("Switched to source 1");
+            sleep(Duration::from_millis(5000)).await;
+            log::info!("Switch to source 2");
+            atem.send_commands(vec![Box::new(ProgramInput::new(0, 2))])
+                .await;
+            log::info!("Switched to source 2");
+        }
     });
 
-    let address = Ipv4Addr::from_str(&args.ip).unwrap();
+    select! {
-    let socket = SocketAddrV4::new(address, 9910);
+        _ = atem_run => {},
-    atem.connect(socket.into()).await;
+        _ = switch_loop => {}
-
-    loop {
-        sleep(Duration::from_millis(5000)).await;
-        log::info!("Switch to source 1");
-        atem.send_commands(vec![Box::new(ProgramInput::new(0, 1))])
-            .await;
-        sleep(Duration::from_millis(5000)).await;
-        log::info!("Switch to source 2");
-        atem.send_commands(vec![Box::new(ProgramInput::new(0, 2))])
-            .await;
     }
 }
 

flake.lock

@@ -2,15 +2,14 @@
   "nodes": {
     "devshell": {
       "inputs": {
-        "flake-utils": "flake-utils",
         "nixpkgs": "nixpkgs"
       },
       "locked": {
-        "lastModified": 1705332421,
+        "lastModified": 1741473158,
-        "narHash": "sha256-USpGLPme1IuqG78JNqSaRabilwkCyHmVWY0M9vYyqEA=",
+        "narHash": "sha256-kWNaq6wQUbUMlPgw8Y+9/9wP0F8SHkjy24/mN3UAppg=",
         "owner": "numtide",
         "repo": "devshell",
-        "rev": "83cb93d6d063ad290beee669f4badf9914cc16ec",
+        "rev": "7c9e793ebe66bcba8292989a68c0419b737a22a0",
         "type": "github"
       },
       "original": {
@@ -19,52 +18,16 @@
         "type": "github"
       }
     },
-    "flake-utils": {
-      "inputs": {
-        "systems": "systems"
-      },
-      "locked": {
-        "lastModified": 1701680307,
-        "narHash": "sha256-kAuep2h5ajznlPMD9rnQyffWG8EM/C73lejGofXvdM8=",
-        "owner": "numtide",
-        "repo": "flake-utils",
-        "rev": "4022d587cbbfd70fe950c1e2083a02621806a725",
-        "type": "github"
-      },
-      "original": {
-        "owner": "numtide",
-        "repo": "flake-utils",
-        "type": "github"
-      }
-    },
-    "flake-utils_2": {
-      "inputs": {
-        "systems": "systems_2"
-      },
-      "locked": {
-        "lastModified": 1681202837,
-        "narHash": "sha256-H+Rh19JDwRtpVPAWp64F+rlEtxUWBAQW28eAi3SRSzg=",
-        "owner": "numtide",
-        "repo": "flake-utils",
-        "rev": "cfacdce06f30d2b68473a46042957675eebb3401",
-        "type": "github"
-      },
-      "original": {
-        "owner": "numtide",
-        "repo": "flake-utils",
-        "type": "github"
-      }
-    },
     "naersk": {
       "inputs": {
         "nixpkgs": "nixpkgs_2"
       },
       "locked": {
-        "lastModified": 1698420672,
+        "lastModified": 1745925850,
-        "narHash": "sha256-/TdeHMPRjjdJub7p7+w55vyABrsJlt5QkznPYy55vKA=",
+        "narHash": "sha256-cyAAMal0aPrlb1NgzMxZqeN1mAJ2pJseDhm2m6Um8T0=",
         "owner": "nix-community",
         "repo": "naersk",
-        "rev": "aeb58d5e8faead8980a807c840232697982d47b9",
+        "rev": "38bc60bbc157ae266d4a0c96671c6c742ee17a5f",
         "type": "github"
       },
       "original": {
@@ -75,11 +38,11 @@
     },
     "nixpkgs": {
       "locked": {
-        "lastModified": 1704161960,
+        "lastModified": 1722073938,
-        "narHash": "sha256-QGua89Pmq+FBAro8NriTuoO/wNaUtugt29/qqA8zeeM=",
+        "narHash": "sha256-OpX0StkL8vpXyWOGUD6G+MA26wAXK6SpT94kLJXo6B4=",
         "owner": "NixOS",
         "repo": "nixpkgs",
-        "rev": "63143ac2c9186be6d9da6035fa22620018c85932",
+        "rev": "e36e9f57337d0ff0cf77aceb58af4c805472bfae",
         "type": "github"
       },
       "original": {
@@ -91,26 +54,26 @@
     },
     "nixpkgs_2": {
       "locked": {
-        "lastModified": 1705883077,
+        "lastModified": 1749401433,
-        "narHash": "sha256-ByzHHX3KxpU1+V0erFy8jpujTufimh6KaS/Iv3AciHk=",
+        "narHash": "sha256-HXIQzULIG/MEUW2Q/Ss47oE3QrjxvpUX7gUl4Xp6lnc=",
         "owner": "NixOS",
         "repo": "nixpkgs",
-        "rev": "5f5210aa20e343b7e35f40c033000db0ef80d7b9",
+        "rev": "08fcb0dcb59df0344652b38ea6326a2d8271baff",
         "type": "github"
       },
       "original": {
-        "id": "nixpkgs",
+        "owner": "NixOS",
-        "type": "indirect"
+        "ref": "nixpkgs-unstable",
+        "repo": "nixpkgs",
+        "type": "github"
       }
     },
     "nixpkgs_3": {
       "locked": {
-        "lastModified": 1705883077,
+        "lastModified": 0,
-        "narHash": "sha256-ByzHHX3KxpU1+V0erFy8jpujTufimh6KaS/Iv3AciHk=",
+        "narHash": "sha256-DDe16FJk18sadknQKKG/9FbwEro7A57tg9vB5kxZ8kY=",
-        "owner": "NixOS",
+        "path": "/nix/store/2d1ahim48jhzg4bbm97mvjlb4p7fpan3-source",
-        "repo": "nixpkgs",
+        "type": "path"
-        "rev": "5f5210aa20e343b7e35f40c033000db0ef80d7b9",
-        "type": "github"
       },
       "original": {
         "id": "nixpkgs",
@@ -119,11 +82,11 @@
     },
     "nixpkgs_4": {
       "locked": {
-        "lastModified": 1681358109,
+        "lastModified": 1744536153,
-        "narHash": "sha256-eKyxW4OohHQx9Urxi7TQlFBTDWII+F+x2hklDOQPB50=",
+        "narHash": "sha256-awS2zRgF4uTwrOKwwiJcByDzDOdo3Q1rPZbiHQg/N38=",
         "owner": "NixOS",
         "repo": "nixpkgs",
-        "rev": "96ba1c52e54e74c3197f4d43026b3f3d92e83ff9",
+        "rev": "18dd725c29603f582cf1900e0d25f9f1063dbf11",
         "type": "github"
       },
       "original": {
@@ -144,15 +107,14 @@
     },
     "rust-overlay": {
       "inputs": {
-        "flake-utils": "flake-utils_2",
         "nixpkgs": "nixpkgs_4"
       },
       "locked": {
-        "lastModified": 1705976279,
+        "lastModified": 1749436897,
-        "narHash": "sha256-Zx97bJ3+O8IP70uJPD//rRsr8bcxICISMTZUT/L9eFk=",
+        "narHash": "sha256-OkDtaCGQQVwVFz5HWfbmrMJR99sFIMXHCHEYXzUJEJY=",
         "owner": "oxalica",
         "repo": "rust-overlay",
-        "rev": "f889dc31ef97835834bdc3662394ebdb3c96b974",
+        "rev": "e7876c387e35dc834838aff254d8e74cf5bd4f19",
         "type": "github"
       },
       "original": {
@@ -176,46 +138,16 @@
         "type": "github"
       }
     },
-    "systems_2": {
-      "locked": {
-        "lastModified": 1681028828,
-        "narHash": "sha256-Vy1rq5AaRuLzOxct8nz4T6wlgyUR7zLU309k9mBC768=",
-        "owner": "nix-systems",
-        "repo": "default",
-        "rev": "da67096a3b9bf56a91d16901293e51ba5b49a27e",
-        "type": "github"
-      },
-      "original": {
-        "owner": "nix-systems",
-        "repo": "default",
-        "type": "github"
-      }
-    },
-    "systems_3": {
-      "locked": {
-        "lastModified": 1681028828,
-        "narHash": "sha256-Vy1rq5AaRuLzOxct8nz4T6wlgyUR7zLU309k9mBC768=",
-        "owner": "nix-systems",
-        "repo": "default",
-        "rev": "da67096a3b9bf56a91d16901293e51ba5b49a27e",
-        "type": "github"
-      },
-      "original": {
-        "owner": "nix-systems",
-        "repo": "default",
-        "type": "github"
-      }
-    },
     "utils": {
       "inputs": {
-        "systems": "systems_3"
+        "systems": "systems"
       },
       "locked": {
-        "lastModified": 1705309234,
+        "lastModified": 1731533236,
-        "narHash": "sha256-uNRRNRKmJyCRC/8y1RqBkqWBLM034y4qN7EprSdmgyA=",
+        "narHash": "sha256-l0KFg5HjrsfsO/JpG+r7fRrqm12kzFHyUHqHCVpMMbI=",
         "owner": "numtide",
         "repo": "flake-utils",
-        "rev": "1ef2e671c3b0c19053962c07dbda38332dcebf26",
+        "rev": "11707dc2f618dd54ca8739b309ec4fc024de578b",
         "type": "github"
       },
       "original": {