First version of asynchronix based mini simulator

Cargo.toml

@@ -4,6 +4,7 @@ members = [
     "satrs",
     "satrs-mib",
     "satrs-example",
+    "satrs-minisim",
     "satrs-shared",
 ]
 

satrs-minisim/Cargo.toml

@@ -0,0 +1,23 @@
+[package]
+name = "satrs-minisim"
+version = "0.1.0"
+edition = "2021"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+serde = { version = "1", features = ["derive"] }
+serde_json = "1"
+log = "0.4"
+
+[dependencies.asynchronix]
+version = "0.2.1"
+# path = "../../asynchronix/asynchronix"
+# git = "https://github.com/us-irs/asynchronix.git"
+# branch = "clock-not-sendable"
+
+[dependencies.satrs]
+path = "../satrs"
+
+[dev-dependencies]
+delegate = "0.12"

satrs-minisim/src/acs.rs

@@ -0,0 +1,150 @@
+use std::{f32::consts::PI, sync::mpsc, time::Duration};
+
+use asynchronix::{
+    model::{Model, Output},
+    time::Scheduler,
+};
+use satrs::power::{SwitchState, SwitchStateBinary};
+use satrs_minisim::{
+    acs::{MgmSensorValues, MgtDipole, MGT_GEN_MAGNETIC_FIELD},
+    SimDevice, SimReply,
+};
+
+use crate::time::current_millis;
+
+// Earth magnetic field varies between -30 uT and 30 uT
+const AMPLITUDE_MGM: f32 = 0.03;
+// Lets start with a simple frequency here.
+const FREQUENCY_MGM: f32 = 1.0;
+const PHASE_X: f32 = 0.0;
+// Different phases to have different values on the other axes.
+const PHASE_Y: f32 = 0.1;
+const PHASE_Z: f32 = 0.2;
+
+/// Simple model for a magnetometer where the measure magnetic fields are modeled with sine waves.
+///
+/// Please note that that a more realistic MGM model wouold include the following components
+/// which are not included here to simplify the model:
+///
+/// 1. It would probably generate signed [i16] values which need to be converted to SI units
+///    because it is a digital sensor
+/// 2. It would sample the magnetic field at a high fixed rate. This might not be possible for
+///    a general purpose OS, but self self-sampling at a relatively high rate (20-40 ms) might
+///    stil lbe possible.
+pub struct MagnetometerModel {
+    pub switch_state: SwitchStateBinary,
+    pub periodicity: Duration,
+    pub external_mag_field: Option<MgmSensorValues>,
+    pub reply_sender: mpsc::Sender<SimReply>,
+}
+
+impl MagnetometerModel {
+    pub fn new(periodicity: Duration, reply_sender: mpsc::Sender<SimReply>) -> Self {
+        Self {
+            switch_state: SwitchStateBinary::Off,
+            periodicity,
+            external_mag_field: None,
+            reply_sender,
+        }
+    }
+
+    pub async fn switch_device(&mut self, switch_state: SwitchStateBinary) {
+        self.switch_state = switch_state;
+    }
+
+    pub async fn send_sensor_values(&mut self, _: (), scheduler: &Scheduler<Self>) {
+        let current_time = scheduler.time();
+        println!("current monotonic time: {:?}", current_time);
+        let value = self.calculate_current_mgm_tuple(current_millis(scheduler.time()));
+        let reply = SimReply {
+            device: SimDevice::Mgm,
+            reply: serde_json::to_string(&value).unwrap(),
+        };
+        self.reply_sender
+            .send(reply)
+            .expect("sending MGM sensor values failed");
+    }
+
+    // Devices like magnetorquers generate a strong magnetic field which overrides the default
+    // model for the measured magnetic field.
+    pub async fn apply_external_magnetic_field(&mut self, field: MgmSensorValues) {
+        self.external_mag_field = Some(field);
+    }
+
+    fn calculate_current_mgm_tuple(&mut self, time_ms: u64) -> MgmSensorValues {
+        if SwitchStateBinary::On == self.switch_state {
+            if let Some(ext_field) = self.external_mag_field {
+                return ext_field;
+            }
+            let base_sin_val = 2.0 * PI as f32 * FREQUENCY_MGM * (time_ms as f32 / 1000.0);
+            return MgmSensorValues {
+                x: AMPLITUDE_MGM * (base_sin_val + PHASE_X).sin(),
+                y: AMPLITUDE_MGM * (base_sin_val + PHASE_Y).sin(),
+                z: AMPLITUDE_MGM * (base_sin_val + PHASE_Z).sin(),
+            };
+        }
+        MgmSensorValues {
+            x: 0.0,
+            y: 0.0,
+            z: 0.0,
+        }
+    }
+}
+
+impl Model for MagnetometerModel {}
+
+pub struct MagnetorquerModel {
+    switch_state: SwitchState,
+    torquing: bool,
+    torque_dipole: Option<MgtDipole>,
+    gen_magnetic_field: Output<MgmSensorValues>,
+}
+
+impl MagnetorquerModel {
+    pub async fn apply_torque(
+        &mut self,
+        dipole: MgtDipole,
+        torque_duration: Duration,
+        scheduler: &Scheduler<Self>,
+    ) {
+        self.torque_dipole = Some(dipole);
+        self.torquing = true;
+        if scheduler
+            .schedule_event(torque_duration, Self::clear_torque, ())
+            .is_err()
+        {
+            log::warn!("torque clearing can only be set for a future time.");
+        }
+        self.generate_magnetic_field(()).await;
+    }
+
+    pub async fn clear_torque(&mut self, _: ()) {
+        self.torque_dipole = None;
+        self.torquing = false;
+        self.generate_magnetic_field(()).await;
+    }
+
+    pub async fn switch_device(&mut self, switch_state: SwitchState) {
+        self.switch_state = switch_state;
+        self.generate_magnetic_field(()).await;
+    }
+
+    fn calc_magnetic_field(&self, _: MgtDipole) -> MgmSensorValues {
+        // Simplified model: Just returns some fixed magnetic field for now.
+        // Later, we could make this more fancy by incorporating the commanded dipole.
+        MGT_GEN_MAGNETIC_FIELD
+    }
+
+    /// A torquing magnetorquer generates a magnetic field. This function can be used to apply
+    /// the magnetic field.
+    async fn generate_magnetic_field(&mut self, _: ()) {
+        if self.switch_state != SwitchState::On || !self.torquing {
+            return;
+        }
+        self.gen_magnetic_field
+            .send(self.calc_magnetic_field(self.torque_dipole.expect("expected valid dipole")))
+            .await;
+    }
+}
+
+impl Model for MagnetorquerModel {}

