fix: less warnings, better logging for helios broadcast

copyme.settings.toml

@@ -22,10 +22,10 @@ redis_url = "redis://127.0.0.1:6379/"
 # [dac.etherdream]
 # ip = "192.168.1.68"
 
-#[[transformers]]
+[[transformers]]
-#[transformers.translate]
+[transformers.translate]
-#x = 20
+x = 2000
-#y = 20
+y = 2000
 
 [[transformers]]
 [transformers.replicate]

examples/etherdream.rs

@@ -3,148 +3,149 @@ extern crate ether_dream;
 use ether_dream::dac;
 
 fn main() {
-    println!("Listening for an Ether Dream DAC...");
+	println!("Listening for an Ether Dream DAC...");
 
-    let (dac_broadcast, source_addr) = ether_dream::recv_dac_broadcasts()
+	let (dac_broadcast, source_addr) = ether_dream::recv_dac_broadcasts()
-        .expect("failed to bind to UDP socket")
+		.expect("failed to bind to UDP socket")
-        .filter_map(Result::ok)
+		.filter_map(Result::ok)
-        .next()
+		.next()
-        .unwrap();
+		.unwrap();
-    let mac_address = dac::MacAddress(dac_broadcast.mac_address);
+	let mac_address = dac::MacAddress(dac_broadcast.mac_address);
 
-    println!(
+	println!(
-        "Discovered DAC \"{}\" at \"{}\"! Connecting...",
+		"Discovered DAC \"{}\" at \"{}\"! Connecting...",
-        mac_address, source_addr
+		mac_address, source_addr
-    );
+	);
 
-    // Establish the TCP connection.
+	// Establish the TCP connection.
-    let mut stream = dac::stream::connect(&dac_broadcast, source_addr.ip().clone()).unwrap();
+	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.
+	// 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;
+	let frames_per_second = 60.0;
-    // Lets use the DAC at an eighth the maximum scan rate.
+	// Lets use the DAC at an eighth the maximum scan rate.
-    let points_per_second = 20_000;
+	let points_per_second = stream.dac().max_point_rate / 32;
-    // Determine the number of points per frame given our target frame and point rates.
+	// 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;
+	let points_per_frame = (points_per_second as f32 / frames_per_second) as u16;
 
-    println!(
+	println!(
-        "Preparing for playback:\n\tframe_hz: {}\n\tpoint_hz: {}\n\tpoints_per_frame: {}\n",
+		"Preparing for playback:\n\tframe_hz: {}\n\tpoint_hz: {}\n\tpoints_per_frame: {}\n",
-        frames_per_second, points_per_second, points_per_frame
+		frames_per_second, points_per_second, points_per_frame
-    );
+	);
 
-    // Prepare the DAC's playback engine and await the repsonse.
+	// Prepare the DAC's playback engine and await the repsonse.
-    stream
+	stream
-        .queue_commands()
+		.queue_commands()
-        .prepare_stream()
+		.prepare_stream()
-        .submit()
+		.submit()
-        .err()
+		.err()
-        .map(|err| {
+		.map(|err| {
-            eprintln!(
+			eprintln!(
-                "err occurred when submitting PREPARE_STREAM \
+				"err occurred when submitting PREPARE_STREAM \
                       command and listening for response: {}",
-                err
+				err
-            );
+			);
-        });
+		});
 
-    println!("Beginning playback!");
+	println!("Beginning playback!");
 
-    // The sine wave used to generate points.
+	// The sine wave used to generate points.
-    let mut sine_wave = SineWave {
+	let mut sine_wave = SineWave {
-        point: 0,
+		point: 0,
-        points_per_frame,
+		points_per_frame,
-        frames_per_second,
+		frames_per_second,
-    };
+	};
 
-    // Queue the initial frame and tell the DAC to begin producing output.
+	// Queue the initial frame and tell the DAC to begin producing output.
-    let n_points = points_to_generate(stream.dac());
+	let n_points = points_to_generate(stream.dac());
-    stream
+	stream
-        .queue_commands()
+		.queue_commands()
-        .data(sine_wave.by_ref().take(n_points))
+		.data(sine_wave.by_ref().take(n_points))
-        .begin(0, points_per_second)
+		.begin(0, points_per_second)
-        .submit()
+		.submit()
-        .err()
+		.err()
-        .map(|err| {
+		.map(|err| {
-            eprintln!(
+			eprintln!(
-                "err occurred when submitting initial DATA and BEGIN \
+				"err occurred when submitting initial DATA and BEGIN \
                       commands and listening for response: {}",
-                err
+				err
-            );
+			);
-        });
+		});
+	eprintln!("Stream dac{:?}", stream.dac());
 
-    // Loop and continue to send points forever.
+	// Loop and continue to send points forever.
-    loop {
+	loop {
-        // Determine how many points the DAC can currently receive.
+		// Determine how many points the DAC can currently receive.
-        let n_points = points_to_generate(stream.dac());
+		let n_points = points_to_generate(stream.dac());
-        if let Err(err) = stream
+		if let Err(err) = stream
-            .queue_commands()
+			.queue_commands()
-            .data(sine_wave.by_ref().take(n_points))
+			.data(sine_wave.by_ref().take(n_points))
-            .submit()
+			.submit()
-        {
+		{
-            eprintln!(
+			eprintln!(
-                "err occurred when submitting DATA command and listening \
+				"err occurred when submitting DATA command and listening \
                       for response: {}",
-                err
+				err
-            );
+			);
-            break;
+			break;
-        }
+		}
-    }
+	}
 
-    // Tell the DAC to stop producing output and return to idle. Wait for the response.
+	// Tell the DAC to stop producing output and return to idle. Wait for the response.
-    //
+	//
-    // Note that the DAC is commanded to stop on `Drop` if this is not called and any errors
+	// Note that the DAC is commanded to stop on `Drop` if this is not called and any errors
-    // produced are ignored.
+	// produced are ignored.
-    stream
+	stream
-        .queue_commands()
+		.queue_commands()
-        .stop()
+		.stop()
-        .submit()
+		.submit()
-        .expect("err occurred when submitting STOP command and listening for response");
+		.expect("err occurred when submitting STOP command and listening for response");
 }
 
 // Determine the number of points needed to fill the DAC.
 fn points_to_generate(dac: &ether_dream::dac::Dac) -> usize {
-    dac.buffer_capacity as usize - 1 - dac.status.buffer_fullness as usize
+	dac.buffer_capacity as usize - 1 - dac.status.buffer_fullness as usize
 }
 
 // An iterator that endlessly generates a sine wave of DAC points.
 //
 // The sine wave oscillates at a rate of once per second.
 struct SineWave {
-    point: u32,
+	point: u32,
-    points_per_frame: u16,
+	points_per_frame: u16,
-    frames_per_second: f32,
+	frames_per_second: f32,
 }
 
 impl Iterator for SineWave {
-    type Item = ether_dream::protocol::DacPoint;
+	type Item = ether_dream::protocol::DacPoint;
-    fn next(&mut self) -> Option<Self::Item> {
+	fn next(&mut self) -> Option<Self::Item> {
-        let coloured_points_per_frame = self.points_per_frame - 1;
+		let coloured_points_per_frame = self.points_per_frame - 1;
-        let i = (self.point % self.points_per_frame as u32) as u16;
+		let i = (self.point % self.points_per_frame as u32) as u16;
-        let hz = 1.0;
+		let hz = 1.0;
-        let fract = i as f32 / coloured_points_per_frame as f32;
+		let fract = i as f32 / coloured_points_per_frame as f32;
-        let phase = (self.point as f32 / coloured_points_per_frame as f32) / self.frames_per_second;
+		let phase = (self.point as f32 / coloured_points_per_frame as f32) / self.frames_per_second;
-        let amp = (hz * (fract + phase) * 2.0 * std::f32::consts::PI).sin();
+		let amp = (hz * (fract + phase) * 2.0 * std::f32::consts::PI).sin();
-        let (r, g, b) = match i {
+		let (r, g, b) = match i {
-            i if i == coloured_points_per_frame || i < 13 => (0, 0, 0),
+			i if i == coloured_points_per_frame || i < 13 => (0, 0, 0),
-            _ => (std::u16::MAX, std::u16::MAX, std::u16::MAX),
+			_ => (std::u16::MAX, std::u16::MAX, std::u16::MAX),
-        };
+		};
-        let x_min = std::i16::MIN;
+		let x_min = std::i16::MIN;
-        let x_max = std::i16::MAX;
+		let x_max = std::i8::MAX as i16;
-        let x = (x_min as f32 + fract * (x_max as f32 - x_min as f32)) as i16;
+		let x = (x_min as f32 + fract * (x_max as f32 - x_min as f32)) as i16;
-        let y = (amp * x_max as f32) as i16;
+		let y = (amp * x_max as f32) as i16;
-        let control = 0;
+		let control = 0;
-        let (u1, u2) = (0, 0);
+		let (u1, u2) = (0, 0);
-        let p = ether_dream::protocol::DacPoint {
+		let p = ether_dream::protocol::DacPoint {
-            control,
+			control,
-            x,
+			x,
-            y,
+			y,
-            i,
+			i,
-            r,
+			r,
-            g,
+			g,
-            b,
+			b,
-            u1,
+			u1,
-            u2,
+			u2,
-        };
+		};
-        self.point += 1;
+		self.point += 1;
-        Some(p)
+		Some(p)
-    }
+	}
 }

