wip: dirty state for debugging etherdream

copyme.settings.toml

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

examples/etherdream.rs

@@ -0,0 +1,150 @@
+extern crate ether_dream;
+
+use ether_dream::dac;
+
+fn main() {
+    println!("Listening for an Ether Dream DAC...");
+
+    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 = 20_000;
+    // 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;
+
+    println!(
+        "Preparing for playback:\n\tframe_hz: {}\n\tpoint_hz: {}\n\tpoints_per_frame: {}\n",
+        frames_per_second, points_per_second, points_per_frame
+    );
+
+    // 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!("Beginning playback!");
+
+    // The sine wave used to generate points.
+    let mut sine_wave = SineWave {
+        point: 0,
+        points_per_frame,
+        frames_per_second,
+    };
+
+    // Queue the initial frame and tell the DAC to begin producing output.
+    let n_points = points_to_generate(stream.dac());
+    stream
+        .queue_commands()
+        .data(sine_wave.by_ref().take(n_points))
+        .begin(0, points_per_second)
+        .submit()
+        .err()
+        .map(|err| {
+            eprintln!(
+                "err occurred when submitting initial DATA and BEGIN \
+                      commands and listening for response: {}",
+                err
+            );
+        });
+
+    // Loop and continue to send points forever.
+    loop {
+        // Determine how many points the DAC can currently receive.
+        let n_points = points_to_generate(stream.dac());
+        if let Err(err) = stream
+            .queue_commands()
+            .data(sine_wave.by_ref().take(n_points))
+            .submit()
+        {
+            eprintln!(
+                "err occurred when submitting DATA command and listening \
+                      for response: {}",
+                err
+            );
+            break;
+        }
+    }
+
+    // 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
+    // produced are ignored.
+    stream
+        .queue_commands()
+        .stop()
+        .submit()
+        .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
+}
+
+// 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,
+    points_per_frame: u16,
+    frames_per_second: f32,
+}
+
+impl Iterator for SineWave {
+    type Item = ether_dream::protocol::DacPoint;
+    fn next(&mut self) -> Option<Self::Item> {
+        let coloured_points_per_frame = self.points_per_frame - 1;
+        let i = (self.point % self.points_per_frame as u32) as u16;
+        let hz = 1.0;
+        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 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),
+        };
+        let x_min = std::i16::MIN;
+        let x_max = std::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;
+        let (u1, u2) = (0, 0);
+        let p = ether_dream::protocol::DacPoint {
+            control,
+            x,
+            y,
+            i,
+            r,
+            g,
+            b,
+            u1,
+            u2,
+        };
+        self.point += 1;
+        Some(p)
+    }
+}

examples/populate_redis.rs

@@ -1,24 +1,34 @@
-///
-/// $ cargo run --example populate_redis
-///
+use std::io::_print;
+/**
+
+# Populate Redis Example
+
+**This script simulates the redis content provided by the LJ Python / web tool**
+
+$ cargo run --example populate_redis
+ **/
 
 use redis::{
-    //RedisResult,
-    Client,
-    Commands,
-    Connection,
+	//RedisResult,
+	Client,
+	Commands,
+	Connection,
 };
 
 fn do_something() -> redis::RedisResult<()> {
-    let client = Client::open("redis://127.0.0.1/")?;
-    let mut con: Connection = client.get_connection()?;
-
+	let client = Client::open("redis://127.0.0.1/")?;
+	let mut con: Connection = client.get_connection()?;
 	let _ = con.set("/clientkey", "/pl/0/")?;
 	let _ = con.set("/EDH/0", "[[1.0, 0.0, 0.0],\n [ 0.0, 1.0, 0.0],\n [ 0.0, 0.0, 1.0]]")?;
 	let _ = con.set("/kpps/0", "5000")?;
 	let _ = con.set("/intensity/0", "255")?;
-    Ok(())
+	let _ = con.set("/pl/0/0", "[(1000, 2000, 0), (1000, 1000, 65535), (2000, 1000, 65535), (2000, 2000, 65535), (1000, 2000, 65535)]")?;
+	Ok(())
 }
+
 fn main() {
-    _ = do_something();
+	match do_something() {
+		Err(err)    => println!("Something wrong occured: {:?}", err),
+		Ok(..)                  => println!("Successfully inserted content in Redis")
+	}
 }

examples/test.rs

@@ -0,0 +1,46 @@
+///
+/// $ cargo run --example simple_client
+///
+
+use redis::{
+	//RedisResult,
+	Client,
+	Commands,
+	Connection,
+};
+
+use std::time::Instant;
+
+fn do_something() -> redis::RedisResult<()> {
+	let client = Client::open("redis://127.0.0.1/")?;
+	let mut con: Connection = client.get_connection()?;
+	let start = Instant::now();
+
+	loop {
+		let elapsed = start.elapsed();
+		let time = 60.0 * elapsed.as_millis() as f32 / 1000.0;
+
+		let mut v: Vec<(f32, f32, u32)> = vec![];
+
+		for i in 0..128 {
+			let a = (time + i as f32) / 128.0 * std::f32::consts::PI * 2.0;
+			let r = 1200.0 + (a * 5.0).cos() * (500.0 * (time / 5.0).cos());
+
+			let x = a.cos() * r;
+			let y = a.sin() * r;
+			let col = if i % 8 < 4 {
+				0x000000ff
+			} else {
+				0x00ff0000
+			};
+			v.push((x, y, col));
+		}
+		// println!("{:?}", v);
+		let _ = con.set("/pl/0/0", format!("{:?}", v))?;
+	}
+	// Ok(())
+}
+
+fn main() {
+	_ = do_something();
+}