satrs-minisim/src/controller.rs

@@ -0,0 +1,110 @@
+use std::{
+    sync::mpsc,
+    time::{Duration, SystemTime},
+};
+
+use asynchronix::{
+    simulation::{Address, Mailbox, SimInit, Simulation},
+    time::{Clock, MonotonicTime, SystemClock},
+};
+use satrs_minisim::{acs::MgmRequest, SimRequest};
+
+use crate::{
+    acs::MagnetometerModel,
+    eps::{PcduModel, PcduRequest},
+};
+
+// The simulation controller processes requests and drives the simulation.
+pub struct SimController {
+    pub sys_clock: SystemClock,
+    pub request_receiver: mpsc::Receiver<SimRequest>,
+    pub simulation: Simulation,
+    pub mgm_addr: Address<MagnetometerModel>,
+    pub pcdu_addr: Address<PcduModel>,
+}
+
+impl SimController {
+    pub fn new(
+        sys_clock: SystemClock,
+        request_receiver: mpsc::Receiver<SimRequest>,
+        simulation: Simulation,
+        mgm_addr: Address<MagnetometerModel>,
+        pcdu_addr: Address<PcduModel>,
+    ) -> Self {
+        Self {
+            sys_clock,
+            request_receiver,
+            simulation,
+            mgm_addr,
+            pcdu_addr,
+        }
+    }
+
+    pub fn run(&mut self, start_time: MonotonicTime) {
+        let mut t = start_time + Duration::from_millis(1);
+        self.sys_clock.synchronize(t);
+        loop {
+            self.simulation
+                .step_until(t)
+                .expect("simulation step failed");
+            // Check for UDP requests every millisecond.
+            t += Duration::from_millis(1);
+            self.handle_sim_requests();
+
+            self.sys_clock.synchronize(t);
+        }
+    }
+
+    pub fn handle_sim_requests(&mut self) {
+        loop {
+            match self.request_receiver.try_recv() {
+                Ok(request) => match request.device() {
+                    satrs_minisim::SimDevice::Mgm => self.handle_mgm_request(request.request()),
+                    satrs_minisim::SimDevice::Mgt => self.handle_mgt_request(request.request()),
+                    satrs_minisim::SimDevice::Pcdu => self.handle_pcdu_request(request.request()),
+                },
+                Err(e) => match e {
+                    mpsc::TryRecvError::Empty => break,
+                    mpsc::TryRecvError::Disconnected => {
+                        panic!("all request sender disconnected")
+                    }
+                },
+            }
+        }
+    }
+    fn handle_mgm_request(&mut self, request: &str) {
+        let mgm_request: serde_json::Result<MgmRequest> = serde_json::from_str(request);
+        if mgm_request.is_err() {
+            log::warn!("received invalid MGM request: {}", mgm_request.unwrap_err());
+            return;
+        }
+        let mgm_request = mgm_request.unwrap();
+        match mgm_request {
+            MgmRequest::RequestSensorData => {
+                self.simulation.send_event(
+                    MagnetometerModel::send_sensor_values,
+                    (),
+                    &self.mgm_addr,
+                );
+            }
+        }
+    }
+
+    fn handle_pcdu_request(&mut self, request: &str) {
+        let pcdu_request: serde_json::Result<PcduRequest> = serde_json::from_str(request);
+        if pcdu_request.is_err() {
+            log::warn!(
+                "received invalid PCDU request: {}",
+                pcdu_request.unwrap_err()
+            );
+            return;
+        }
+        let pcdu_request = pcdu_request.unwrap();
+        match pcdu_request {
+            PcduRequest::RequestSwitchInfo => todo!(),
+            PcduRequest::SwitchDevice => todo!(),
+        }
+    }
+
+    fn handle_mgt_request(&mut self, request: &str) {}
+}