src/conf.rs

@@ -78,7 +78,7 @@ impl Conf {
 				TransformConf::Intensity(r) => Box::new(*r),
 				TransformConf::Replicate(r) => Box::new(*r),
 				TransformConf::Rotate(r) => Box::new(*r),
-				TransformConf::Translate(r) => Box::new(*r),
+				TransformConf::Translate(t) => Box::new(*t),
 			};
 			v.push(t);
 		}

src/device/etherdream.rs

@@ -1,5 +1,5 @@
 #[warn(unused_imports)]
-use log::{debug, info, warn};
+use log::{ debug, info, warn};
 
 use std::net::SocketAddr;
 use std::thread::sleep;
@@ -14,7 +14,6 @@ use crate::device::{Device, Status, PlaybackState};
 use crate::errors::{LJError, LJResult};
 use crate::point::{Color, Point};
 use ether_dream::protocol::{DacBroadcast, DacResponse};
-use crate::device::PlaybackState::PLAYING;
 
 
 #[warn(dead_code)]
@@ -22,6 +21,18 @@ pub struct EtherdreamDevice {
 	pub conf: EtherDreamConf,
 	dac: DacBroadcast,
 	stream: Stream,
+
+	// "a": ACK   "F": Full  "I": invalid. 64 or 35 for no connection.
+	//     /// The previous command was accepted.
+	//     pub const ACK: u8 = 0x61;
+	//     /// The write command could not be performed because there was not enough buffer space when it
+	//     /// was received.
+	//     pub const NAK_FULL: u8 = 0x46;
+	//     /// The command contained an invalid `command` byte or parameters.
+	//     pub const NAK_INVALID: u8 = 0x49;
+	//     /// An emergency-stop condition still exists.
+	//     pub const NAK_STOP_CONDITION: u8 = 0x21;
+	// }
 	dac_response: u8,
 }
 
