feat: adds Etherdream device

src/device.rs

@@ -1,4 +1,3 @@
-
 mod helios;
 mod dummy;
 mod etherdream;
@@ -10,6 +9,7 @@ use crate::device::dummy::DummyDevice;
 use crate::errors::LJResult;
 use crate::point::Point;
 use serde::Serialize;
+use crate::device::etherdream::EtherdreamDevice;
 
 /*
 self.protocol_version,
@@ -27,22 +27,24 @@ self.point_count
 #[derive(Debug, PartialEq, Serialize, Copy, Clone)]
 pub enum PlaybackState {
 	IDLE = 0,
-	PREPARE,
+	PREPARE = 1,
-	PLAYING,
+	PLAYING = 2,
-}
+	UNKNOWN = 99,
-impl fmt::Display for PlaybackState {
-    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
-        write!(f, "{:?}", self)
-    }
 }
 
-#[derive(Debug, Default )]
+impl fmt::Display for PlaybackState {
+	fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+		write!(f, "{:?}", self)
+	}
+}
+
+#[derive(Debug)]
 pub struct Status {
 	pub last_traced_at: String,
 	pub properties: Vec<String>,
 	pub playback_state: PlaybackState,
 	pub capacity: u16,
-	pub lack: String
+	pub lack: String,
 }
 
 // /lstt/lasernumber       etherdream last_status.playback_state  (0: idle   1: prepare   2: playing)
@@ -50,7 +52,7 @@ pub struct Status {
 // /lack/lasernumber       "a": ACK   "F": Full  "I": invalid. 64 or 35 for no connection.
 
 pub trait Device {
-	fn status(&self) -> Status;
+	fn status(&mut self) -> Status;
 	fn draw(
 		&mut self,
 		frame: Vec<Point>,
@@ -61,10 +63,10 @@ pub trait Device {
 }
 
 pub fn device_factory(config: &Conf) -> LJResult<Box<dyn Device>> {
-	let device : Box<dyn Device> = match &config.dac {
+	let device: Box<dyn Device> = match &config.dac {
-	    DacFamily::Helios(conf) => Box::new(HeliosDevice::new(conf)?),
+		DacFamily::Helios(conf) => Box::new(HeliosDevice::new(conf)?),
-	    DacFamily::Etherdream(_conf) => todo!(),
+		DacFamily::Etherdream( conf) => Box::new( EtherdreamDevice::new(conf)?),
-	    DacFamily::Dummy => Box::new(DummyDevice::new()?)
+		DacFamily::Dummy => Box::new(DummyDevice::new()?)
 	};
 	Ok(device)
 }

src/device/dummy.rs

@@ -14,7 +14,7 @@ impl DummyDevice {
 }
 
 impl Device for DummyDevice {
-	fn status(&self) -> Status {
+	fn status(&mut self) -> Status {
 		Status {
 			last_traced_at: "never".to_string(),
 			properties: vec!["foo".to_string()],

src/device/etherdream.rs

@@ -1,108 +1,147 @@
-use ether_dream::{dac};
+use std::time;
-
+use std::net::SocketAddr;
-
+use ether_dream::dac::stream::connect;
+use ether_dream::dac::Stream;
 
 use crate::conf::EtherDreamConf;
 use crate::device::{Device, Status, PlaybackState};
 use crate::errors::{LJError, LJResult};
-use crate::point::Point;
+use crate::point::{Color, Point};
-use chrono::Utc;
+use ether_dream::protocol::{DacBroadcast, DacStatus};
+use log::info;
 
+#[warn(dead_code)]
 pub struct EtherdreamDevice {
 	pub conf: EtherDreamConf,
-	dac: dac,
+	dac: DacBroadcast,
-	sent_points: u16,
+	// source_address: SocketAddr,
-	state: PlaybackState,
+	stream: Stream,
+	// sent_points: u16,
 	lack: String,
-	last_traced_at : String
+	last_traced_at: String,
 }
 
 impl EtherdreamDevice {
 	pub fn new(conf: &EtherDreamConf) -> LJResult<Self> {
-		let id = conf.id;
+		let (dac, _source_address, stream) = EtherdreamDevice::get_dac(conf)?;
-
+		// let (dac, source_address) = EtherdreamDevice::get_dac(conf)?;
-		// // Todo : use the config
-	    // let (dac_broadcast, source_addr) = ether_dream::recv_dac_broadcasts()
-	    //     .expect("failed to bind to UDP socket")
-	    //     .filter_map(Result::ok)
-	    //     .next()
-	    //     .unwrap();
-	    // let mac_address = dac::MacAddress(dac_broadcast.mac_address);
-	    //
-	    // println!(
-	    //     "Discovered DAC \"{}\" at \"{}\"! Connecting...",
-	    //     mac_address, source_addr
-	    // );
-
-
-	    // Establish the TCP connection.
-	    let mut stream = dac::stream::connect(&dac_broadcast, source_addr.ip().clone()).unwrap();
-
-	    // If we want to create an animation (in our case a moving sine wave) we need a frame rate.
-	    let frames_per_second = 60.0;
-	    // Lets use the DAC at an eighth the maximum scan rate.
-	    let points_per_second = stream.dac().max_point_rate / 32;
-	    // Determine the number of points per frame given our target frame and point rates.
-	    let points_per_frame = (points_per_second as f32 / frames_per_second) as u16;
-
-	    // Prepare the DAC's playback engine and await the repsonse.
-	    stream
-	        .queue_commands()
-	        .prepare_stream()
-	        .submit()
-	        .err()
-	        .map(|err| {
-	            eprintln!(
-	                "err occurred when submitting PREPARE_STREAM \
-	                      command and listening for response: {}",
-	                err
-	            );
-	        });
-
-	    println!("Ready for playback!");
-
 		Ok(Self {
 			conf: (*conf).clone(),
 			dac,
-			sent_points: 0,
+			// source_address,
-			state: PlaybackState::PREPARE,
+			stream,
+			// sent_points: 0,
 			lack: "".to_string(),
-			last_traced_at: "1985-04-12T23:20:50.52Z".to_string()
+			last_traced_at: "1985-04-12T23:20:50.52Z".to_string(),
 		})
 	}
+	pub fn get_dac(conf: &EtherDreamConf) -> LJResult<(DacBroadcast, SocketAddr, Stream)> {
+		let ip = &conf.ip;
+		let dac_broadcast = ether_dream::recv_dac_broadcasts()?;
+		dac_broadcast.set_nonblocking(true)?;
+		dac_broadcast.set_timeout(Some(time::Duration::new(60, 30)))?;
+		let broadcast = dac_broadcast
+			.filter_map(|result| {
+				info!("Received new broadcast {:?}", result);
+				if result.is_err() { return None; }
+				let (dac, source_addr) = result.unwrap();
+				if source_addr.is_ipv6() { return None; }
+				if source_addr.ip().to_string() != ip.clone() { return None; }
+				info!("Valid broadcast");
+				Some(Ok((dac, source_addr)))
+			})
+			.next()
+			.ok_or(LJError::EtherdreamError("Failed to receive broadcast".to_string()))?;
+		match broadcast {
+			Err(err) => {
+				Err(Box::new(LJError::EtherdreamConnectError(err)))
+			}
+			Ok((dac, source_addr)) => {
+				let stream = EtherdreamDevice::get_tcp_stream(&dac, &source_addr)?;
+				Ok((dac, source_addr, stream))
+			}
+		}
+	}
+
+	pub fn get_tcp_stream(dac: &DacBroadcast, source_address: &SocketAddr) -> LJResult<Stream> {
+		// Establish the TCP connection.
+		let mut stream = connect(dac, source_address.ip())?;
+
+		// Prepare  stream
+		stream
+			.queue_commands()
+			.prepare_stream()
+			.submit()
+			.err()
+			.map(|err| {
+				eprintln!(
+					"err occurred when submitting PREPARE_STREAM \
+		                  command and listening for response: {}",
+					err
+				);
+			});
+
+		Ok(stream)
+	}
+
+
+	pub fn check_tcp_stream(&mut self) -> LJResult<()> {
+		// todo Reinit stream if needed
+		// self.stream = EtherdreamDevice::get_tcp_stream(&self.dac, &self.source_address)?
+		Ok(())
+	}
+
+	// Determine the number of points needed to fill the DAC.
+	fn points_to_generate(&self) -> usize {
+		self.dac.buffer_capacity as usize - 1 - self.dac.dac_status.buffer_fullness as usize
+	}
 }
 
 impl Device for EtherdreamDevice {
-	fn status(&self) -> Status {
+	fn status(&mut self) -> Status {
+		let _ = self.check_tcp_stream();
+
+		// "a": ACK   "F": Full  "I": invalid. 64 or 35 for no connection.
+		let playback_state = match self.dac.dac_status.playback_state {
+			DacStatus::PLAYBACK_IDLE => PlaybackState::IDLE,
+			DacStatus::PLAYBACK_PREPARED => PlaybackState::PREPARE,
+			DacStatus::PLAYBACK_PLAYING => PlaybackState::PLAYING,
+			_ => PlaybackState::UNKNOWN
+		};
 
-		let lack = self.lack.clone();
 		Status {
-			last_traced_at:  self.last_traced_at.clone(),
+			last_traced_at: self.last_traced_at.clone(),
 			properties: vec!["foo".to_string()],
-			playback_state: self.state,
+			playback_state,
-			capacity: self.sent_points,
+			capacity: self.dac.dac_status.buffer_fullness,
-			lack,
+			lack: String::from(&self.lack),
 		}
 	}
 
 	fn draw(&mut self,
 	        line: Vec<Point>,
-	        speed: u32,
+	        _speed: u32,
 	) -> LJResult<()> {
-		self.state = PlaybackState::IDLE;
+		let n_points = self.points_to_generate();
-		while let Ok(DeviceStatus::NotReady) = self.dac.status() {}
+		self.stream
-		self.state = PlaybackState::PLAYING;
+			.queue_commands()
-
+			.data(line.into_iter().map(|point| point.into()).take(n_points))
-		let points: Vec<helios_dac::Point> = line.into_iter().map(|p| p.into()).collect();
+			.submit()?;
-		let frame = Frame::new(speed, points.clone());
-		self.dac.write_frame(frame.clone())?;
-		self.sent_points = points.len() as u16;
-		self.last_traced_at = Utc::now().to_rfc3339();
 		Ok(())
 	}
 
 	fn stop(&mut self) -> LJResult<()> {
-		self.dac.stop()?;
+		self.stream
+			.queue_commands()
+			.stop()
+			.submit()
+			.expect("err occurred when submitting STOP command and listening for response");
 		Ok(())
 	}
+
+	fn grid(&mut self) -> Vec<Point> {
+		vec!(
+			Point { x: 0.0, y: 0.0, color: Color { r: 255, g: 255, b: 255 } }
+		)
+	}
 }

src/device/helios.rs

@@ -46,7 +46,7 @@ impl HeliosDevice {
 }
 
 impl Device for HeliosDevice {
-	fn status(&self) -> Status {
+	fn status(&mut self) -> Status {
 		let lack = self.lack.clone();
 		Status {
 			last_traced_at: self.last_traced_at.clone(),

src/errors.rs

@@ -1,5 +1,5 @@
 use std::error::Error;
-use std::fmt;
+use std::{fmt, io};
 use redis::RedisError;
 use config::ConfigError;
 
@@ -7,41 +7,48 @@ pub type LJResult<T> = Result<T, Box<dyn std::error::Error>>;
 
 #[derive(Debug)]
 pub enum LJError {
-    Config(ConfigError),
+	Config(ConfigError),
-    RedisConnect(RedisError),
+	RedisConnect(RedisError),
-    HeliosDeviceMissing,
+	HeliosDeviceMissing,
-    BadEDH
+	BadEDH,
+	EtherdreamError(String),
+	EtherdreamConnectError(io::Error),
 }
 
 impl fmt::Display for LJError {
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+	fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-	use LJError::*;
+		use LJError::*;
-	
-	match self {
-	    Config(err) => {
-		write!(f, "unable to load config file: {err}")
-	    },
 
-	    RedisConnect(err) => {
+		match self {
-		write!(f, "unable to connect to redis server: {err}")
+			Config(err) => {
-	    },
+				write!(f, "unable to load config file: {err}")
-	    HeliosDeviceMissing => {
+			}
-		write!(f, "helios device not found")
+			RedisConnect(err) => {
-	    },
+				write!(f, "unable to connect to redis server: {err}")
-	    BadEDH => {
+			}
-		write!(f, "EDH matrix is not a 3x3 matrix")
+			HeliosDeviceMissing => {
-	    }
+				write!(f, "helios device not found")
+			}
+			BadEDH => {
+				write!(f, "EDH matrix is not a 3x3 matrix")
+			}
+			EtherdreamError(msg) => {
+				write!(f, "Unexpected Etherdream device error: {msg}")
+			}
+			EtherdreamConnectError(err) => {
+				write!(f, "Failed to retrieve Etherdream device: {err}")
+			}
+		}
 	}
-    }
 }
 
 impl Error for LJError {
-    fn source(&self) -> Option<&(dyn Error + 'static)> {
+	fn source(&self) -> Option<&(dyn Error + 'static)> {
-	use LJError::*;
+		use LJError::*;
 
-	match self {
+		match self {
-	    RedisConnect(err) => Some(err),
+			RedisConnect(err) => Some(err),
-	    _ => None
+			_ => None
+		}
 	}
-    }
 }

src/main.rs

@@ -114,7 +114,7 @@ fn run_all() -> LJResult<()> {
 				// 4 : Resampler Change (longs and shorts lsteps)
 				// 5 : Client Key Change        = reread redis key /clientkey
 				// 8 : color balance change     = reread redis keys /red /green /blue
-				// 9 : poweroff LJ				world_state.intensity = rs.get_int("intensity")?
+				// 9 : poweroff LJ
 				info!("Order: {:?}", order);
 			}
 		}
@@ -152,17 +152,15 @@ fn get_next_frame(
 
 	// Handle the grid case
 
-	let mut line: Vec<Point>;
+	let mut line: Vec<Point> = if world_state.draw_grid {
-	if world_state.draw_grid {
+		world_state.grid.clone()
-		line = world_state.grid.clone();
 	} else {
 		let redis_line = rs.get_line(&format_key)?;
-		line = redis_line.into_iter()
+		redis_line.into_iter()
 			.map(|tpl| tpl.into())
-			.collect();
+			.collect()
 	};
 
-
 	for transformer in transformers {
 		line = transformer.apply(&line, world_state);
 	}

src/point.rs

@@ -1,3 +1,5 @@
+use ether_dream::protocol::DacPoint;
+
 #[derive(Debug, Clone, Copy, Default, PartialEq)]
 pub struct Point {
 	pub x: f32,
@@ -41,3 +43,22 @@ impl From<Point> for helios_dac::Point {
 		}
 	}
 }
+
+impl From<Point> for DacPoint {
+	fn from(pt: Point) -> DacPoint {
+		let control = 0;
+		let (u1, u2) = (0, 0);
+		let i = 255;
+		DacPoint {
+			control,
+			x: pt.x as i16,
+			y: pt.y as i16,
+			i,
+			r: pt.color.r.into(),
+			g: pt.color.g.into(),
+			b: pt.color.b.into(),
+			u1,
+			u2,
+		}
+	}
+}