src/conf.rs

@@ -1,98 +1,106 @@
 use config::Config;
-use serde::{Serialize,Deserialize};
-use crate::errors::{LJError,LJResult};
+use serde::{Serialize, Deserialize};
+use crate::errors::{LJError, LJResult};
 use crate::transformer;
 
 #[derive(Serialize, Deserialize, Debug, Clone)]
 pub struct Conf {
-    pub laser_id: u8,
-    pub debug: bool,
-    pub redis_url: String,
-    pub dac: DacFamily,
-    #[serde(default)]
-    pub transformers: Vec<TransformConf>
+	pub laser_id: u8,
+	pub debug: bool,
+	pub redis_url: String,
+	pub dac: DacFamily,
+	#[serde(default)]
+	pub transformers: Vec<TransformConf>,
 }
 
 #[derive(Serialize, Deserialize, Debug, Clone)]
 pub enum DacFamily {
-    #[serde(rename = "helios")] 
-    Helios(HeliosConf),
-    #[serde(rename = "etherdream")] 
-    Etherdream(EtherDreamConf),
-    #[serde(rename = "dummy")] 
-    Dummy,
+	#[serde(rename = "helios")]
+	Helios(HeliosConf),
+	#[serde(rename = "etherdream")]
+	Etherdream(EtherDreamConf),
+	#[serde(rename = "dummy")]
+	Dummy,
 }
 
 #[derive(Serialize, Deserialize, Debug, Clone)]
 pub struct HeliosConf {
-    pub id: u8
+	pub id: u8,
 }
 
 #[derive(Serialize, Deserialize, Debug, Clone)]
 pub struct EtherDreamConf {
-    pub ip: String
+	pub ip: String,
 }
 
 #[derive(Serialize, Deserialize, Debug, Clone)]
 pub enum TransformConf {
-    #[serde(rename = "translate")] 
-    Translate(transformer::Translate),
-    #[serde(rename = "replicate")] 
-    Replicate(transformer::Replicate),
-    #[serde(rename = "rotate")] 
-    Rotate(transformer::Rotate),
-    #[serde(rename = "flip_horizontal")] 
-    FlipH(transformer::FlipHorizontal),
-    #[serde(rename = "flip_vertical")] 
-    FlipV(transformer::FlipVertical),    
-    #[serde(rename = "grid")] 
-    Grid(transformer::Grid),
-    #[serde(rename = "homography")] 
-    Homography(transformer::Homography)
+	#[serde(rename = "translate")]
+	Translate(transformer::Translate),
+	#[serde(rename = "replicate")]
+	Replicate(transformer::Replicate),
+	#[serde(rename = "rotate")]
+	Rotate(transformer::Rotate),
+	#[serde(rename = "flip_horizontal")]
+	FlipH(transformer::FlipHorizontal),
+	#[serde(rename = "flip_vertical")]
+	FlipV(transformer::FlipVertical),
+	#[serde(rename = "grid")]
+	Grid(transformer::Grid),
+	#[serde(rename = "homography")]
+	Homography(transformer::Homography),
+	#[serde(rename = "helios_to_etherdream")]
+	HeliosToEtherdream(transformer::HeliosToEtherdream),
+	#[serde(rename = "intensity")]
+	Intensity(transformer::Intensity),
 }
 
 
 impl Conf {
-    pub fn new(path: &str) -> LJResult<Conf> {
-	let settings = Config::builder()
-            .add_source(config::File::with_name(path))
-	    .build()?;
-        
-	let conf : Conf = settings.try_deserialize().map_err(LJError::Config)?;
-	Ok(conf)
-    }
+	pub fn new(path: &str) -> LJResult<Conf> {
+		let settings = Config::builder()
+			.add_source(config::File::with_name(path))
+			.build()?;
 
-    pub fn get_transformers(&self) -> Vec<Box<dyn transformer::Transformers>> {
-	let mut v = vec![];
-	for t in &self.transformers {
-	    let t : Box<dyn transformer::Transformers> = match t {
-		TransformConf::Translate(t) => Box::new(*t),
-		TransformConf::Replicate(r) => Box::new(*r),
-		TransformConf::Rotate(r) => Box::new(*r),
-		TransformConf::FlipH(r) => Box::new(*r),
-		TransformConf::FlipV(r) => Box::new(*r),		
-		TransformConf::Grid(r) => Box::new(*r),
-		TransformConf::Homography(r) => Box::new(*r),
-	    };
-	    v.push(t);
+		let conf: Conf = settings.try_deserialize().map_err(LJError::Config)?;
+		Ok(conf)
 	}
-	v
-    }
 
-    #[allow(dead_code)]
-    pub fn dump() {
-	let conf = Conf {
-	    laser_id: 0,
-	    debug: true,
-	    redis_url: "redis://127.0.0.1:6379/".to_string(),
-	    dac: DacFamily::Helios(HeliosConf { id: 0 }),
-	    transformers: vec![
-		TransformConf::Translate(transformer::Translate { x: 2000.0,
-								  y: 2000.0 } ),
-		TransformConf::Replicate(transformer::Replicate::Until(48))
-	    ]
-	};
-	let s = toml::to_string(&conf).unwrap();
-	println!("{}", s);
-    }
+	pub fn get_transformers(&self) -> Vec<Box<dyn transformer::Transformers>> {
+		let mut v = vec![];
+		for t in &self.transformers {
+			let t: Box<dyn transformer::Transformers> = match t {
+				TransformConf::FlipH(r) => Box::new(*r),
+				TransformConf::FlipV(r) => Box::new(*r),
+				TransformConf::Grid(r) => Box::new(*r),
+				TransformConf::HeliosToEtherdream(r) => Box::new(*r),
+				TransformConf::Homography(r) => Box::new(*r),
+				TransformConf::Intensity(r) => Box::new(*r),
+				TransformConf::Replicate(r) => Box::new(*r),
+				TransformConf::Rotate(r) => Box::new(*r),
+				TransformConf::Translate(r) => Box::new(*r),
+			};
+			v.push(t);
+		}
+		v
+	}
+
+	#[allow(dead_code)]
+	pub fn dump() {
+		let conf = Conf {
+			laser_id: 0,
+			debug: true,
+			redis_url: "redis://127.0.0.1:6379/".to_string(),
+			dac: DacFamily::Helios(HeliosConf { id: 0 }),
+			transformers: vec![
+				TransformConf::Translate(transformer::Translate {
+					x: 2000.0,
+					y: 2000.0,
+				}),
+				TransformConf::Replicate(transformer::Replicate::Until(48)),
+			],
+		};
+		let s = toml::to_string(&conf).unwrap();
+		println!("{}", s);
+	}
 }

src/device.rs

@@ -28,8 +28,7 @@ self.point_count
 pub enum PlaybackState {
 	IDLE = 0,
 	PREPARE = 1,
-	PLAYING = 2,
-	UNKNOWN = 99,
+	PLAYING = 2
 }
 
 impl fmt::Display for PlaybackState {
@@ -43,7 +42,7 @@ pub struct Status {
 	pub last_traced_at: String,
 	pub properties: Vec<String>,
 	pub playback_state: PlaybackState,
-	pub capacity: u16,
+	pub capacity: usize,
 	pub lack: String,
 }
 

src/device/etherdream.rs

@@ -1,57 +1,58 @@
-use std::time;
+#[warn(unused_imports)]
+use log::{debug, info, warn};
+
 use std::net::SocketAddr;
-use ether_dream::dac::stream::connect;
-use ether_dream::dac::Stream;
+use std::thread::sleep;
+use ether_dream::dac::stream::{CommunicationError, connect};
+use ether_dream::dac::{Playback, Stream};
+use chrono::{DateTime, Utc};
+use std::time;
+use std::time::{Duration, SystemTime};
 
 use crate::conf::EtherDreamConf;
 use crate::device::{Device, Status, PlaybackState};
 use crate::errors::{LJError, LJResult};
 use crate::point::{Color, Point};
-use ether_dream::protocol::{DacBroadcast, DacStatus};
-use log::{info, warn};
+use ether_dream::protocol::{DacBroadcast, DacResponse};
+use crate::device::PlaybackState::PLAYING;
+
 
 #[warn(dead_code)]
 pub struct EtherdreamDevice {
 	pub conf: EtherDreamConf,
 	dac: DacBroadcast,
-	// source_address: SocketAddr,
 	stream: Stream,
-	// sent_points: u16,
-	lack: String,
-	last_traced_at: String,
+	dac_response: u8,
 }
 
 impl EtherdreamDevice {
 	pub fn new(conf: &EtherDreamConf) -> LJResult<Self> {
-		let (dac, _source_address, stream) = EtherdreamDevice::get_dac(conf)?;
-		// let (dac, source_address) = EtherdreamDevice::get_dac(conf)?;
+		let (dac, _source_address, stream) = EtherdreamDevice::connect(conf)?;
 		Ok(Self {
 			conf: (*conf).clone(),
 			dac,
-			// source_address,
 			stream,
-			// sent_points: 0,
-			lack: "".to_string(),
-			last_traced_at: "1985-04-12T23:20:50.52Z".to_string(),
+			dac_response: DacResponse::ACK,
 		})
 	}
-	pub fn get_dac(conf: &EtherDreamConf) -> LJResult<(DacBroadcast, SocketAddr, Stream)> {
+	fn connect(conf: &EtherDreamConf) -> LJResult<(DacBroadcast, SocketAddr, Stream)> {
 		let ip = &conf.ip;
 		let dac_broadcast = ether_dream::recv_dac_broadcasts()?;
 		dac_broadcast.set_timeout(Some(time::Duration::new(10, 0)))?;
 		info!("Attempting to get DAC broadcast...");
 		let broadcast = dac_broadcast
+			.take(5)
 			.filter_map(|result| {
 				match result {
 					Err(err) => {
 						warn!( "Failed to find a valid DAC via broadcast. Error: {:?}", err);
 						info!( "Retrying...");
 						None
-					},
+					}
 					Ok((dac, source_addr)) => {
+						info!("Valid broadcast, source_addr: {}", source_addr);
 						if source_addr.is_ipv6() { return None; }
 						if &source_addr.ip().to_string() != ip { return None; }
-						info!("Valid broadcast");
 						Some(Ok((dac, source_addr)))
 					}
 				}
@@ -63,91 +64,247 @@ impl EtherdreamDevice {
 				Err(Box::new(LJError::EtherdreamConnectError(err)))
 			}
 			Ok((dac, source_addr)) => {
+				info!("Trying to open TCP stream...");
 				let stream = EtherdreamDevice::get_tcp_stream(&dac, &source_addr)?;
+				info!("Finished configuring DAC and TCP stream.");
 				Ok((dac, source_addr, stream))
 			}
 		}
 	}
 
-	pub fn get_tcp_stream(dac: &DacBroadcast, source_address: &SocketAddr) -> LJResult<Stream> {
-		// Establish the TCP connection.
+	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());
+		debug!("Playback State BEFORE PREPARE {:?}, {:?}", stream.dac().dac.status.playback,  Playback::Playing );
+		if stream.dac().dac.status.playback == Playback::Playing {
+			warn!("DAC was in playback PLAYING, attempting to stop");
+			match stream
+				.queue_commands()
+				.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 = 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;
 
-		// Prepare  stream
-		stream
+		let mut sine_wave = SineWave {
+			point: 0,
+			points_per_frame,
+			frames_per_second,
+		};
+		debug!("Stream dac BEFORE BEGIN {:?}", stream.dac());
+
+		match stream
 			.queue_commands()
-			.prepare_stream()
-			.submit()
-			.err()
-			.map(|err| {
-				eprintln!(
-					"err occurred when submitting PREPARE_STREAM \
-		                  command and listening for response: {}",
-					err
-				);
-			});
+			.data(sine_wave.by_ref().take(400))
+			// .data(begin_list.into_iter().take(400 as usize))
+			.begin(0, points_per_second)
+			.submit() {
+			Err(err) => warn!("err occurred when submitting first data: {}",err),
+			Ok(_) => info!("Sent first data to Etherdream.")
+		}
 
 		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(())
+	fn points_capacity(&self) -> usize {
+		/***
+		Determine the number of points needed to fill the DAC.
+		 ***/
+		let cap = self.dac.buffer_capacity as usize;
+		let fullness = self.stream.dac().dac.status.buffer_fullness as usize;
+		// 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
 	}
 
-	// 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
+	fn ping(&mut self) -> LJResult<()> {
+		Ok(self.stream.queue_commands().ping().submit()?)
 	}
 }
 
 impl Device for EtherdreamDevice {
 	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 playback_state = match self.stream.dac().dac.status.playback {
+			Playback::Idle => PlaybackState::IDLE,
+			Playback::Prepared => PlaybackState::PREPARE,
+			Playback::Playing => PlaybackState::PLAYING,
 		};
+		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: self.last_traced_at.clone(),
+			last_traced_at: now,
 			properties: vec!["foo".to_string()],
 			playback_state,
-			capacity: self.dac.dac_status.buffer_fullness,
-			lack: String::from(&self.lack),
+			capacity: self.points_capacity(),
+			lack: self.dac_response.to_string(),
 		}
+		// status
 	}
 
 	fn draw(&mut self,
 	        line: Vec<Point>,
 	        _speed: u32,
 	) -> LJResult<()> {
-		let n_points = self.points_to_generate();
-		self.stream
-			.queue_commands()
-			.data(line.into_iter().map(|point| point.into()).take(n_points))
-			.submit()?;
+		let chunk_size = 111;
+		let points_iter = line.into_iter();
+		for chunk in points_iter.as_slice().chunks(chunk_size) {
+			debug!("New chunk length: {:?}", chunk.len());
+			loop {
+				let capacity = self.points_capacity();
+				debug!("capacity : {:?}", capacity);
+				if chunk.len() > capacity as usize {
+					debug!("Sleep");
+					// Sleep for 1/100th of a sec
+					sleep(Duration::new(0, 10_000_000));
+					self.ping();
+				} else {
+					break;
+				}
+			}
+			debug!("Drawing");
+			match self.stream
+				.queue_commands()
+				.data(
+					chunk.into_iter()
+						.map(|point| (*point).into())
+						.take(chunk_size as usize)
+				)
+				.submit() {
+				Err(err) => {
+					// We should account for
+					// 'Broken pipe (os error 32)'
+					// Connection reset by peer (os error 104)
+					self.dac_response = match err {
+						CommunicationError::Io(err) => {
+							warn!("IO ERROR while drawing: '{}'",err);
+							DacResponse::ACK
+						}
+						CommunicationError::Protocol(err) => {
+							warn!("Protocol ERROR while drawing: '{}'",err);
+							DacResponse::ACK
+						}
+						CommunicationError::Response(err) => {
+							warn!("Response ERROR while drawing: '{}'",err);
+							err.response.response
+						}
+					};
+				}
+				Ok(_) => {
+					self.dac_response = DacResponse::ACK;
+					// debug!("Draw is ok");
+				}
+			};
+		}
 		Ok(())
 	}
 
 	fn stop(&mut self) -> LJResult<()> {
-		self.stream
+		info!("Stopping Etherdream device...");
+		info!("Stream dac{:?}", self.stream.dac());
+		match self.stream
 			.queue_commands()
 			.stop()
 			.submit()
-			.expect("err occurred when submitting STOP command and listening for response");
-		Ok(())
+		{
+			Err(err) => {
+				warn!("Failed to stop EtherDream device with error {:?}", err);
+				Err(Box::new(err))
+			}
+			Ok(_) => {
+				info!("Sucessfully closed EtherDream device.");
+				Ok(())
+			}
+		}
 	}
 
 	fn grid(&mut self) -> Vec<Point> {
-		vec!(
-			Point { x: 0.0, y: 0.0, color: Color { r: 255, g: 255, b: 255 } }
-		)
+		let dim_mid = 16000 as f32;
+		let dim_max = 32000 as f32;
+		let col_min = Color { r: 0, g: 0, b: 0 };
+		let col_max = Color { r: 255, g: 255, b: 255 };
+
+		vec![
+			Point { x: -dim_max, y: dim_max, color: col_min },
+			Point { x: -dim_max, y: dim_max, color: col_max },
+			Point { x: dim_max, y: dim_max, color: col_max },
+			Point { x: dim_max, y: -dim_max, color: col_max },
+			Point { x: -dim_max, y: -dim_max, color: col_max },
+			Point { x: -dim_max, y: -dim_mid, color: col_min },
+			Point { x: -dim_mid, y: dim_mid, color: col_min },
+			Point { x: -dim_mid, y: dim_mid, color: col_max },
+			Point { x: dim_mid, y: dim_mid, color: col_max },
+			Point { x: dim_mid, y: -dim_mid, color: col_max },
+			Point { x: -dim_mid, y: -dim_mid, color: col_max },
+			Point { x: -dim_mid, y: -dim_mid, color: col_min },
+		]
 	}
 }
+
+// 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,
+	points_per_frame: u16,
+	frames_per_second: f32,
+}
+
+impl Iterator for SineWave {
+	type Item = ether_dream::protocol::DacPoint;
+	fn next(&mut self) -> Option<Self::Item> {
+		let coloured_points_per_frame = self.points_per_frame - 1;
+		let i = (self.point % self.points_per_frame as u32) as u16;
+		let hz = 1.0;
+		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 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),
+		};
+		let x_min = std::i16::MIN;
+		let x_max = std::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;
+		let (u1, u2) = (0, 0);
+		let p = ether_dream::protocol::DacPoint {
+			control,
+			x,
+			y,
+			i,
+			r,
+			g,
+			b,
+			u1,
+			u2,
+		};
+		// debug!("{:?}",p);
+		self.point += 1;
+		Some(p)
+	}
+}

