lj_rust/src/device/etherdream.rs

use ether_dream::dac::stream::connect;
use ether_dream::dac::Stream;
use std::net::SocketAddr;
use std::time;

use crate::conf::EtherDreamConf;
use crate::device::{Device, PlaybackState, Status};
use crate::errors::{LJError, LJResult};
use crate::point::{Color, Point};
use ether_dream::protocol::{DacBroadcast, DacStatus};
use log::{info, warn};

#[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,
}

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)?;
        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(),
        })
    }
    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_timeout(Some(time::Duration::new(10, 0)))?;
        info!("Attempting to get DAC broadcast...");
        let broadcast = dac_broadcast
            .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)) => {
                    if source_addr.is_ipv6() {
                        return None;
                    }
                    if &source_addr.ip().to_string() != ip {
                        return None;
                    }
                    info!("Valid broadcast");
                    Some(Ok((dac, source_addr)))
                }
            })
            .next()
            .expect("Failed to receive broadcast.");
        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(&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,
        };

        Status {
            last_traced_at: self.last_traced_at.clone(),
            properties: vec!["foo".to_string()],
            playback_state,
            capacity: self.dac.dac_status.buffer_fullness,
            lack: String::from(&self.lack),
        }
    }

    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()?;
        Ok(())
    }

    fn stop(&mut self) -> LJResult<()> {
        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,
            },
        }]
    }
}