satrs-minisim/src/eps.rs

@@ -0,0 +1,74 @@
+use std::{sync::mpsc, time::Duration};
+
+use asynchronix::{
+    model::{Model, Output},
+    time::Scheduler,
+};
+use satrs::power::SwitchStateBinary;
+use satrs_minisim::{SimDevice, SimReply};
+use serde::{Deserialize, Serialize};
+
+pub const SWITCH_INFO_DELAY_MS: u64 = 10;
+
+#[derive(Debug, Clone, PartialEq, Serialize)]
+pub struct SwitchInfo(Vec<SwitchStateBinary>);
+
+#[derive(Debug, Copy, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub enum PcduSwitches {
+    Mgm = 0,
+    Mgt = 1,
+}
+
+#[derive(Debug, Copy, Clone, Serialize, Deserialize)]
+pub enum PcduRequest {
+    SwitchDevice,
+    RequestSwitchInfo,
+}
+
+pub struct PcduModel {
+    pub current_switch_info: Vec<SwitchStateBinary>,
+    pub mgm_switch: Output<SwitchStateBinary>,
+    pub mgt_switch: Output<SwitchStateBinary>,
+    pub reply_sender: mpsc::Sender<SimReply>,
+}
+
+impl PcduModel {
+    pub fn new(reply_sender: mpsc::Sender<SimReply>) -> Self {
+        Self {
+            current_switch_info: vec![SwitchStateBinary::Off; 2],
+            mgm_switch: Output::new(),
+            mgt_switch: Output::new(),
+            reply_sender,
+        }
+    }
+
+    pub async fn request_switch_info(&mut self, _: (), scheduler: &Scheduler<Self>) {
+        scheduler
+            .schedule_event(
+                Duration::from_millis(SWITCH_INFO_DELAY_MS),
+                Self::send_switch_info,
+                (),
+            )
+            .expect("requesting switch info failed");
+    }
+
+    pub fn send_switch_info(&mut self) {
+        let switch_info = SwitchInfo(self.current_switch_info.clone());
+        let reply = SimReply::new(SimDevice::Pcdu, switch_info);
+        self.reply_sender.send(reply).unwrap();
+    }
+
+    pub async fn switch_device(&mut self, switch: PcduSwitches, switch_state: SwitchStateBinary) {
+        self.current_switch_info[switch as usize] = switch_state;
+        match switch {
+            PcduSwitches::Mgm => {
+                self.mgm_switch.send(switch_state).await;
+            }
+            PcduSwitches::Mgt => {
+                self.mgt_switch.send(switch_state).await;
+            }
+        }
+    }
+}
+
+impl Model for PcduModel {}