src/device/helios.rs

@@ -1,3 +1,4 @@
+use std::time::SystemTime;
 ///
 /// Configure udev:
 /// https://github.com/Grix/helios_dac/blob/master/docs/udev_rules_for_linux.md
@@ -14,7 +15,7 @@ use crate::conf::HeliosConf;
 use crate::device::{Device, Status, PlaybackState};
 use crate::errors::{LJError, LJResult};
 use crate::point::{Color, Point};
-use chrono::Utc;
+use chrono::{DateTime, Utc};
 
 pub struct HeliosDevice {
 	pub conf: HeliosConf,
@@ -34,13 +35,17 @@ impl HeliosDevice {
 			return Err(Box::new(LJError::HeliosDeviceMissing));
 		};
 		let dac = device.open()?;
+		let now = SystemTime::now();
+		let now: DateTime<Utc> = now.into();
+		let last_traced_at = now.to_rfc3339();
+
 		Ok(Self {
 			conf: (*conf).clone(),
 			dac,
 			sent_points: 0,
 			state: PlaybackState::PREPARE,
 			lack: "".to_string(),
-			last_traced_at: "1985-04-12T23:20:50.52Z".to_string(),
+			last_traced_at,
 		})
 	}
 }
@@ -52,7 +57,7 @@ impl Device for HeliosDevice {
 			last_traced_at: self.last_traced_at.clone(),
 			properties: vec!["foo".to_string()],
 			playback_state: self.state,
-			capacity: self.sent_points,
+			capacity: self.sent_points as usize,
 			lack,
 		}
 	}