@@ -41,7 +52,7 @@ impl EtherdreamDevice {
 		dac_broadcast.set_timeout(Some(time::Duration::new(10, 0)))?;
 		info!("Attempting to get DAC broadcast...");
 		let broadcast = dac_broadcast
-			.take(5)
+			.take(3)
 			.filter_map(|result| {
 				match result {
 					Err(err) => {
@@ -51,8 +62,15 @@ impl EtherdreamDevice {
 					}
 					Ok((dac, source_addr)) => {
 						info!("Valid broadcast, source_addr: {}", source_addr);
-						if source_addr.is_ipv6() { return None; }
+						if source_addr.is_ipv6() {
-						if &source_addr.ip().to_string() != ip { return None; }
+							warn!("Broadcast ignored: ipv6 address.");
+							return None;
+					        }
+						let src_ip = source_addr.ip().to_string();
+						if &src_ip != ip {
+							warn!("Broadcast ignored: expected {ip}, got: {src_ip} ");
+							return None;
+						}
 						Some(Ok((dac, source_addr)))
 					}
 				}
@@ -74,34 +92,17 @@ impl EtherdreamDevice {
 
 	fn get_tcp_stream(dac: &DacBroadcast, source_address: &SocketAddr) -> LJResult<Stream> {
 		let mut stream = connect(dac, source_address.ip())?;
-		debug!("Stream dac BEFORE PREPARE {:?}", stream.dac());
+		match stream
-		debug!("Playback State BEFORE PREPARE {:?}, {:?}", stream.dac().dac.status.playback,  Playback::Playing );
+			.queue_commands()
-		if stream.dac().dac.status.playback == Playback::Playing {
+			.prepare_stream()
-			warn!("DAC was in playback PLAYING, attempting to stop");
+			.submit() {
-			match stream
+			Err(err) => warn!("err occurred when submitting PREPARE_STREAM command and listening for response: {}",err),
-				.queue_commands()
+			Ok(_) => info!("Prepared Stream.")
-				.stop()
-				.submit() {
-				Err(err) => warn!("err occurred when submitting STOP command: {}",err),
-				Ok(_) => info!("Prepared Stream.")
-			}
-		}
-		if stream.dac().dac.status.playback != Playback::Prepared {
-			warn!("DAC was not in playback state PREPARED, attempting to prepare");
-			match stream
-				.queue_commands()
-				.prepare_stream()
-				.submit() {
-				Err(err) => warn!("err occurred when submitting PREPARE_STREAM command and listening for response: {}",err),
-				Ok(_) => info!("Prepared Stream.")
-			}
 		}
 		// 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 = 20_000;
+		let points_per_second = stream.dac().max_point_rate / 32;
-		// let points_per_second = 30_000;
-		debug!("points per second {:?}", points_per_second);
 		// 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;
 
@@ -110,7 +111,6 @@ impl EtherdreamDevice {
 			points_per_frame,
 			frames_per_second,
 		};
-		debug!("Stream dac BEFORE BEGIN {:?}", stream.dac());
 
 		match stream
 			.queue_commands()
@@ -129,15 +129,15 @@ impl EtherdreamDevice {
 		/***
 		Determine the number of points needed to fill the DAC.
 		 ***/
-		let cap = self.dac.buffer_capacity as usize;
+		// Fixme thread 'main' panicked at 'attempt to subtract with overflow', src/device/etherdream.rs:144:24
-		let fullness = self.stream.dac().dac.status.buffer_fullness as usize;
+		let n_points = self.dac.buffer_capacity as usize - self.stream.dac().dac.status.buffer_fullness as usize - 1;
-		// Sometimes we had thread 'main' panicked at 'attempt to subtract with overflow', src/device/etherdream.rs:144:24
-		let n_points = if cap > fullness { cap - fullness } else { 0 };
 		n_points
 	}
 
 	fn ping(&mut self) -> LJResult<()> {
+
 		Ok(self.stream.queue_commands().ping().submit()?)
+
 	}
 }
 
@@ -151,9 +151,6 @@ impl Device for EtherdreamDevice {
 		let now = SystemTime::now();
 		let now: DateTime<Utc> = now.into();
 		let now = now.to_rfc3339();
-		// debug!("Dac Status: {:?} ", status );
-		// debug!("Etherdream Dac {:?} ", self.dac );
-		debug!("Stream dac{:?}", self.stream.dac());
 
 		Status {
 			last_traced_at: now,
@@ -162,6 +159,9 @@ impl Device for EtherdreamDevice {
 			capacity: self.points_capacity(),
 			lack: self.dac_response.to_string(),
 		}
+		// debug!("Dac Status: {:?} ", status );
+		// debug!("Etherdream Dac {:?} ", self.dac );
+		// debug!("Stream dac{:?}", self.stream.dac());
 		// status
 	}
 
@@ -169,30 +169,28 @@ impl Device for EtherdreamDevice {
 	        line: Vec<Point>,
 	        _speed: u32,
 	) -> LJResult<()> {
-		let chunk_size = 512;
+		let chunk_size = 64;
 		let points_iter = line.into_iter();
-		for chunk in points_iter.as_slice().chunks(chunk_size) {
+		for chunk in  points_iter.as_slice().chunks(chunk_size){
 			debug!("New chunk length: {:?}", chunk.len());
-			let capacity = self.points_capacity();
-			debug!("capacity : {:?}", capacity);
 			loop {
-				if chunk.len() > capacity as usize {
+				let capacity = self.points_capacity();
-					debug!("Sleep");
+				if chunk.len() > capacity {
+					debug!("Sleep, capacity : {:?}", capacity);
 					// Sleep for 1/100th of a sec
-					sleep(Duration::new(0, 100_000_000));
+					sleep(Duration::new( 0,     10000000));
-					break;
+					self.ping()?;
-					// self.ping();
 				} else {
 					break;
 				}
 			}
-			debug!("Drawing");
+			debug!("drawing");
 			match self.stream
 				.queue_commands()
 				.data(
-					chunk.into_iter()
+					chunk.iter()
 						.map(|point| (*point).into())
-						.take(chunk_size as usize)
+						.take(chunk_size)
 				)
 				.submit() {
 				Err(err) => {
@@ -225,7 +223,6 @@ impl Device for EtherdreamDevice {
 
 	fn stop(&mut self) -> LJResult<()> {
 		info!("Stopping Etherdream device...");
-		info!("Stream dac{:?}", self.stream.dac());
 		match self.stream
 			.queue_commands()
 			.stop()
@@ -243,8 +240,8 @@ impl Device for EtherdreamDevice {
 	}
 
 	fn grid(&mut self) -> Vec<Point> {
-		let dim_mid = 16000 as f32;
+		let dim_mid = 16000.0;
-		let dim_max = 32000 as f32;
+		let dim_max = 32000.0;
 		let col_min = Color { r: 0, g: 0, b: 0 };
 		let col_max = Color { r: 255, g: 255, b: 255 };
 
@@ -285,10 +282,10 @@ impl Iterator for SineWave {
 		let amp = (hz * (fract + phase) * 2.0 * std::f32::consts::PI).sin();
 		let (r, g, b) = match i {
 			i if i == coloured_points_per_frame || i < 13 => (0, 0, 0),
-			_ => (std::u16::MAX, std::u16::MAX, std::u16::MAX),
+			_ => (u16::MAX, u16::MAX, u16::MAX),
 		};
-		let x_min = std::i16::MIN;
+		let x_min = i16::MIN;
-		let x_max = std::i16::MAX;
+		let x_max = i16::MAX;
 		let x = (x_min as f32 + fract * (x_max as f32 - x_min as f32)) as i16;
 		let y = (amp * x_max as f32) as i16;
 		let control = 0;

src/device/helios.rs

@@ -85,8 +85,8 @@ impl Device for HeliosDevice {
 
 	fn grid(&mut self) -> Vec<Point> {
 		let dim_min = 0 as f32;
-		let dim_mid = 2047 as f32;
+		let dim_mid = 2047.0;
-		let dim_max = 4095 as f32;
+		let dim_max = 4095.0;
 		let col_min = Color { r: 0, g: 0, b: 0 };
 		let col_max = Color { r: 255, g: 255, b: 255 };
 

src/main.rs

@@ -76,7 +76,7 @@ fn run_all() -> LJResult<()> {
 	// Dispatch based on redis requests
 	while running.load(Ordering::SeqCst) {
 		rs.set_status(tracer.status())?;
-		// let _ = framerate_handler.handle_time()?;
+		framerate_handler.handle_time()?;
 		let order = rs.get_order(config.laser_id)?;
 		match order {
 			Order::Draw | Order::Black | Order::Grid => {

src/point.rs

@@ -15,7 +15,6 @@ pub struct Color {
 	pub g: u8,
 	pub b: u8,
 }
-
 impl Mul<u8> for Color {
     type Output = Self;
 

src/redis_ctrl.rs

@@ -98,10 +98,10 @@ impl RedisCtrl {
 
 	pub fn init_world_state(&mut self) -> LJResult<WorldState> {
 		Ok(WorldState {
-			client_key: self.get_client_key().unwrap(),
+			client_key: self.get_client_key()?,
-			edh: self.get_edh().unwrap(),
+			edh: self.get_edh()?,
-			kpps: self.get_int("kpps").unwrap().try_into().unwrap(),
+			kpps: self.get_int("kpps")?.try_into()?,
-			intensity: self.get_int("intensity").unwrap().try_into().unwrap(),
+			intensity: self.get_int("intensity")?.try_into()?,
 			..WorldState::default()
 		})
 	}

src/transformer/helios_to_etherdream.rs

@@ -7,15 +7,16 @@ use serde::{Serialize, Deserialize};
 /// Converts helios Geometry to Helios
 #[allow(dead_code)]
 #[derive(Serialize, Deserialize, Debug, Clone, Copy)]
-pub struct HeliosToEtherdream {}
+pub struct HeliosToEtherdream {
+}
 
 impl Transformers for HeliosToEtherdream {
 	fn apply(&self, point_list: &[Point], _ws: &WorldState) -> Vec<Point> {
 		// debug!("list helios {:?}", point_list);
 		let out = point_list.iter().map(|pt| {
 			Point {
-				x: 50.0 * (320.0 - pt.x),
+				x: 8.0 * (pt.x - 2047.0),
-				y: 40.0 * (240.0 - pt.y),
+				y: 8.0 * (pt.y - 2047.0),
 				..*pt
 			}
 		}).collect();

src/transformer/intensity.rs

@@ -8,18 +8,20 @@ use serde::{Serialize, Deserialize};
 /// Converts helios Geometry to Helios
 #[allow(dead_code)]
 #[derive(Serialize, Deserialize, Debug, Clone, Copy)]
-pub struct Intensity {}
+pub struct Intensity {
+}
 
 impl Transformers for Intensity {
 	fn apply(&self, point_list: &[Point], ws: &WorldState) -> Vec<Point> {
 		// debug!("list helios {:?}", point_list);
 		let out = point_list.iter().map(|pt| {
 			Point {
-				color: pt.color * ws.intensity,
+				x: pt.x,
-				..*pt
+				y: pt.y,
-		}
+				color: pt.color * ws.intensity
-	}).collect();
+			}
-		// debug!("list intensity {:?}", out);
+		}).collect();
+		debug!("list intensity {:?}", out);
 		out
 	}
 }