satrs-minisim/src/lib.rs

@@ -0,0 +1,83 @@
+use serde::{Deserialize, Serialize};
+
+#[derive(Debug, Copy, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub enum SimDevice {
+    Mgm,
+    Mgt,
+    Pcdu,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct SimRequest {
+    device: SimDevice,
+    request: String,
+}
+
+impl SimRequest {
+    pub fn new<T: Serialize>(device: SimDevice, reply: T) -> Self {
+        Self {
+            device,
+            request: serde_json::to_string(&reply).unwrap(),
+        }
+    }
+
+    pub fn device(&self) -> SimDevice {
+        self.device
+    }
+
+    pub fn request(&self) -> &String {
+        &self.request
+    }
+}
+
+#[derive(Serialize, Deserialize)]
+pub struct SimReply {
+    pub device: SimDevice,
+    pub reply: String,
+}
+
+impl SimReply {
+    pub fn new<T: Serialize>(device: SimDevice, reply: T) -> Self {
+        Self {
+            device,
+            reply: serde_json::to_string(&reply).unwrap(),
+        }
+    }
+
+    pub fn reply(&self) -> &String {
+        &self.reply
+    }
+}
+
+pub mod acs {
+    use super::*;
+
+    #[derive(Debug, Copy, Clone, Serialize, Deserialize)]
+    pub enum MgmRequest {
+        RequestSensorData,
+    }
+
+    // Normally, small magnetometers generate their output as a signed 16 bit raw format or something
+    // similar which needs to be converted to a signed float value with physical units. We will
+    // simplify this now and generate the signed float values directly.
+    #[derive(Debug, Copy, Clone, PartialEq, Serialize, Deserialize)]
+    pub struct MgmSensorValues {
+        pub x: f32,
+        pub y: f32,
+        pub z: f32,
+    }
+
+    pub const MGT_GEN_MAGNETIC_FIELD: MgmSensorValues = MgmSensorValues {
+        x: 0.03,
+        y: -0.03,
+        z: 0.03,
+    };
+
+    // Simple model using i16 values.
+    #[derive(Debug, Copy, Clone, PartialEq, Serialize, Deserialize)]
+    pub struct MgtDipole {
+        pub x: i16,
+        pub y: i16,
+        pub z: i16,
+    }
+}

satrs-minisim/src/main.rs