src/framerate.rs

@@ -0,0 +1,39 @@
+use log::{debug, warn};
+use std::time::{Duration, Instant};
+use crate::errors::LJResult;
+use std::{thread};
+
+/// Converts helios Geometry to Helios
+#[derive(Debug, Clone, Copy)]
+pub struct Framerate {
+	prev_trace_time: Instant,
+	fps: u8,
+}
+
+impl Framerate {
+	pub fn new() -> LJResult<Self> {
+		Ok(Framerate {
+			prev_trace_time: Instant::now(),
+			fps: 20,
+		})
+	}
+	pub fn handle_time(&mut self) -> LJResult<()> {
+		let frame_time = 1000000000 / self.fps as u128;
+		let now = Instant::now();
+		// How long since last loop ?
+		let nanotime_spent = self.prev_trace_time.elapsed().as_nanos();
+		// Diw it go too fast? If so : sleep a bit
+		if frame_time > nanotime_spent {
+			let nanotime_towait = frame_time - nanotime_spent;
+			let dur = Duration::new(0, (nanotime_towait as f32 * 0.9) as u32);
+			// debug!("{:?} - {:?} : {:?}", nanotime_towait, self.prev_trace_time, now );
+			thread::sleep(dur);
+			debug!("Framerate OK");
+		} else {
+			warn!("Frame longer than expected {:?} > {:?}", nanotime_spent, frame_time, );
+		}
+
+		self.prev_trace_time = now;
+		Ok(())
+	}
+}

