floorbell/src/main.rs

use std::cell::{Cell, RefCell};
use std::convert::Infallible;
use std::process;
use std::sync::{Arc, Mutex};
use std::time::Duration;

use futures_util::{FutureExt, select, stream, StreamExt};
#[allow(unused_imports)] use log::{debug, error, info, trace};
use rppal::gpio::{Gpio, Trigger};
use smallvec::SmallVec;
use tokio::{ sync::mpsc,
             time::{delay_for, interval} };
use warp::{Filter, sse, sse::ServerSentEvent};

const HUE_ADDRESS: &str = "philips-hue.local";
const OUTDOOR_CHIME_FILE: &str = "static/outdoor.mp3";

const HUE_KEY: &str = include_str!("../hue.key");
const BUTTON_PIN: u8 = 26;
const CHANNEL_VEC_SIZE: usize = 32;

#[tokio::main(basic_scheduler)]
async fn main() {
  pretty_env_logger::init_timed();

  let clients = Arc::new(Mutex::new(SmallVec::new()));
  let events_clients = clients.clone();

  let gpio = Gpio::new().expect("gpio init");
  let mut pin = gpio.get(BUTTON_PIN).expect("pin init").into_input_pullup();

  pin.set_async_interrupt(Trigger::FallingEdge, move |_| button_pressed(&clients))
     .expect("set interrupt");

  let (tx, rx) = mpsc::channel(1);
  events_clients.lock().unwrap().push(tx);

  let reqwest = &reqwest::Client::new();
  let audio_child = &RefCell::new(None);
  let hue_busy = &Cell::new(false);
  let pushes = rx.for_each_concurrent(2, |()| async move {
    if !audio_busy(&audio_child) {
      play_chime(&audio_child)
    } else { debug!("doorbell still ringing, not playing new chime"); }

    if !hue_busy.replace(true) {
      flash_porch(&reqwest).await;
      hue_busy.set(false);
    } else { debug!("hue already in use, not scheduling new flashing"); }
  });

  select! {
    _ = warp(events_clients).fuse() => (),
    _ = pushes.fuse() => ()
  }
}

fn button_pressed(clients: &Arc<Mutex<SmallVec<[mpsc::Sender<()>; CHANNEL_VEC_SIZE]>>>) {
  info!("DOORBELL PRESS");
  clients.lock().unwrap().retain(|tx: &mut mpsc::Sender<()>| {
    match tx.try_send(()) {
      Ok(_) => true,
      Err(mpsc::error::TrySendError::Full(_))   => true, // we just get some free debouncing
      Err(mpsc::error::TrySendError::Closed(_)) => { info!("Event client disconnected"); false }
    }
  });
}

async fn warp(clients: Arc<Mutex<SmallVec<[mpsc::Sender<()>; CHANNEL_VEC_SIZE]>>>) {
  // GET /
  let root = warp::path::end()
    .and(warp::get())
    .map(|| info!("GET /")).untuple_one()
    .and(warp::fs::file("./static/main.html"));

  // GET /events
  let events = warp::path!("events")
    .and(warp::get())
    .map(move || {
      info!("GET /events");
      let (tx, rx) = mpsc::channel(1);
      clients.lock().unwrap().push(tx);

      let stream = stream::select(
        interval(Duration::from_secs(3)).map(move |_| {
          trace!("sending sse keepalive ping");
          Ok::<_, Infallible>((sse::event("ping"), sse::data("")).into_a())
        }),

        rx.map(|()| {
          debug!("sending ring sse");
          Ok::<_, Infallible>((sse::event("ring"), sse::data("")).into_b())
        })
      );
      sse::reply(stream)
    });

  // GET /*
  let statics = warp::get()
    .and(warp::path::peek())
    .map(|path| info!("GET /* : {:?}", path)).untuple_one()
    .and(warp::fs::dir("static"));

  let routes = root.or(events).or(statics);
  warp::serve(routes).run(([0, 0, 0, 0], 8060)).await;
}

fn audio_busy(audio_child: &RefCell<Option<process::Child>>) -> bool {
  if let Some(ref mut audio_child) = *audio_child.borrow_mut() {
    if let Ok(None) = audio_child.try_wait() {
      return true;
    }
  }
  false
}

fn play_chime(audio_child: &RefCell<Option<process::Child>>) {
  trace!("Playing doorbell chime");
  match process::Command::new("mplayer")
                         .arg(OUTDOOR_CHIME_FILE)
                         .stdout(process::Stdio::null())
                         .stderr(process::Stdio::null())
                         .spawn() {
    Ok(child) => { audio_child.replace(Some(child)); },
    Err(err) => error!("Error playing outdoor chime: {}", err)
  };
}

async fn hue_base(reqwest: &reqwest::Client, body: &'static str, delay_millis: u64) {
  let _ = reqwest.put(&format!("http://{}/api/{}/lights/10/state", HUE_ADDRESS, HUE_KEY))
                 .body(body).send().await;
  delay_for(Duration::from_millis(delay_millis)).await;
}

async fn flash_porch(reqwest: &reqwest::Client) {
  hue_base(reqwest,
           r#"{"transitiontime":0,"on":true,"bri":254,"sat":254}"#,  // activate/resaturate
           250).await;
  for _ in 0..2 {
    hue_base(reqwest,
             r#"{"transitiontime":0,"hue":2125}"#,  // orange
             250).await;
    hue_base(reqwest,
             r#"{"transitiontime":0,"hue":25500}"#,  // green
             250).await;
    hue_base(reqwest,
             r#"{"transitiontime":0,"hue":56228}"#,  // purple
             250).await;
  }
  delay_for(Duration::from_millis(250)).await;
  hue_base(reqwest,
           r#"{"transitiontime":20,"bri":20}"#,  // fade
           2_250).await;
  hue_base(reqwest,
           r#"{"transitiontime":1,"bri":20,"sat":0}"#,  // desaturate
           250).await;
  hue_base(reqwest,
           r#"{"transitiontime":40,"bri":180,"sat":0}"#,  // fade up
           4_000).await;
}