@@ -0,0 +1,175 @@
+use acs::MagnetometerModel;
+use asynchronix::simulation::{Mailbox, SimInit};
+use asynchronix::time::{MonotonicTime, SystemClock};
+use controller::SimController;
+use eps::PcduModel;
+use satrs_minisim::{SimReply, SimRequest};
+use std::sync::mpsc;
+use std::thread;
+use std::time::{Duration, SystemTime};
+use udp::{SharedSocketAddr, UdpTcServer, UdpTmClient};
+
+mod acs;
+mod controller;
+mod eps;
+mod time;
+mod udp;
+
+#[derive(Debug, Copy, Clone, PartialEq, Eq)]
+pub enum ThreadingModel {
+    Default = 0,
+    Single = 1,
+}
+
+fn create_sim_controller(
+    threading_model: ThreadingModel,
+    start_time: MonotonicTime,
+    reply_sender: mpsc::Sender<SimReply>,
+    request_receiver: mpsc::Receiver<SimRequest>,
+) -> SimController {
+    // Instantiate models and their mailboxes.
+    let mgm_model = MagnetometerModel::new(Duration::from_millis(50), reply_sender.clone());
+
+    let mgm_mailbox = Mailbox::new();
+    let mgm_addr = mgm_mailbox.address();
+    let pcdu_mailbox = Mailbox::new();
+    let pcdu_addr = pcdu_mailbox.address();
+
+    let mut pcdu_model = PcduModel::new(reply_sender.clone());
+
+    pcdu_model
+        .mgm_switch
+        .connect(MagnetometerModel::switch_device, &mgm_addr);
+
+    // Instantiate the simulator
+    let sys_clock = SystemClock::from_system_time(start_time, SystemTime::now());
+    let sim_init = if threading_model == ThreadingModel::Single {
+        SimInit::with_num_threads(1)
+    } else {
+        SimInit::new()
+    };
+    let simulation = sim_init
+        .add_model(mgm_model, mgm_mailbox)
+        .add_model(pcdu_model, pcdu_mailbox)
+        .init(start_time);
+    SimController::new(sys_clock, request_receiver, simulation, mgm_addr, pcdu_addr)
+}
+
+fn main() {
+    let shared_socket_addr = SharedSocketAddr::default();
+    let (request_sender, request_receiver) = mpsc::channel();
+    let (reply_sender, reply_receiver) = mpsc::channel();
+    let t0 = MonotonicTime::EPOCH;
+    let mut sim_ctrl =
+        create_sim_controller(ThreadingModel::Default, t0, reply_sender, request_receiver);
+
+    // This thread schedules the simulator.
+    let sim_thread = thread::spawn(move || {
+        sim_ctrl.run(t0);
+    });
+
+    let mut server = UdpTcServer::new(request_sender, shared_socket_addr.clone()).unwrap();
+    // This thread manages the simulator UDP TC server.
+    let udp_tc_thread = thread::spawn(move || {
+        server.run();
+    });
+
+    let mut client = UdpTmClient::new(reply_receiver, 200, shared_socket_addr);
+    // This thread manages the simulator UDP TM client.
+    let udp_tm_thread = thread::spawn(move || {
+        client.run();
+    });
+
+    sim_thread.join().expect("joining simulation thread failed");
+    udp_tc_thread.join().expect("joining UDP TC thread failed");
+    udp_tm_thread.join().expect("joining UDP TM thread failed");
+}
+
+#[cfg(test)]
+mod tests {
+    use delegate::delegate;
+    use satrs_minisim::{
+        acs::{MgmRequest, MgmSensorValues},
+        SimDevice, SimReply, SimRequest,
+    };
+
+    use crate::eps::PcduRequest;
+
+    use super::*;
+
+    struct SimTestbench {
+        pub sim_controller: SimController,
+        pub reply_receiver: mpsc::Receiver<SimReply>,
+        pub request_sender: mpsc::Sender<SimRequest>,
+    }
+
+    impl SimTestbench {
+        fn new() -> Self {
+            let (request_sender, request_receiver) = mpsc::channel();
+            let (reply_sender, reply_receiver) = mpsc::channel();
+            let t0 = MonotonicTime::EPOCH;
+            let sim_ctrl =
+                create_sim_controller(ThreadingModel::Single, t0, reply_sender, request_receiver);
+
+            Self {
+                sim_controller: sim_ctrl,
+                reply_receiver,
+                request_sender,
+            }
+        }
+
+        delegate! {
+            to self.sim_controller {
+                pub fn handle_sim_requests(&mut self);
+            }
+            to self.sim_controller.simulation {
+                pub fn step(&mut self);
+            }
+        }
+
+        pub fn send_request(&self, request: SimRequest) -> Result<(), mpsc::SendError<SimRequest>> {
+            self.request_sender.send(request)
+        }
+
+        pub fn try_receive_next_reply(&self) -> Option<SimReply> {
+            match self.reply_receiver.try_recv() {
+                Ok(reply) => Some(reply),
+                Err(e) => {
+                    if e == mpsc::TryRecvError::Empty {
+                        None
+                    } else {
+                        panic!("reply_receiver disconnected");
+                    }
+                }
+            }
+        }
+    }
+
+    #[test]
+    fn test_basic_mgm_request() {
+        let mut sim_testbench = SimTestbench::new();
+        let mgm_request = MgmRequest::RequestSensorData;
+        let request = SimRequest::new(SimDevice::Mgm, mgm_request);
+        sim_testbench
+            .send_request(request)
+            .expect("sending MGM request failed");
+        sim_testbench.handle_sim_requests();
+        sim_testbench.step();
+        let sim_reply = sim_testbench.try_receive_next_reply();
+        assert!(sim_reply.is_some());
+        let sim_reply = sim_reply.unwrap();
+        assert_eq!(sim_reply.device, SimDevice::Mgm);
+        let reply: MgmSensorValues = serde_json::from_str(&sim_reply.reply)
+            .expect("failed to deserialize MGM sensor values");
+        assert_eq!(reply.x, 0.0);
+        assert_eq!(reply.y, 0.0);
+        assert_eq!(reply.z, 0.0);
+    }
+
+    #[test]
+    fn test_basic_mgm_request_switched_on() {
+        let mut sim_testbench = SimTestbench::new();
+        let pcdu_request = PcduRequest::RequestSwitchInfo;
+        let request = SimRequest::new(SimDevice::Pcdu, pcdu_request);
+    }
+}