src/lib.rs

@@ -5,3 +5,4 @@ pub mod device;
 pub mod point;
 pub mod transformer;
 pub mod worldstate;
+pub mod framerate;

src/main.rs

@@ -9,6 +9,7 @@ mod point;
 mod transformer;
 mod device;
 mod worldstate;
+mod framerate;
 
 use device::device_factory;
 use std::sync::atomic::{AtomicBool, Ordering};
@@ -21,6 +22,7 @@ use transformer::Transformers;
 use log::{LevelFilter, info, /* warn, */ error};
 use env_logger::Builder;
 use worldstate::WorldState;
+use framerate::Framerate;
 
 const DEFAULT_CONF_FILE: &str = "settings.toml";
 
@@ -35,6 +37,13 @@ pub fn main() {
 
 
 fn run_all() -> LJResult<()> {
+	// Setup handler for interrupt Signals
+	let running = Arc::new(AtomicBool::new(true));
+	let r = running.clone();
+	ctrlc::set_handler(move || {
+		r.store(false, Ordering::SeqCst);
+	})?;
+
 	// Setup configuration file and set up logs
 	let filename = std::env::args().nth(1).unwrap_or_else(|| {
 		DEFAULT_CONF_FILE.to_string()
@@ -53,30 +62,22 @@ fn run_all() -> LJResult<()> {
 	let mut world_state = rs.init_world_state().unwrap();
 	info!("WorldState: {:?}", world_state);
 
-	// Setup handler for interrupt Signals
-	let running = Arc::new(AtomicBool::new(true));
-	let r = running.clone();
-	ctrlc::set_handler(move || {
-		r.store(false, Ordering::SeqCst);
-	})?;
 
 	// Setup Laser Device based on conf
 	let mut tracer = device_factory(&config)?;
 	world_state.grid = tracer.grid();
 
-	// can't work, but we can add + Debug to Device to make it work...
-	//dbg!(tracer);
-
 	// Setup geometry transformers on points lists
 	let transformers = config.get_transformers();
 
+	// Setup framerate limiter
+	let mut framerate_handler = Framerate::new()?;
+
 	// Dispatch based on redis requests
 	while running.load(Ordering::SeqCst) {
 		rs.set_status(tracer.status())?;
-
+		// let _ = framerate_handler.handle_time()?;
 		let order = rs.get_order(config.laser_id)?;
-
-
 		match order {
 			Order::Draw | Order::Black | Order::Grid => {
 				// 0 : Draw Normal point list
@@ -108,12 +109,15 @@ fn run_all() -> LJResult<()> {
 			}
 			Order::ClientKey => {
 				world_state.client_key = rs.get_client_key()?;
-			},
-			// Order::ColorBalance => {},
+			}
+			Order::ColorBalance => {
+				let (r, g, b) = rs.get_color_balance()?;
+				world_state.color_balance = Color { r, g, b };
+			}
+			Order::Resampler => {
+				world_state.resampler = rs.get_resampler()?;
+			}
 			_ => {
-				// 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
 				info!("Order: {:?}", order);
 			}
@@ -127,14 +131,17 @@ fn run_all() -> LJResult<()> {
 
 fn init_logging(config: &LJResult<Conf>) {
 	if let Ok(ref config) = config {
-		if config.debug {
-			let mut builder = Builder::from_default_env();
-			builder
-				.filter(None, LevelFilter::Info)
-				.init();
-			info!("Debug mode enabled from configuration file");
-			return;
-		}
+		let level = if config.debug {
+			LevelFilter::Debug
+		} else {
+			LevelFilter::Info
+		};
+		let mut builder = Builder::from_default_env();
+		builder
+			.filter(None, level)
+			.init();
+		info!("Debug mode enabled from configuration file");
+		return;
 	}
 	info!("Logging level inherited from env");
 	env_logger::init();
@@ -165,8 +172,8 @@ fn get_next_frame(
 		line = transformer.apply(&line, world_state);
 	}
 
-	info!("Draw Black -> {}", world_state.draw_black);
-	info!("Draw Grid -> {}", world_state.draw_grid);
+	// info!("Draw Black -> {}", world_state.draw_black);
+	// info!("Draw Grid -> {}", world_state.draw_grid);
 
 	// LIMITER and BLACK
 	line = line.into_iter()

src/point.rs

@@ -1,4 +1,6 @@
 use ether_dream::protocol::DacPoint;
+use std::ops::Mul;
+
 
 #[derive(Debug, Clone, Copy, Default, PartialEq)]
 pub struct Point {
@@ -14,6 +16,18 @@ pub struct Color {
 	pub b: u8,
 }
 
+impl Mul<u8> for Color {
+    type Output = Self;
+
+    fn mul(self, rhs: u8) -> Self {
+	    Self{
+		    r: (255 * self.r as u16 / rhs as u16) as u8,
+		    g: (255 * self.g as u16 / rhs as u16) as u8,
+		    b: (255 * self.b as u16 / rhs as u16) as u8,
+	    }
+    }
+}
+
 impl From<Color> for u32 {
 	fn from(value: Color) -> Self {
 		let r = value.r as u32;
@@ -48,17 +62,21 @@ impl From<Point> for DacPoint {
 	fn from(pt: Point) -> DacPoint {
 		let control = 0;
 		let (u1, u2) = (0, 0);
-		let i = 255;
-		DacPoint {
+		let i = 0;
+		let x = pt.x.clamp(-32000.0, 32000.0);
+		let y = pt.y.clamp(-32000.0, 32000.0);
+		let pt = DacPoint {
 			control,
-			x: pt.x as i16,
-			y: pt.y as i16,
+			x: x as i16,
+			y: y as i16,
 			i,
-			r: pt.color.r.into(),
-			g: pt.color.g.into(),
-			b: pt.color.b.into(),
+			r: (pt.color.r as u16) * 255,
+			g: (pt.color.g as u16) * 255,
+			b: (pt.color.b as u16) * 255,
 			u1,
 			u2,
-		}
+		};
+		// debug!("point {:?}", pt);
+		pt
 	}
 }

src/redis_ctrl.rs

@@ -2,7 +2,7 @@ use redis::{Client, Commands, Connection};
 use ron::de::from_str;
 use crate::device::Status;
 use crate::errors::{LJError, LJResult};
-use crate::worldstate::{WorldState,EDH};
+use crate::worldstate::{WorldState, EDH};
 // use log::info;
 
 #[repr(u8)]
@@ -10,7 +10,6 @@ use crate::worldstate::{WorldState,EDH};
 pub enum Order {
 	Draw = 0,
 	Edh,
-	//homography
 	Black,
 	Grid,
 	Resampler,
@@ -18,6 +17,7 @@ pub enum Order {
 	Intensity,
 	Kpps,
 	ColorBalance,
+	PowerOff
 }
 
 impl TryFrom<u8> for Order {
@@ -40,12 +40,14 @@ impl TryFrom<u8> for Order {
 			6 => Intensity,
 			7 => Kpps,
 			8 => ColorBalance,
+			9 => PowerOff,
 			_ => unreachable!()
 		})
 	}
 }
 
 pub type Line = Vec<(f32, f32, u32)>;
+pub type Resampler = Vec<Vec<(f32,f32)>>;
 
 pub struct RedisCtrl {
 	pub client: Client,
@@ -84,11 +86,6 @@ impl RedisCtrl {
 		Ok(val.try_into()?)
 	}
 
-	/**
-	/lstt/lasernumber       etherdream last_status.playback_state  (0: idle   1: prepare   2: playing)
-	/cap/lasernumber        number of empty points sent to fill etherdream buffer (up to 1799)
-	/lack/lasernumber       "a": ACK   "F": Full  "I": invalid. 64 or 35 for no connection.
-	 **/
 	pub fn set_status(&mut self, status: Status) -> LJResult<()> {
 		let lstt_key = format!("/lstt/{}", self.laser_id);
 		let cap_key = format!("/cap/{}", self.laser_id);
@@ -99,34 +96,48 @@ impl RedisCtrl {
 		Ok(())
 	}
 
-	pub fn init_world_state( &mut self) -> LJResult<WorldState>{
-	    Ok(WorldState {
-		client_key: self.get_client_key().unwrap(),
-		edh: self.get_edh().unwrap(),
-		kpps: self.get_int("kpps").unwrap().try_into().unwrap(),
-		intensity: self.get_int("intensity").unwrap().try_into().unwrap(),
-		..WorldState::default()
-	    })
+	pub fn init_world_state(&mut self) -> LJResult<WorldState> {
+		Ok(WorldState {
+			client_key: self.get_client_key().unwrap(),
+			edh: self.get_edh().unwrap(),
+			kpps: self.get_int("kpps").unwrap().try_into().unwrap(),
+			intensity: self.get_int("intensity").unwrap().try_into().unwrap(),
+			..WorldState::default()
+		})
 	}
 
-	pub fn get_edh( &mut self ) -> LJResult<EDH> {
-	    // Get new EDH
-	    let edh_key = format!("/EDH/{}", self.laser_id);
-	    let edh : String = self.connection.get(edh_key)?;
-	    let edh : Vec<Vec<f32>> = from_str(&edh)?;
-	    let edh = EDH::new(edh)?;
-	    Ok(edh)
+	pub fn get_edh(&mut self) -> LJResult<EDH> {
+		// Get new EDH
+		let edh_key = format!("/EDH/{}", self.laser_id);
+		let edh: String = self.connection.get(edh_key)?;
+		let edh: Vec<Vec<f32>> = from_str(&edh)?;
+		let edh = EDH::new(edh)?;
+		Ok(edh)
 	}
 
-	pub fn get_client_key( &mut self ) -> LJResult<String> {
-	    let key : String = self.connection.get("/clientkey")?;
-	    Ok(key)
+	pub fn get_client_key(&mut self) -> LJResult<String> {
+		let key: String = self.connection.get("/clientkey")?;
+		Ok(key)
 	}
 
-	pub fn get_int(&mut self, key: &str ) -> LJResult<u32> {
-	    // Get new Int
-	    let fmt = format!("/{key}/{}", self.laser_id);
-	    let val : u32 = self.connection.get(fmt)?;
-	    Ok(val)
+	pub fn get_color_balance(&mut self) -> LJResult<(u8, u8, u8)> {
+		Ok((
+			self.connection.get("/red")?,
+			self.connection.get("/green")?,
+			self.connection.get("/blue")?,
+		))
+	}
+
+	pub fn get_resampler(&mut self ) -> LJResult<Resampler> {
+		let val: String = self.connection.get(format!("/resampler/{}", self.laser_id))?;
+		let resampler : Resampler = from_str(&val)?;
+		Ok(resampler)
+	}
+
+	pub fn get_int(&mut self, key: &str) -> LJResult<u32> {
+		// Get new Int
+		let fmt = format!("/{key}/{}", self.laser_id);
+		let val: u32 = self.connection.get(fmt)?;
+		Ok(val)
 	}
 }

src/transformer.rs

@@ -1,11 +1,12 @@
-//pub mod common;
-mod translate;
-mod replicate;
-mod rotate;
 mod flip_horizontal;
 mod flip_vertical;
 mod grid;
+mod helios_to_etherdream;
 mod homography;
+mod intensity;
+mod replicate;
+mod rotate;
+mod translate;
 
 use crate::point::Point;
 use crate::worldstate::WorldState;
@@ -18,6 +19,8 @@ pub use flip_horizontal::FlipHorizontal;
 pub use flip_vertical::FlipVertical;
 pub use grid::Grid;
 pub use self::homography::Homography;
+pub use helios_to_etherdream::HeliosToEtherdream;
+pub use intensity::Intensity;
 
 pub trait Transformers {
     fn apply(

src/transformer/helios_to_etherdream.rs

@@ -0,0 +1,26 @@
+use crate::transformer::Transformers;
+use crate::point::Point;
+use crate::worldstate::WorldState;
+
+use serde::{Serialize, Deserialize};
+
+/// Converts helios Geometry to Helios
+#[allow(dead_code)]
+#[derive(Serialize, Deserialize, Debug, Clone, Copy)]
+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),
+				y: 40.0 * (240.0 - pt.y),
+				..*pt
+			}
+		}).collect();
+		// debug!("list etherdream {:?}", out);
+		out
+	}
+}
+

src/transformer/intensity.rs

@@ -0,0 +1,26 @@
+use log::debug;
+use crate::transformer::Transformers;
+use crate::point::Point;
+use crate::worldstate::WorldState;
+
+use serde::{Serialize, Deserialize};
+
+/// Converts helios Geometry to Helios
+#[allow(dead_code)]
+#[derive(Serialize, Deserialize, Debug, Clone, Copy)]
+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,
+				..*pt
+		}
+	}).collect();
+		// debug!("list intensity {:?}", out);
+		out
+	}
+}
+

src/worldstate.rs

@@ -2,6 +2,7 @@ use crate::point::{Point, Color};
 use nalgebra::base::{Matrix3, Matrix1x3};
 use crate::errors::{LJError, LJResult};
 use log::debug;
+use crate::redis_ctrl::Resampler;
 
 #[derive(Debug, Default)]
 pub struct EDH {
@@ -39,7 +40,7 @@ impl EDH {
 #[derive(Debug, Default)]
 pub struct WorldState {
 	pub edh: EDH,
-	pub resampler: Vec<f32>,
+	pub resampler: Resampler,
 	pub client_key: String,
 	pub intensity: u8,
 	pub kpps: u32,
@@ -47,6 +48,7 @@ pub struct WorldState {
 	pub draw_black: bool,
 	pub draw_grid: bool,
 	pub grid: Vec<Point>,
+	pub color_balance: Color,
 }
 
 impl WorldState {}