satrs-minisim/src/time.rs

@@ -0,0 +1,5 @@
+use asynchronix::time::MonotonicTime;
+
+pub fn current_millis(time: MonotonicTime) -> u64 {
+    (time.as_secs() as u64 * 1000) + (time.subsec_nanos() as u64 / 1_000_000)
+}

satrs-minisim/src/udp.rs

@@ -0,0 +1,152 @@
+use std::{
+    collections::VecDeque,
+    net::{SocketAddr, UdpSocket},
+    sync::{mpsc, Arc, Mutex},
+    time::Duration,
+};
+
+use satrs_minisim::{SimReply, SimRequest};
+
+pub type SharedSocketAddr = Arc<Mutex<Option<SocketAddr>>>;
+
+// A UDP server which handles all TC received by a client application.
+pub struct UdpTcServer {
+    socket: UdpSocket,
+    request_sender: mpsc::Sender<SimRequest>,
+    shared_last_sender: SharedSocketAddr,
+}
+
+impl UdpTcServer {
+    pub fn new(
+        request_sender: mpsc::Sender<SimRequest>,
+        shared_last_sender: SharedSocketAddr,
+    ) -> std::io::Result<Self> {
+        let socket = UdpSocket::bind("0.0.0.0:7303")?;
+        Ok(Self {
+            socket,
+            request_sender,
+            shared_last_sender,
+        })
+    }
+
+    pub fn run(&mut self) {
+        let mut last_socket_addr = None;
+        loop {
+            // Buffer to store incoming data.
+            let mut buffer = [0u8; 4096];
+            // Block until data is received. `recv_from` returns the number of bytes read and the
+            // sender's address.
+            let (bytes_read, src) = self
+                .socket
+                .recv_from(&mut buffer)
+                .expect("could not read from socket");
+
+            // Convert the buffer into a string slice and print the message.
+            let req_string = std::str::from_utf8(&buffer[..bytes_read])
+                .expect("Could not write buffer as string");
+            println!("Received from {}: {}", src, req_string);
+            let sim_req: serde_json::Result<SimRequest> = serde_json::from_str(req_string);
+            if sim_req.is_err() {
+                log::warn!(
+                    "received UDP request with invalid format: {}",
+                    sim_req.unwrap_err()
+                );
+                continue;
+            }
+            self.request_sender.send(sim_req.unwrap()).unwrap();
+            // Only set last sender if it has changed.
+            if last_socket_addr.is_some() && src != last_socket_addr.unwrap() {
+                self.shared_last_sender.lock().unwrap().replace(src);
+            }
+            last_socket_addr = Some(src);
+        }
+    }
+}
+
+// A helper object which sends back all replies to the UDP client.
+//
+// This helper is scheduled separately to minimize the delay between the requests and replies.
+pub struct UdpTmClient {
+    reply_receiver: mpsc::Receiver<SimReply>,
+    reply_queue: VecDeque<SimReply>,
+    max_num_replies: usize,
+    socket: UdpSocket,
+    last_sender: SharedSocketAddr,
+}
+
+impl UdpTmClient {
+    pub fn new(
+        reply_receiver: mpsc::Receiver<SimReply>,
+        max_num_replies: usize,
+        last_sender: SharedSocketAddr,
+    ) -> Self {
+        let socket =
+            UdpSocket::bind("127.0.0.1:0").expect("creating UDP client for TM sender failed");
+        Self {
+            reply_receiver,
+            reply_queue: VecDeque::new(),
+            max_num_replies,
+            socket,
+            last_sender,
+        }
+    }
+
+    pub fn run(&mut self) {
+        loop {
+            let processed_replies = self.process_replies();
+            let last_sender_lock = self
+                .last_sender
+                .lock()
+                .expect("locking last UDP sender failed");
+            let last_sender = *last_sender_lock;
+            drop(last_sender_lock);
+            let mut sent_replies = false;
+            if let Some(last_sender) = last_sender {
+                sent_replies = self.send_replies(last_sender);
+            }
+            if !processed_replies && !sent_replies {
+                std::thread::sleep(Duration::from_millis(20));
+            }
+        }
+    }
+
+    fn process_replies(&mut self) -> bool {
+        let mut processed_replies = false;
+        loop {
+            match self.reply_receiver.try_recv() {
+                Ok(reply) => {
+                    if self.reply_queue.len() >= self.max_num_replies {
+                        self.reply_queue.pop_front();
+                    }
+                    self.reply_queue.push_back(reply);
+                    processed_replies = true;
+                }
+                Err(e) => match e {
+                    mpsc::TryRecvError::Empty => return processed_replies,
+                    mpsc::TryRecvError::Disconnected => {
+                        log::error!("all UDP reply senders disconnected")
+                    }
+                },
+            }
+        }
+    }
+
+    fn send_replies(&mut self, last_sender: SocketAddr) -> bool {
+        let mut sent_replies = false;
+        self.socket
+            .connect(last_sender)
+            .expect("connecting to last sender failed");
+        while !self.reply_queue.is_empty() {
+            let next_reply_to_send = self.reply_queue.pop_front().unwrap();
+            self.socket
+                .send(
+                    serde_json::to_string(&next_reply_to_send)
+                        .unwrap()
+                        .as_bytes(),
+                )
+                .expect("sending reply failed");
+            sent_replies = true;
+        }
+        sent_replies
+    }
+}

satrs/src/power.rs

@@ -24,6 +24,42 @@ pub enum SwitchState {
     Faulty = 3,
 }
 
+#[derive(Debug, Eq, PartialEq, Copy, Clone)]
+#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+pub enum SwitchStateBinary {
+    Off = 0,
+    On = 1,
+}
+
+impl TryFrom<SwitchState> for SwitchStateBinary {
+    type Error = ();
+    fn try_from(value: SwitchState) -> Result<Self, Self::Error> {
+        match value {
+            SwitchState::Off => Ok(SwitchStateBinary::Off),
+            SwitchState::On => Ok(SwitchStateBinary::On),
+            _ => Err(()),
+        }
+    }
+}
+
+impl<T: Into<u64>> From<T> for SwitchStateBinary {
+    fn from(value: T) -> Self {
+        if value.into() == 0 {
+            return SwitchStateBinary::Off;
+        }
+        SwitchStateBinary::On
+    }
+}
+
+impl From<SwitchStateBinary> for SwitchState {
+    fn from(value: SwitchStateBinary) -> Self {
+        match value {
+            SwitchStateBinary::Off => SwitchState::Off,
+            SwitchStateBinary::On => SwitchState::On,
+        }
+    }
+}
+
 pub type SwitchId = u16;
 
 /// Generic trait for a device capable of turning on and off switches.