diff --git a/drivers/drmem-drv-hue/README.md b/drivers/drmem-drv-hue/README.md index ffb65b5..e9eeb08 100644 --- a/drivers/drmem-drv-hue/README.md +++ b/drivers/drmem-drv-hue/README.md @@ -11,9 +11,9 @@ The driver uses a **polling-based** architecture rather than event streaming: ## Supported Device Types -- **Switch** - On/off control with indicator -- **Dimmer/Bulb** - On/off and brightness control (0-100%) with indicator -- **ColorBulb** - Full RGB color control and brightness +- **Switch** - On/off control +- **Dimmer/Bulb** - Brightness control (0-100%) +- **ColorBulb** - Full color control, with brightness conveyed by the color's alpha channel - **Group** - Controls Hue groups/rooms (uses ColorBulb interface) ## Configuration @@ -93,9 +93,9 @@ You need to create a Hue application key before using this driver. Follow the of ### State Updates -- Brightness values are clamped to 0-100 range -- A brightness of 0 turns the device off -- For color bulbs, colors are converted between RGB and CIE XY color space +- Dimmer/Bulb brightness values are clamped to 0-100 range; a brightness of 0 turns the device off +- For color bulbs, brightness is conveyed by the color's alpha channel: 0 is off, 255 is 100% +- For color bulbs, colors are converted between RGB (or color temperature) and CIE XY color space - Missing brightness values when a device is "on" default to 100% ### Override Timeout @@ -120,7 +120,7 @@ This allows temporary manual control without permanently losing DrMem's automati ## Color Conversion For ColorBulb devices, the driver automatically converts between: -- RGB (0-255 per channel) used by DrMem -- CIE XY color space used by Hue devices +- RGB (0-255 per channel), or a Kelvin color temperature, used by DrMem +- CIE XY color space, plus separate on/off and brightness settings, used by Hue devices -This conversion uses the `palette` crate for accurate color representation. \ No newline at end of file +A ColorBulb's single `color` device conveys all three bridge parameters: the RGB (or Kelvin) value maps to the bridge's on/off and XY settings, while the color's alpha channel (0-255) maps to the bridge's brightness (0-100%). This conversion uses the `palette` crate for accurate color representation. \ No newline at end of file diff --git a/drivers/drmem-drv-hue/src/device.rs b/drivers/drmem-drv-hue/src/device.rs index 99398da..3b95a68 100644 --- a/drivers/drmem-drv-hue/src/device.rs +++ b/drivers/drmem-drv-hue/src/device.rs @@ -2,7 +2,7 @@ use super::{config, driver::device_traits}; use drmem_api::{ Result, device::Path, - driver::{Registrator, Reporter, ResettableState, classes}, + driver::{OverrideConfig, Registrator, Reporter, ResettableState, classes}, }; use std::{collections::HashMap, sync::Arc}; use tokio::time::Duration; @@ -21,39 +21,42 @@ impl Set { cfg: &config::DeviceConfig, max_history: Option, ) -> Result<(Arc, DeviceWrapper)> { - let tmo = cfg.override_timeout.map(|v| Duration::from_secs(v)); + let override_cfg = OverrideConfig { + override_duration: cfg.override_timeout.map(Duration::from_secs), + envelope: Some(Duration::from_secs(30)), + }; Ok(( cfg.id.clone(), match cfg.r#type { config::DevCfgType::Switch => { - DeviceWrapper::Switch(device_traits::SwitchDevice { - inner: classes::Switch::register_devices( + DeviceWrapper::Switch(device_traits::SwitchDevice( + classes::Switch::register_devices( drc, Some(cfg.subpath.as_ref()), - &tmo, + &override_cfg, max_history, ) .await?, - }) + )) } config::DevCfgType::Dimmer | config::DevCfgType::Bulb => { - DeviceWrapper::Dimmer(device_traits::DimmerDevice { - inner: classes::Dimmer::register_devices( + DeviceWrapper::Dimmer( + device_traits::DimmerDevice::register_devices( drc, Some(cfg.subpath.as_ref()), - &tmo, + &override_cfg, max_history, ) .await?, - }) + ) } config::DevCfgType::ColorBulb => DeviceWrapper::ColorBulb( device_traits::ColorBulbDevice::new( classes::ColorBulb::register_devices( drc, Some(cfg.subpath.as_ref()), - &tmo, + &override_cfg, max_history, ) .await?, @@ -65,7 +68,7 @@ impl Set { classes::ColorBulb::register_devices( drc, Some(cfg.subpath.as_ref()), - &tmo, + &override_cfg, max_history, ) .await?, diff --git a/drivers/drmem-drv-hue/src/driver/color.rs b/drivers/drmem-drv-hue/src/driver/color.rs index f296afc..6a3ce5f 100644 --- a/drivers/drmem-drv-hue/src/driver/color.rs +++ b/drivers/drmem-drv-hue/src/driver/color.rs @@ -1,4 +1,5 @@ /// Color conversion utilities for Hue CIE XY color space +use drmem_api::device::ColorType; use palette::{IntoColor, LinSrgb, LinSrgba, Yxy}; /// Convert Hue CIE XY coordinates to RGBA @@ -25,6 +26,109 @@ pub fn rgba_to_cie_xy(rgba: &LinSrgba) -> (f32, f32) { (yxy.x, yxy.y) } +/// Approximates a color temperature (in Kelvin) as CIE xy coordinates +/// using the Kim et al. polynomial fit to the Planckian locus. Valid +/// over the range of colors Hue bulbs can produce (roughly +/// 1667K-25000K). +pub fn kelvin_to_cie_xy(kelvin: u16) -> (f32, f32) { + let t = kelvin as f64; + let (t2, t3) = (t * t, t * t * t); + + let x = if t <= 4000.0 { + -0.2661239e9 / t3 - 0.2343589e6 / t2 + 0.8776956e3 / t + 0.179910 + } else { + -3.0258469e9 / t3 + 2.1070379e6 / t2 + 0.2226347e3 / t + 0.24039 + }; + let (x2, x3) = (x * x, x * x * x); + + let y = if t <= 2222.0 { + -1.1063814 * x3 - 1.34811020 * x2 + 2.18555832 * x - 0.20219683 + } else if t <= 4000.0 { + -0.9549476 * x3 - 1.37418593 * x2 + 2.09137015 * x - 0.16748867 + } else { + 3.0817580 * x3 - 5.87338670 * x2 + 3.75112997 * x - 0.37001483 + }; + + (x as f32, y as f32) +} + +/// The three parameters the Hue bridge needs to represent a light's +/// setting: whether it's on, its brightness (0.0-100.0), and its +/// chromaticity. +#[derive(Debug, Clone, Copy, PartialEq)] +pub struct BridgeState { + pub on: bool, + pub brightness: f32, + pub xy: (f32, f32), +} + +/// Breaks a `ColorType` down into the on/off, brightness, and XY +/// coordinates the Hue bridge needs. Brightness is conveyed by the +/// color's alpha channel: 0 is off and 255 is full brightness. +pub fn color_to_bridge(color: &ColorType) -> BridgeState { + let (alpha, xy) = match color { + ColorType::Rgba { color } => (color.alpha, rgba_to_cie_xy(color)), + ColorType::Ccta { kelvin, a } => (*a, kelvin_to_cie_xy(*kelvin)), + }; + + BridgeState { + on: alpha > 0, + brightness: alpha as f32 / 255.0 * 100.0, + xy, + } +} + +/// Merges the (possibly partial) on/off, brightness, and XY state +/// reported by the bridge into a new `ColorType`, using `prev` to fill +/// in any component that wasn't reported. The bridge only ever echoes +/// XY coordinates, so a reported `xy` always yields a `ColorType::Rgba`. +/// But when `xy` is `None` (the caller filters out echoes that just +/// match what was last sent), `prev`'s representation -- `Rgba` or +/// `Ccta` -- is preserved. This matters because a `Ccta` setting can +/// never compare equal to the `Rgba` the bridge would otherwise force +/// it into, which would make `OverridableDevice` treat every polled +/// reading as a mismatch and re-send the setting forever. +pub fn merge_bridge_update( + prev: &ColorType, + on: Option, + brightness: Option, + xy: Option<(f32, f32)>, +) -> ColorType { + let prev_alpha = match prev { + ColorType::Rgba { color } => color.alpha, + ColorType::Ccta { a, .. } => *a, + }; + let is_on = on.unwrap_or(prev_alpha > 0); + + let alpha = if !is_on { + 0 + } else if let Some(brightness) = brightness { + (brightness.clamp(0.0, 100.0) / 100.0 * 255.0).round() as u8 + } else if on == Some(true) { + // Just turned on with no explicit brightness: default to 100%. + 255 + } else { + prev_alpha + }; + + match (xy, prev) { + (None, ColorType::Ccta { kelvin, .. }) => ColorType::Ccta { + kelvin: *kelvin, + a: alpha, + }, + (None, ColorType::Rgba { color }) => ColorType::Rgba { + color: LinSrgba::new(color.red, color.green, color.blue, alpha), + }, + (Some((x, y)), _) => { + let rgb = cie_xy_to_rgba(x, y); + + ColorType::Rgba { + color: LinSrgba::new(rgb.red, rgb.green, rgb.blue, alpha), + } + } + } +} + #[cfg(test)] mod tests { use super::*; @@ -86,4 +190,131 @@ mod tests { assert_eq!(converted.alpha, 255); } } + + #[test] + fn test_color_to_bridge_alpha_is_brightness() { + let rgb = LinSrgba::new(255, 128, 64, 0); + + // alpha == 0 means off, and brightness reads as 0%. + let off = color_to_bridge(&ColorType::Rgba { color: rgb }); + assert!(!off.on); + assert_eq!(off.brightness, 0.0); + + // alpha == 255 means fully on, brightness reads as 100%. + let full = color_to_bridge(&ColorType::Rgba { + color: LinSrgba::new(255, 128, 64, 255), + }); + assert!(full.on); + assert_eq!(full.brightness, 100.0); + + // A mid-range alpha is on, with a proportional brightness. + let half = color_to_bridge(&ColorType::Rgba { + color: LinSrgba::new(255, 128, 64, 128), + }); + assert!(half.on); + assert!((half.brightness - 50.196).abs() < 0.01); + + // XY is derived only from RGB, independent of alpha. + assert_eq!(off.xy, full.xy); + assert_eq!(off.xy, half.xy); + } + + #[test] + fn test_color_to_bridge_ccta() { + let bright = color_to_bridge(&ColorType::Ccta { + kelvin: 2700, + a: 255, + }); + assert!(bright.on); + assert_eq!(bright.brightness, 100.0); + + let off = color_to_bridge(&ColorType::Ccta { kelvin: 2700, a: 0 }); + assert!(!off.on); + + // xy should match a direct kelvin conversion. + assert_eq!(bright.xy, kelvin_to_cie_xy(2700)); + } + + #[test] + fn test_kelvin_to_cie_xy_known_white_points() { + // D65 (~6500K) and a warm-white (~2700K) sanity check, using a + // generous tolerance for the polynomial approximation. + let (x, y) = kelvin_to_cie_xy(6500); + assert!((x - 0.3127).abs() < 0.01, "x={x}"); + assert!((y - 0.3290).abs() < 0.02, "y={y}"); + + let (x, y) = kelvin_to_cie_xy(2700); + assert!((x - 0.4578).abs() < 0.01, "x={x}"); + assert!((y - 0.4101).abs() < 0.01, "y={y}"); + } + + #[test] + fn test_merge_bridge_update_on_off_only() { + let prev = ColorType::Rgba { + color: LinSrgba::new(200, 100, 50, 255), + }; + + // Turning off keeps the RGB hue but zeroes the alpha. + let merged = merge_bridge_update(&prev, Some(false), None, None); + assert_eq!( + merged, + ColorType::Rgba { + color: LinSrgba::new(200, 100, 50, 0) + } + ); + } + + #[test] + fn test_merge_bridge_update_dimming_only() { + let prev = ColorType::Rgba { + color: LinSrgba::new(200, 100, 50, 255), + }; + + // A brightness-only update preserves the RGB hue. + let merged = merge_bridge_update(&prev, None, Some(50.0), None); + assert_eq!( + merged, + ColorType::Rgba { + color: LinSrgba::new(200, 100, 50, 128) + } + ); + } + + #[test] + fn test_merge_bridge_update_xy_only() { + let prev = ColorType::Rgba { + color: LinSrgba::new(200, 100, 50, 128), + }; + let (x, y) = rgba_to_cie_xy(&LinSrgba::new(10, 20, 30, 255)); + + // An XY-only update changes the RGB hue but keeps the alpha. + let merged = merge_bridge_update(&prev, None, None, Some((x, y))); + assert_eq!( + merged, + ColorType::Rgba { + color: LinSrgba::new( + cie_xy_to_rgba(x, y).red, + cie_xy_to_rgba(x, y).green, + cie_xy_to_rgba(x, y).blue, + 128 + ) + } + ); + } + + #[test] + fn test_merge_bridge_update_turned_on_defaults_to_full_brightness() { + let prev = ColorType::Rgba { + color: LinSrgba::new(200, 100, 50, 0), + }; + + // Turning on with no explicit brightness defaults to 100%. + let merged = merge_bridge_update(&prev, Some(true), None, None); + assert_eq!( + merged, + ColorType::Rgba { + color: LinSrgba::new(200, 100, 50, 255) + } + ); + } } diff --git a/drivers/drmem-drv-hue/src/driver/device_traits.rs b/drivers/drmem-drv-hue/src/driver/device_traits.rs index 7b81da9..0b0867e 100644 --- a/drivers/drmem-drv-hue/src/driver/device_traits.rs +++ b/drivers/drmem-drv-hue/src/driver/device_traits.rs @@ -1,6 +1,14 @@ /// Trait-based device handling for Hue devices -use super::{constants, payload}; -use drmem_api::driver::{Reporter, ResettableState, classes}; +use super::{color, constants, payload}; +use drmem_api::{ + Result, + device::{ColorType, Path}, + driver::{ + OverrideConfig, Registrator, Reporter, RequestChan, ResettableState, + classes, + }, +}; +use palette::LinSrgba; use tracing::debug; /// Common interface for all Hue device types @@ -9,24 +17,17 @@ pub trait HueDevice { fn resource_type(&self) -> &'static str; /// Wait for the next setting change and return a command if one is ready - fn next_setting( - &mut self, - ) -> impl std::future::Future> + Send; + async fn next_setting(&mut self) -> Option; /// Apply an update from the bridge to the device - fn apply_update( - &mut self, - update: &payload::ResourceData, - ) -> impl std::future::Future + Send; + async fn apply_update(&mut self, update: &payload::ResourceData) -> (); /// Reset the device state (called when driver restarts) fn reset(&mut self); } /// Wrapper for Switch devices -pub struct SwitchDevice { - pub inner: classes::Switch, -} +pub struct SwitchDevice(pub classes::Switch); impl HueDevice for SwitchDevice { fn resource_type(&self) -> &'static str { @@ -34,49 +35,56 @@ impl HueDevice for SwitchDevice { } async fn next_setting(&mut self) -> Option { - loop { - tokio::select! { - // Check state device - opt_txn = self.inner.state.next_setting() => { - if let Some((val, reply)) = opt_txn { - debug!("switch state setting ready: {}", val); - if let Some(r) = reply { - r.ok(val); - } - return Some(payload::LightCommand { - on: Some(payload::On { on: val }), - dimming: None, - color: None, - }); - } - } - - // Check indicator device (drain but don't send command) - opt_txn = self.inner.indicator.next_setting() => { - if let Some((val, Some(r))) = opt_txn { - r.ok(val); - } - } + self.0.next_setting().await.map(|val| { + debug!("switch state setting ready: {}", val.state); + payload::LightCommand { + on: Some(payload::On { on: val.state }), + dimming: None, + color: None, } - } + }) } async fn apply_update(&mut self, update: &payload::ResourceData) { if let Some(on) = &update.on { debug!("switch: reporting state update: {}", on.on); - self.inner.state.report_update(on.on).await; + self.0 + .report_update(classes::SwitchProperty { state: on.on }) + .await; } } fn reset(&mut self) { - self.inner.state.reset_state(); - self.inner.indicator.reset_state(); + self.0.reset_state() } } /// Wrapper for Dimmer/Bulb devices pub struct DimmerDevice { - pub inner: classes::Dimmer, + pub dimmer: classes::Dimmer, +} + +impl ResettableState for DimmerDevice {} + +impl Registrator for DimmerDevice { + type Config = OverrideConfig; + + async fn register_devices( + drc: &mut RequestChan, + subpath: Option<&Path>, + cfg: &Self::Config, + max_history: Option, + ) -> Result { + Ok(DimmerDevice { + dimmer: classes::Dimmer::register_devices( + drc, + subpath, + cfg, + max_history, + ) + .await?, + }) + } } impl HueDevice for DimmerDevice { @@ -85,74 +93,48 @@ impl HueDevice for DimmerDevice { } async fn next_setting(&mut self) -> Option { - loop { - tokio::select! { - // Check brightness device - opt_txn = self.inner.brightness.next_setting() => { - if let Some((val, reply)) = opt_txn { - let val = val.clamp(0.0, 100.0).round(); - - debug!("dimmer brightness setting ready: {}", val); - - if let Some(r) = reply { - r.ok(val); - } - - let cmd = if val == 0.0 { - payload::LightCommand { - on: Some(payload::On { on: false }), - dimming: None, - color: None, - } - } else { - payload::LightCommand { - on: Some(payload::On { on: true }), - dimming: Some(payload::Dimming { - brightness: val as f32, - }), - color: None, - } - }; - return Some(cmd); - } + if let Some(brightness) = self.dimmer.next_setting().await { + let cmd = if brightness.brightness == 0.0 { + payload::LightCommand { + on: Some(payload::On { on: false }), + dimming: None, + color: None, } - - // Check indicator device (drain but don't send command) - opt_txn = self.inner.indicator.next_setting() => { - if let Some((val, Some(r))) = opt_txn { - r.ok(val); - } + } else { + payload::LightCommand { + on: Some(payload::On { on: true }), + dimming: Some(payload::Dimming { + brightness: brightness.brightness as f32, + }), + color: None, } - } + }; + Some(cmd) + } else { + Some(payload::LightCommand { + on: None, + dimming: None, + color: None, + }) } } async fn apply_update(&mut self, update: &payload::ResourceData) { - if let Some(on) = &update.on { - if !on.on { - debug!("dimmer: reporting brightness update: 0"); - self.inner.brightness.report_update(0.0).await; - } else if let Some(dim) = &update.dimming { - let brightness = (dim.brightness as f64).round(); - - debug!("dimmer: reporting brightness update: {}", brightness); - self.inner.brightness.report_update(brightness).await; - } else { - // Use 100% if brightness is missing when device is on - debug!("dimmer: reporting brightness update: 100 (default)"); - self.inner.brightness.report_update(100.0).await; - } - } else if let Some(dim) = &update.dimming { - let brightness = (dim.brightness as f64).round(); - - debug!("dimmer: reporting brightness update: {}", brightness); - self.inner.brightness.report_update(brightness).await; - } + let brightness = match (&update.on, &update.dimming) { + (Some(payload::On { on: false }), _) => 0.0, + (Some(payload::On { on: true }), None) => 100.0, + (_, Some(dim)) => (dim.brightness as f64).round(), + (None, None) => return, + }; + + debug!("dimmer: brightness update: {}", brightness); + self.dimmer + .report_update(classes::DimmerProperty { brightness }) + .await; } fn reset(&mut self) { - self.inner.brightness.reset_state(); - self.inner.indicator.reset_state(); + self.dimmer.reset_state(); } } @@ -163,6 +145,10 @@ pub struct ColorBulbDevice { /// Stores the last XY coordinates sent to the bridge to avoid /// round-off errors when comparing RGB <-> XY conversions last_xy: Option<(f32, f32)>, + /// The last color reported to DrMem, used as the baseline when + /// merging partial bridge updates (on/dimming/color can each + /// arrive independently). + last_color: ColorType, } impl ColorBulbDevice { @@ -175,6 +161,9 @@ impl ColorBulbDevice { constants::LIGHT_RESOURCE }, last_xy: None, + last_color: ColorType::Rgba { + color: LinSrgba::new(255, 255, 255, 0), + }, } } } @@ -185,118 +174,76 @@ impl HueDevice for ColorBulbDevice { } async fn next_setting(&mut self) -> Option { - loop { - tokio::select! { - // Check brightness device - opt_txn = self.inner.brightness.next_setting() => { - if let Some((val, reply)) = opt_txn { - let val = val.clamp(0.0, 100.0).round(); - debug!("colorbulb brightness setting ready: {}", val); - - if let Some(r) = reply { - r.ok(val); - } - - let cmd = if val == 0.0 { - payload::LightCommand { - on: Some(payload::On { on: false }), - dimming: None, - color: None, - } - } else { - payload::LightCommand { - on: Some(payload::On { on: true }), - dimming: Some(payload::Dimming { - brightness: val as f32, - }), - color: None, - } - }; - return Some(cmd); - } - } + let (val, reply) = self.inner.color.next_setting().await?; - // Check color device - opt_txn = self.inner.color.next_setting() => { - if let Some((val, reply)) = opt_txn { - debug!("colorbulb color setting ready: {:?}", val); - - if let Some(r) = reply { - r.ok(val.clone()); - } - - let (x, y) = super::color::rgba_to_cie_xy(&val); - - // Store the XY coordinates we're sending to the bridge - self.last_xy = Some((x, y)); - debug!("colorbulb: saved XY coordinates: ({}, {})", x, y); - - return Some(payload::LightCommand { - on: Some(payload::On { on: true }), - dimming: None, - color: Some(payload::Color { - xy: Some(payload::XyCoordinates { x, y }), - }), - }); - } - } - } + debug!("colorbulb color setting ready: {:?}", val); + + if let Some(r) = reply { + r.ok(val.clone()); } + + let bridge = color::color_to_bridge(&val); + + // Store the XY coordinates we're sending to the bridge, and the + // setting itself so `apply_update` knows which `ColorType` + // variant (`Rgba` or `Ccta`) to preserve when the bridge just + // echoes back the xy we sent. + self.last_xy = Some(bridge.xy); + self.last_color = val; + + Some(payload::LightCommand { + on: Some(payload::On { on: bridge.on }), + dimming: bridge.on.then_some(payload::Dimming { + brightness: bridge.brightness, + }), + color: Some(payload::Color { + xy: Some(payload::XyCoordinates { + x: bridge.xy.0, + y: bridge.xy.1, + }), + }), + }) } async fn apply_update(&mut self, update: &payload::ResourceData) { - // Handle brightness updates - if let Some(on) = &update.on { - if !on.on { - debug!("colorbulb: reporting brightness update: 0"); - self.inner.brightness.report_update(0.0).await; - } else if let Some(dim) = &update.dimming { - let brightness = (dim.brightness as f64).round(); - - debug!( - "colorbulb: reporting brightness update: {}", - brightness - ); - self.inner.brightness.report_update(brightness).await; - } else { - // Use 100% if brightness is missing when device is on - debug!("colorbulb: reporting brightness update: 100 (default)"); - self.inner.brightness.report_update(100.0).await; - } - } else if let Some(dim) = &update.dimming { - let brightness = (dim.brightness as f64).round(); + let on = update.on.as_ref().map(|on| on.on); + let brightness = update.dimming.as_ref().map(|dim| dim.brightness); + + // Ignore XY coordinates that just echo what we last sent, to + // avoid round-off errors from the RGB <-> XY conversion. + let xy = update + .color + .as_ref() + .and_then(|c| c.xy.as_ref()) + .map(|xy| (xy.x, xy.y)) + .filter(|&(x, y)| { + !self.last_xy.is_some_and(|(last_x, last_y)| { + (last_x - x).abs() < 0.001 && (last_y - y).abs() < 0.001 + }) + }); - debug!("colorbulb: reporting brightness update: {}", brightness); - self.inner.brightness.report_update(brightness).await; + if on.is_none() && brightness.is_none() && xy.is_none() { + return; } - // Handle color updates - if let Some(color) = &update.color.as_ref().and_then(|c| c.xy.as_ref()) - { - let bridge_xy = (color.x, color.y); + let merged = + color::merge_bridge_update(&self.last_color, on, brightness, xy); - // Check if the bridge's XY coordinates match our last sent XY - // Use a tolerance to handle floating point precision + debug!("colorbulb: reporting color update: {:?}", merged); - let xy_matches = self.last_xy.map_or(false, |(last_x, last_y)| { - (last_x - bridge_xy.0).abs() < 0.001 - && (last_y - bridge_xy.1).abs() < 0.001 - }); - - if !xy_matches { - let rgba = - super::color::cie_xy_to_rgba(bridge_xy.0, bridge_xy.1); - - self.inner.color.report_update(rgba).await; - self.last_xy = Some(bridge_xy); - } + if let Some(xy) = xy { + self.last_xy = Some(xy); } + self.last_color = merged.clone(); + self.inner.color.report_update(merged).await; } fn reset(&mut self) { - self.inner.brightness.reset_state(); self.inner.color.reset_state(); self.last_xy = None; + self.last_color = ColorType::Rgba { + color: LinSrgba::new(255, 255, 255, 0), + }; } } diff --git a/drivers/drmem-drv-tplink/src/device.rs b/drivers/drmem-drv-tplink/src/device.rs index 1be087f..3f8c037 100644 --- a/drivers/drmem-drv-tplink/src/device.rs +++ b/drivers/drmem-drv-tplink/src/device.rs @@ -1,18 +1,89 @@ use drmem_api::{ device::Path, - driver::{classes, Registrator, Reporter, RequestChan, ResettableState}, + driver::{ + classes, OverridableDevice, OverrideConfig, Registrator, Reporter, + RequestChan, ResettableState, + }, Result, }; use tokio::time::Duration; use crate::config; +pub enum Setting { + Dimmer(classes::DimmerProperty), + Indicator(bool), +} + +pub struct DimmerWithIndicator { + pub dimmer: classes::Dimmer, + pub indicator: OverridableDevice, +} + +impl DimmerWithIndicator { + /// Reports any new properties specified in the `prop` parameter. + pub async fn report_update(&mut self, prop: classes::DimmerProperty) { + self.dimmer.report_update(prop).await + } + + pub async fn next_setting(&mut self) -> Option { + tokio::select! { + Some(res) = self.dimmer.next_setting() => { + Some(Setting::Dimmer(res)) + }, + res = self.indicator.next_setting() => { + if let Some((value, resp)) = res { + if let Some(resp) = resp { + resp.ok(value); + } + Some(Setting::Indicator(value)) + } else { + None + } + } + } + } +} + +impl Registrator for DimmerWithIndicator { + type Config = OverrideConfig; + + async fn register_devices( + drc: &mut RequestChan, + subpath: Option<&Path>, + cfg: &Self::Config, + max_history: Option, + ) -> Result { + Ok(DimmerWithIndicator { + dimmer: classes::Dimmer::register_devices( + drc, + subpath, + cfg, + max_history, + ) + .await?, + indicator: drc + .add_overridable_device( + "indicator", + subpath, + None, + cfg.override_duration, + cfg.envelope, + max_history, + ) + .await?, + }) + } +} + +impl ResettableState for DimmerWithIndicator {} + // An instance of a driver can be a switch, dimmer, or outlet device. // This type specifies the device channels for the given types. The // driver instance will have one of these variants for its device set. pub enum Set { Switch(classes::Switch), - Dimmer(classes::Dimmer), + Dimmer(DimmerWithIndicator), } // A set of devices must be resettable (in case the device gets @@ -22,11 +93,10 @@ impl ResettableState for Set { fn reset_state(&mut self) { match self { Set::Switch(dev) => { - dev.state.reset_state(); - dev.indicator.reset_state(); + dev.reset_state(); } Set::Dimmer(dev) => { - dev.brightness.reset_state(); + dev.dimmer.reset_state(); dev.indicator.reset_state(); } } @@ -43,24 +113,30 @@ impl Registrator for Set { cfg: &Self::Config, max_history: Option, ) -> Result { + let override_cfg = OverrideConfig { + override_duration: cfg + .override_timeout + .map(|v| Duration::from_secs(60 * v)), + envelope: Some(Duration::from_secs(5)), + }; + match cfg.r#type { config::DevCfgType::Switch | config::DevCfgType::Outlet => { Ok(Set::Switch( classes::Switch::register_devices( drc, subpath, - &cfg.override_timeout - .map(|v| Duration::from_secs(60 * v)), + &override_cfg, max_history, ) .await?, )) } config::DevCfgType::Dimmer => Ok(Set::Dimmer( - classes::Dimmer::register_devices( + DimmerWithIndicator::register_devices( drc, subpath, - &cfg.override_timeout.map(|v| Duration::from_secs(60 * v)), + &override_cfg, max_history, ) .await?, diff --git a/drivers/drmem-drv-tplink/src/driver.rs b/drivers/drmem-drv-tplink/src/driver.rs index 107c97f..fe70726 100644 --- a/drivers/drmem-drv-tplink/src/driver.rs +++ b/drivers/drmem-drv-tplink/src/driver.rs @@ -440,22 +440,21 @@ impl Instance { if self.reported_error != Some(value) { self.reported_error = Some(value); match dev { - device::Set::Switch(classes::Switch { error, .. }) => { - error.report_update(value).await - } - device::Set::Dimmer(classes::Dimmer { error, .. }) => { - error.report_update(value).await - } + device::Set::Switch(sw) => sw.report_error(value).await, + device::Set::Dimmer(device::DimmerWithIndicator { + dimmer, + .. + }) => dimmer.report_error(value).await, } } } async fn pause_as_dimmer( &mut self, - dev: &mut classes::Dimmer, + dev: &mut device::DimmerWithIndicator, ) { - let classes::Dimmer { - brightness: ref mut d_b, + let device::DimmerWithIndicator { + dimmer: ref mut d_d, indicator: ref mut d_i, .. } = dev; @@ -469,11 +468,7 @@ impl Instance { _ = &mut timeout => { break; } - Some((v, reply)) = d_b.next_setting() => { - if let Some(reply) = reply { - reply.ok(v); - } - } + Some(_) = d_d.next_setting() => {} Some((v, reply)) = d_i.next_setting() => { if let Some(reply) = reply { reply.ok(v); @@ -487,12 +482,6 @@ impl Instance { &mut self, dev: &mut classes::Switch, ) { - let classes::Switch { - state: ref mut d_r, - indicator: ref mut d_i, - .. - } = dev; - let timeout = time::sleep(Duration::from_secs(60)); tokio::pin!(timeout); @@ -502,16 +491,7 @@ impl Instance { _ = &mut timeout => { break; } - Some((v, reply)) = d_r.next_setting() => { - if let Some(reply) = reply { - reply.ok(v); - } - } - Some((v, reply)) = d_i.next_setting() => { - if let Some(reply) = reply { - reply.ok(v); - } - } + Some(_) = dev.next_setting() => {} } } } @@ -521,12 +501,6 @@ impl Instance { task: JoinHandle>, dev: &mut classes::Switch, ) -> Result { - let classes::Switch { - state: ref mut d_r, - indicator: ref mut d_i, - .. - } = dev; - tokio::pin!(task); loop { @@ -543,16 +517,7 @@ impl Instance { } } } - Some((v, reply)) = d_r.next_setting() => { - if let Some(reply) = reply { - reply.ok(v); - } - } - Some((v, reply)) = d_i.next_setting() => { - if let Some(reply) = reply { - reply.ok(v); - } - } + Some(_) = dev.next_setting() => {} } } } @@ -560,10 +525,10 @@ impl Instance { async fn manage_connect_as_dimmer( &mut self, task: JoinHandle>, - dev: &mut classes::Dimmer, + dev: &mut device::DimmerWithIndicator, ) -> Result { - let classes::Dimmer { - brightness: ref mut d_b, + let device::DimmerWithIndicator { + dimmer: ref mut d_d, indicator: ref mut d_i, .. } = dev; @@ -584,11 +549,7 @@ impl Instance { } } } - Some((v, reply)) = d_b.next_setting() => { - if let Some(reply) = reply { - reply.ok(v); - } - } + Some(_) = d_d.next_setting() => {} Some((v, reply)) = d_i.next_setting() => { if let Some(reply) = reply { reply.ok(v); @@ -603,16 +564,6 @@ impl Instance { s: &mut TcpStream, dev: &mut classes::Switch, ) -> bool { - // Get mutable references to the setting channels. - - let classes::Switch { - state: ref mut d_r, - indicator: ref mut d_i, - .. - } = dev; - - // Now wait for one of three events to occur. - #[rustfmt::skip] tokio::select! { // If our poll timeout expires, we need to request the @@ -630,12 +581,8 @@ impl Instance { match self.info_rpc(s).await { Ok(info) => { - if let Some(indicator) = info.indicator { - d_i.report_update(indicator).await - } - if let Some(relay) = info.relay { - d_r.report_update(relay).await + dev.report_update(classes::SwitchProperty { state: relay }).await } } Err(e) => { @@ -645,27 +592,13 @@ impl Instance { } } - // Handle settings to the brightness device. + // Handle settings to the state device. - Some((v, reply)) = d_r.next_setting() => { - if let Some(reply) = reply { - reply.ok(v); - } + Some(classes::SwitchProperty { state: v }) = dev.next_setting() => { if let Err(e) = self.relay_state_rpc(s, v).await { error!("couldn't set relay state -- {e}"); - return false - }; - self.poll_timeout = Duration::from_secs(0); - } - - // Handle settings to the LED indicator device. - - Some((v, reply)) = d_i.next_setting() => { - debug!("led setting -> {}", &v); - if let Err(e) = self.handle_led_setting(s, v, reply).await { - error!("couldn't set indicator -- {e}"); - return false - } + return false + } self.poll_timeout = Duration::from_secs(0); } } @@ -675,17 +608,9 @@ impl Instance { async fn manage_dimmer( &mut self, s: &mut TcpStream, - dev: &mut classes::Dimmer, + dev: &mut device::DimmerWithIndicator, ) -> bool { - // Get mutable references to the setting channels. - - let classes::Dimmer { - brightness: ref mut d_b, - indicator: ref mut d_i, - .. - } = dev; - - // Now wait for one of three events to occur. + // Now wait for one of two events to occur. #[rustfmt::skip] tokio::select! { @@ -705,11 +630,11 @@ impl Instance { match self.info_rpc(s).await { Ok(info) => { if let Some(indicator) = info.indicator { - d_i.report_update(indicator).await + dev.indicator.report_update(indicator).await } if let Some(brightness) = info.brightness { - d_b.report_update(brightness).await + dev.dimmer.report_update(classes::DimmerProperty{brightness}).await } } Err(e) => { @@ -719,24 +644,22 @@ impl Instance { } } - // Handle settings to the brightness device. - - Some((v, reply)) = d_b.next_setting() => { - if let Err(e) = self.handle_brightness_setting(s, v, reply).await { - error!("couldn't set brightness -- {e}"); - return false - }; - self.poll_timeout = Duration::from_secs(0); - } - - // Handle settings to the LED indicator device. - - Some((v, reply)) = d_i.next_setting() => { - debug!("led setting -> {}", &v); - if let Err(e) = self.handle_led_setting(s, v, reply).await { - error!("couldn't set indicator -- {e}"); - return false - } + Some(setting) = dev.next_setting() => { + match setting { + device::Setting::Dimmer(prop) => { + if let Err(e) = self.handle_brightness_setting(s, prop.brightness, None).await { + error!("couldn't set brightness -- {e}"); + return false + } + } + device::Setting::Indicator(v) => { + debug!("led setting -> {}", &v); + if let Err(e) = self.handle_led_setting(s, v, None).await { + error!("couldn't set indicator -- {e}"); + return false + } + } + } self.poll_timeout = Duration::from_secs(0); } } diff --git a/drmem-api/src/driver.rs b/drmem-api/src/driver.rs index 7513685..0d7ee81 100644 --- a/drmem-api/src/driver.rs +++ b/drmem-api/src/driver.rs @@ -28,7 +28,7 @@ mod ro_device; mod rw_device; pub use config::DriverConfig; -pub use overridable_device::OverridableDevice; +pub use overridable_device::{OverridableDevice, OverrideConfig}; pub use ro_device::ReadOnlyDevice; pub use rw_device::{ ReadWriteDevice, RxDeviceSetting, SettingRequest, SettingResponder, @@ -265,6 +265,7 @@ impl RequestChan { subpath: Option<&Path>, units: Option<&str>, override_duration: Option, + envelope: Option, max_history: Option, ) -> Result> where @@ -292,6 +293,7 @@ impl RequestChan { rs, prev.and_then(|v| T::try_from(v).ok()), override_duration, + envelope, ) }); } diff --git a/drmem-api/src/driver/classes.rs b/drmem-api/src/driver/classes.rs index 53a5f28..757a6a2 100644 --- a/drmem-api/src/driver/classes.rs +++ b/drmem-api/src/driver/classes.rs @@ -16,6 +16,6 @@ mod weather_type; // Make top-level types available to driver writers. pub use colorbulb_type::ColorBulb; -pub use dimmer_type::Dimmer; +pub use dimmer_type::{Dimmer, DimmerProperty}; pub use switch_type::{Switch, SwitchProperty}; pub use weather_type::{Weather, WeatherConfig, WeatherUnits}; diff --git a/drmem-api/src/driver/classes/colorbulb_type.rs b/drmem-api/src/driver/classes/colorbulb_type.rs index 7a441dd..2d2dfed 100644 --- a/drmem-api/src/driver/classes/colorbulb_type.rs +++ b/drmem-api/src/driver/classes/colorbulb_type.rs @@ -17,27 +17,25 @@ //! ``` use crate::{ - device::Path, + device::{ColorType, Path}, driver::{ - OverridableDevice, ReadOnlyDevice, Registrator, Reporter, RequestChan, - Result, + OverridableDevice, OverrideConfig, ReadOnlyDevice, Registrator, + Reporter, RequestChan, Result, }, }; -use tokio::time::Duration; /// Defines the common API used by Dimmers. pub struct ColorBulb { /// This device returns `true` when the driver has a problem /// communicating with the hardware. pub error: ReadOnlyDevice, - /// Controls the brightness setting of the bulb. Off is 0.0 and - /// full-on is 100.0. - pub brightness: OverridableDevice, - pub color: OverridableDevice, R>, + /// Controls the color of the bulb. Brightness is conveyed via the + /// color's alpha channel: 0 is off and 255 is full brightness. + pub color: OverridableDevice, } impl Registrator for ColorBulb { - type Config = Option; + type Config = OverrideConfig; async fn register_devices( drc: &mut RequestChan, @@ -49,21 +47,13 @@ impl Registrator for ColorBulb { error: drc .add_ro_device("error", subpath, None, max_history) .await?, - brightness: drc - .add_overridable_device( - "brightness", - subpath, - Some("%"), - *cfg, - max_history, - ) - .await?, color: drc .add_overridable_device( "color", subpath, None, - *cfg, + cfg.override_duration, + cfg.envelope, max_history, ) .await?, @@ -73,7 +63,6 @@ impl Registrator for ColorBulb { impl crate::driver::ResettableState for ColorBulb { fn reset_state(&mut self) { - self.brightness.reset_state(); self.color.reset_state(); } } diff --git a/drmem-api/src/driver/classes/dimmer_type.rs b/drmem-api/src/driver/classes/dimmer_type.rs index 8c48df2..0952445 100644 --- a/drmem-api/src/driver/classes/dimmer_type.rs +++ b/drmem-api/src/driver/classes/dimmer_type.rs @@ -19,64 +19,52 @@ use crate::{ device::Path, driver::{ - overridable_device::OverridableDevice, ro_device::ReadOnlyDevice, + overridable_device::{OverridableDevice, OverrideConfig}, + ro_device::ReadOnlyDevice, Registrator, Reporter, RequestChan, Result, }, }; -use tokio::time::Duration; pub struct DimmerProperty { - pub brightness: Option, - pub indicator: Option, + pub brightness: f64, } /// Defines the common API used by Dimmers. pub struct Dimmer { - /// This device returns `true` when the driver has a problem - /// communicating with the hardware. - pub error: ReadOnlyDevice, - /// Controls the brightness setting of the dimmer. Off is 0.0 and - /// full-on is 100.0. - pub brightness: OverridableDevice, - /// A product might include an indicator. If the hardware does, - /// this device can turn it on and off. - pub indicator: OverridableDevice, + // This device returns `true` when the driver has a problem + // communicating with the hardware. + error: ReadOnlyDevice, + // Controls the brightness setting of the dimmer. Off is 0.0 and + // full-on is 100.0. + brightness: OverridableDevice, } impl Dimmer { // Reports any new properties specified in the `prop` parameter. pub async fn report_update(&mut self, prop: DimmerProperty) { - if let Some(v) = prop.brightness { - self.brightness.report_update(v).await - } + self.brightness.report_update(prop.brightness).await; + } - if let Some(v) = prop.indicator { - self.indicator.report_update(v).await - } + pub async fn report_error(&mut self, error: bool) { + self.error.report_update(error).await; } - pub async fn next_setting(&mut self) -> DimmerProperty { - tokio::select! { - Some((value, resp)) = self.brightness.next_setting() => { - let value = value.clamp(0.0, 100.0); + pub async fn next_setting(&mut self) -> Option { + if let Some((value, resp)) = self.brightness.next_setting().await { + let value = value.clamp(0.0, 100.0); - if let Some(resp) = resp { - resp.ok(value); - } - DimmerProperty { brightness: Some(value), indicator: None } - } - Some((value, resp)) = self.indicator.next_setting() => { - if let Some(resp) = resp { - resp.ok(value); - } - DimmerProperty { brightness: None, indicator: Some(value) } + if let Some(resp) = resp { + resp.ok(value); } + return Some(DimmerProperty { brightness: value }); + } else { + None } } } impl Registrator for Dimmer { - type Config = Option; + type Config = OverrideConfig; async fn register_devices( drc: &mut RequestChan, @@ -93,16 +81,8 @@ impl Registrator for Dimmer { "brightness", subpath, Some("%"), - *cfg, - max_history, - ) - .await?, - indicator: drc - .add_overridable_device( - "indicator", - subpath, - None, - *cfg, + cfg.override_duration, + cfg.envelope, max_history, ) .await?, diff --git a/drmem-api/src/driver/classes/switch_type.rs b/drmem-api/src/driver/classes/switch_type.rs index a3f195e..05d8188 100644 --- a/drmem-api/src/driver/classes/switch_type.rs +++ b/drmem-api/src/driver/classes/switch_type.rs @@ -19,62 +19,48 @@ use crate::{ device::Path, driver::{ - overridable_device::OverridableDevice, ro_device::ReadOnlyDevice, + overridable_device::{OverridableDevice, OverrideConfig}, + ro_device::ReadOnlyDevice, Registrator, Reporter, RequestChan, Result, }, }; -use tokio::time::Duration; pub struct SwitchProperty { - pub state: Option, - pub indicator: Option, + pub state: bool, } /// Defines the common API used by Switches. pub struct Switch { /// This device returns `true` when the driver has a problem /// communicating with the hardware. - pub error: ReadOnlyDevice, + error: ReadOnlyDevice, /// Indicates the state of the switch. Writing `true` or `false` /// turns the switch on and off, respectively. - pub state: OverridableDevice, - /// A product might include an indicator. If the hardware does, - /// this device can turn it on and off. - pub indicator: OverridableDevice, + state: OverridableDevice, } impl Switch { // Reports any new properties specified in the `prop` parameter. pub async fn report_update(&mut self, prop: SwitchProperty) { - if let Some(v) = prop.state { - self.state.report_update(v).await - } + self.state.report_update(prop.state).await + } - if let Some(v) = prop.indicator { - self.indicator.report_update(v).await - } + pub async fn report_error(&mut self, error: bool) { + self.error.report_update(error).await } - pub async fn next_setting(&mut self) -> SwitchProperty { - tokio::select! { - Some((value, resp)) = self.state.next_setting() => { - if let Some(resp) = resp { - resp.ok(value); - } - SwitchProperty { state: Some(value), indicator: None } - } - Some((value, resp)) = self.indicator.next_setting() => { - if let Some(resp) = resp { - resp.ok(value); - } - SwitchProperty { state: None, indicator: Some(value) } - } + pub async fn next_setting(&mut self) -> Option { + let (value, resp) = self.state.next_setting().await?; + + if let Some(resp) = resp { + resp.ok(value); } + Some(SwitchProperty { state: value }) } } impl Registrator for Switch { - type Config = Option; + type Config = OverrideConfig; async fn register_devices( drc: &mut RequestChan, @@ -91,16 +77,8 @@ impl Registrator for Switch { "state", subpath, None, - *cfg, - max_history, - ) - .await?, - indicator: drc - .add_overridable_device( - "indicator", - subpath, - None, - *cfg, + cfg.override_duration, + cfg.envelope, max_history, ) .await?, diff --git a/drmem-api/src/driver/overridable_device.rs b/drmem-api/src/driver/overridable_device.rs index 8b9fb39..2cc203f 100644 --- a/drmem-api/src/driver/overridable_device.rs +++ b/drmem-api/src/driver/overridable_device.rs @@ -1,11 +1,10 @@ /// Defines a settable device that is "shared" with the outside world. /// -/// Some devices are purely controlled by DrMem. However, there are -/// many commercial devices that are intended to be controlled by -/// users outside of DrMem. LED WiFi light bulbs are one, obvious -/// example. For devices that can be controlled outside of DrMem, we -/// need a way to cooperatively control them. That's what -/// `OverridableDevice`s do. +/// Some devices are purely controlled by DrMem. However, there are many +/// commercial devices that are intended to be controlled by users outside of +/// DrMem. LED WiFi light bulbs are one, obvious example. For devices that can +/// be controlled outside of DrMem, we need a way to cooperatively control +/// them. That's what `OverridableDevice`s do. /// /// A driver that uses this type of device must do these steps in /// their main loop: @@ -16,17 +15,38 @@ /// - after setting the hardware to a new value, a poll should /// immediately be done followed by a `.report_update()` /// -/// `OverridableDevice`s implement a simple state machine to know how -/// to handle incoming incoming settings. +/// `OverridableDevice`s implement a simple state machine to know how to handle +/// incoming settings. Some hardware (e.g. a Philips Hue bridge) can optimistic- +/// ally confirm a setting before the physical device actually has it applied, +/// and later revert to the stale, real reading if the device never became +/// reachable. To tolerate this, a device can be given an "envelope": a settle +/// period, started when a setting is applied, during which polled readings are +/// compared against the setting but don't yet commit to `Synced` or +/// `Overridden`. Mismatches during the envelope reassert the setting and extend +/// the envelope; when it finally elapses, the last polled reading decides the +/// outcome. use crate::{ device, driver::{rw_device, Reporter, RxDeviceSetting, SettingResponder}, }; use tokio_stream::StreamExt; -use tracing::{info, instrument, warn}; +use tracing::{debug, info, instrument, warn}; pub type SettingTransaction = (T, Option>); +/// Bundles the two independent timeouts with which an `OverridableDevice` can +/// be configured. +#[derive(Clone, Copy, Debug, Default)] +pub struct OverrideConfig { + /// How long the device stays `Overridden` before DrMem re-asserts its last + /// setting. `None` means DrMem never re-asserts on its own. + pub override_duration: Option, + /// How long to wait, after applying a setting, before trusting a polled + /// reading enough to leave the "applying" state. `None` disables the + /// envelope, matching a polled reading immediately. + pub envelope: Option, +} + // Describes the states that the device goes through as it receives // settings and polled readings. @@ -34,7 +54,7 @@ enum State { Unknown, UnknownTrans { value: T, - report: SettingResponder, + report: Option>, }, Synced { value: T, @@ -42,18 +62,36 @@ enum State { SyncedTrans { value: T, }, - UnreportedSetting { - value: T, - }, - Setting { - value: T, - }, - SettingTrans { - value: (T, Option>), + + // A setting is being applied to the hardware. `last_seen` is the most + // recent polled reading observed since `setting` was (re)issued; it starts + // out equal to `setting` (optimistic) until a `report_update()` says + // otherwise. `deadline` only matters when an envelope is configured; it's + // when we stop waiting for corroborating readings and commit to an outcome. + // `needs_reaffirm` is set when a client redundantly re-sends the value + // we're already applying; it tells `.next_setting()` to re-report that + // value to the backend without bothering the driver again. + Applying { + setting: T, + last_seen: T, + deadline: tokio::time::Instant, + needs_reaffirm: bool, }, - ReassertSetting { - value: T, + + // Same as `Applying`, but there's a value that still needs to be handed to + // the driver via `.next_setting()` -- either a brand new target + // (`needs_report` is `true`, since the backend hasn't seen it yet) or a + // reassertion of the current setting after a mismatched reading + // (`needs_report` is `false`, since the backend already reflects this + // target from when it was first applied). + ApplyingTrans { + setting: T, + last_seen: T, + deadline: tokio::time::Instant, + to_send: SettingTransaction, + needs_report: bool, }, + Overridden { setting: T, r#override: T, @@ -61,9 +99,28 @@ enum State { }, } +impl std::fmt::Debug for State { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + State::Unknown => write!(f, "<>"), + State::UnknownTrans { .. } => { + write!(f, "<>") + } + State::Synced { .. } => write!(f, "<>"), + State::SyncedTrans { .. } => { + write!(f, "<>") + } + State::Applying { .. } => write!(f, "<>",), + State::ApplyingTrans { .. } => write!(f, "<>",), + State::Overridden { .. } => write!(f, "<>",), + } + } +} + pub struct OverridableDevice { state: State, override_duration: Option, + envelope: tokio::time::Duration, reporter: R, set_stream: rw_device::SettingStream, } @@ -78,30 +135,46 @@ where setting_chan: RxDeviceSetting, desired_value: Option, override_duration: Option, + envelope: Option, ) -> Self { + let state = desired_value + .map(|value| State::ApplyingTrans { + setting: value.clone(), + last_seen: value.clone(), + deadline: tokio::time::Instant::now() + + envelope.unwrap_or_default(), + to_send: (value, None), + needs_report: true, + }) + .unwrap_or(State::Unknown); + + debug!("initial state: {:?}", state); OverridableDevice { - state: desired_value - .map(|value| State::SettingTrans { - value: (value, None), - }) - .unwrap_or(State::Unknown), + state, reporter, set_stream: rw_device::create_setting_stream(setting_chan), override_duration, + envelope: envelope.unwrap_or_default(), } } /// Saves a new value, returned by the device, to the backend /// storage. This only writes values that have changed. /// - /// This method is not cancel-safe. + /// This method is cancel-safe: every branch either performs no + /// `.await` or mutates `self.state` only after its single + /// `.await` resolves, so a cancelled call can be retried without + /// losing or duplicating a report. #[inline(never)] - #[instrument(skip(self))] + #[instrument(skip(self), fields(state = ?self.state))] pub async fn report_update(&mut self, new_value: T) { + debug!("waiting for the next report in state {:?}", self.state); + let next_deadline = self .override_duration .map(|duration| tokio::time::Instant::now() + duration) .unwrap_or_else(tokio::time::Instant::now); + let next_envelope = tokio::time::Instant::now() + self.envelope; match &mut self.state { State::Unknown => { @@ -111,6 +184,7 @@ where // reading, we switch to the synced state. If a // setting comes it, it can further modify the state. + debug!("entering Synced state with value {:?}", &new_value); self.state = State::Synced { value: new_value } } @@ -136,14 +210,17 @@ where // save the new reading. if setting == &new_value { - info!("value matches setting, transitioning to Synced"); self.reporter.report_value(new_value.clone().into()).await; + debug!("entering Synced state with value {:?}", &new_value); self.state = State::Synced { value: setting.clone(), }; } else if value != &new_value { - info!("value differs from setting, transitioning to Overridden"); self.reporter.report_value(new_value.clone().into()).await; + debug!( + "reset Overridden state with new value {:?}", + &new_value + ); self.state = State::Overridden { deadline: next_deadline, setting: value.clone(), @@ -158,6 +235,7 @@ where if value != &new_value { self.reporter.report_value(new_value.clone().into()).await; + debug!("value differs from previous, transitioning to Overridden"); self.state = State::Overridden { deadline: next_deadline, setting: value.clone(), @@ -166,70 +244,70 @@ where } } - State::Setting { value } => { - // When we're handling a setting, we enter the - // `Synced` state if the polled value matches. If not, - // then we have to reassert the setting. - - self.state = if value == &new_value { - State::Synced { value: new_value } + // We're applying a setting and nothing is pending + // delivery to the driver. A matching reading is + // remembered but, if an envelope is configured, doesn't + // commit to `Synced` until the envelope elapses (that + // decision is made in `.next_setting()`). A mismatch + // always reasserts the setting. + State::Applying { + setting, last_seen, .. + } => { + if setting == &new_value { + *last_seen = new_value; } else { - warn!("reasserting setting"); - State::ReassertSetting { - value: value.clone(), - } - } - } - - State::ReassertSetting { value } - | State::UnreportedSetting { value } => { - // When we're handling a setting, we enter the - // `Synced` state if the polled value matches. - - if value == &new_value { - self.state = State::Synced { value: new_value } + let setting = setting.clone(); + + warn!("reasserting setting, via ApplyingTrans"); + self.state = State::ApplyingTrans { + setting: setting.clone(), + last_seen: new_value, + deadline: next_envelope, + to_send: (setting, None), + needs_report: false, + }; } } - State::SettingTrans { - value: (value, resp_ref), + // Same as `Applying`, but a value is still waiting to be + // handed to the driver. A matching reading means that + // value is no longer needed (the hardware is already + // there), so the pending client is ack'd and we collapse + // back to a settled state. A mismatch just extends the + // envelope; the pending delivery is untouched. + State::ApplyingTrans { + setting, + last_seen, + deadline, + to_send, + .. } => { - // If the polled value happens to equal the incoming - // setting that's still being processed, we should go - // into the `Synced` state. - - if value == &new_value { - let value = value.clone(); + *last_seen = new_value.clone(); - // If there was a function to reply to the client, - // we need to perform the reply. + if setting == &new_value { + let setting = setting.clone(); + let old_deadline = *deadline; - if let Some(resp) = resp_ref.take() { - resp.ok(value.clone()); + if let Some(resp) = to_send.1.take() { + resp.ok(new_value); } - // Go to the `SyncedTrans` state, which will - // report the new value to the backend. - - self.state = State::SyncedTrans { value }; + info!("has envelope, transitioning to Applying"); + self.state = State::Applying { + last_seen: setting.clone(), + setting, + deadline: old_deadline, + needs_reaffirm: false, + }; + } else { + *deadline = next_envelope; } } State::SyncedTrans { value } => { if value != &new_value { - // These two statements are the reason this method - // isn't cancel-safe. We could try to add an - // `OverriddenTrans` state, but then we have to - // figure out what to do when settings or new - // polled values arrive. - // - // The sad part is that this state and the - // previous `SettingTrans` state are probably - // never going to be active when a new value is - // reported with this method. - - self.reporter.report_value(value.clone().into()).await; self.reporter.report_value(new_value.clone().into()).await; + info!("value differs from previous, transitioning to Overridden"); self.state = State::Overridden { deadline: next_deadline, setting: value.clone(), @@ -250,10 +328,8 @@ where State::UnknownTrans { value, .. } | State::Synced { value } | State::SyncedTrans { value } - | State::Setting { value } - | State::SettingTrans { value: (value, _) } - | State::ReassertSetting { value } - | State::UnreportedSetting { value } => Some(value), + | State::Applying { setting: value, .. } + | State::ApplyingTrans { setting: value, .. } => Some(value), State::Overridden { r#override, .. } => Some(r#override), } } @@ -262,118 +338,202 @@ where /// /// This method is cancel-safe. #[inline(never)] - #[instrument(skip(self))] + #[instrument(skip(self), fields(state = ?self.state))] pub async fn next_setting(&mut self) -> Option> { + debug!("waiting for the next setting in state {:?}", self.state); let result = loop { match &mut self.state { - // At this point, we have no known state. If a - // setting comes in, we're going to assume it's - // different from the hardware's state so we - // switch to Setting. - State::Unknown => match self.set_stream.next().await { - Some(reply) => { - self.state = State::UnknownTrans { - value: reply.0.clone(), - report: reply.1, - }; - } - None => break None, - }, + // At this point, we have no known state. If a setting comes in, + // we're going to assume it's different from the hardware's + // state so we switch to Applying (via UnknownTrans). + State::Unknown => { + let reply = self.set_stream.next().await?; + + debug!( + "received new setting, transitioning to UnknownTrans" + ); + self.state = State::UnknownTrans { + value: reply.0, + report: Some(reply.1), + }; + } - // This is a transition state between Unknown and - // Setting. This was needed to break up the Unknown - // state so that each state has one future to - // await. This makes the function "cancel safe". - State::UnknownTrans { value, .. } => { + // This is a transition state between Unknown and Applying. + // This was needed to break up the Unknown state so that each + // state has one future to await. This makes the function + // "cancel safe". + State::UnknownTrans { value, report } => { self.reporter.report_value(value.clone().into()).await; let value = value.clone(); + let deadline = tokio::time::Instant::now() + self.envelope; + let report = report.take(); - if let State::UnknownTrans { value, report } = - std::mem::replace( - &mut self.state, - State::Setting { value }, - ) - { - break Some((value, Some(report))); + debug!("transitioning to Applying"); + self.state = State::Applying { + setting: value.clone(), + last_seen: value.clone(), + deadline, + needs_reaffirm: false, + }; + break Some((value, report)); + } + + State::Synced { value } => { + let reply = self.set_stream.next().await?; + + self.state = if reply.0 != *value { + debug!("value differs from previous, transitioning to ApplyingTrans"); + State::ApplyingTrans { + setting: reply.0.clone(), + last_seen: value.clone(), + deadline: tokio::time::Instant::now() + + self.envelope, + to_send: (reply.0, Some(reply.1)), + needs_report: true, + } } else { - unreachable!() - } + reply.1.ok(reply.0.clone()); + debug!("value matches previous, transitioning to SyncedTrans"); + State::SyncedTrans { value: reply.0 } + }; } - // If we have an unreported setting, re-report it and - // switch to the "reported" setting state. - State::UnreportedSetting { value } => { + State::SyncedTrans { value } => { self.reporter.report_value(value.clone().into()).await; - self.state = State::Setting { + debug!("transitioning from SyncedTrans to Synced"); + self.state = State::Synced { value: value.clone(), }; } - // We need to reassert the setting. Immediately return - // it and switch to the "reported" setting state. - State::ReassertSetting { value } => { - let result = (value.clone(), None); + // A setting is being applied and there's nothing + // pending delivery to the driver. + State::Applying { + setting, + last_seen, + deadline, + needs_reaffirm, + } => { + // The setting needs to be reasserted. + // Re-report it to the backend (without bothering the + // driver again) before doing anything else. + if *needs_reaffirm { + self.reporter + .report_value(setting.clone().into()) + .await; + *needs_reaffirm = false; + } - self.state = State::Setting { - value: value.clone(), - }; - break Some(result); - } + let now = tokio::time::Instant::now(); - State::Setting { value } => { - match self.set_stream.next().await { - Some(reply) => { - if reply.0 != *value { - self.state = State::SettingTrans { - value: (reply.0, Some(reply.1)), - }; - } else { - reply.1.ok(reply.0.clone()); - self.state = - State::UnreportedSetting { value: reply.0 }; - } + if *deadline <= now { + if last_seen == setting { + debug!("deadline reached and last_seen matches setting, transitioning to Synced"); + self.state = State::Synced { + value: setting.clone(), + }; + } else { + // The backend never saw this reading while we were + // waiting out the envelope, so report it now or + // clients are stuck seeing the stale setting. + self.reporter + .report_value(last_seen.clone().into()) + .await; + debug!("deadline reached and last_seen differs from setting, transitioning to Overridden"); + self.state = State::Overridden { + setting: setting.clone(), + r#override: last_seen.clone(), + deadline: now + + self + .override_duration + .unwrap_or_default(), + }; } - None => break None, + continue; } - } - State::Synced { value } => match self.set_stream.next().await { - Some(reply) => { - self.state = if reply.0 != *value { - State::SettingTrans { - value: (reply.0, Some(reply.1)), + let delay = deadline.saturating_duration_since(now); + + // Wait for a setting, or for the envelope to + // elapse so we can decide the outcome. + + tokio::select! { + reply = self.set_stream.next() => { + match reply { + Some(r) => { + if r.0 != *setting { + debug!("new setting differs from current, transitioning to ApplyingTrans"); + self.state = State::ApplyingTrans { + setting: r.0.clone(), + last_seen: last_seen.clone(), + deadline: tokio::time::Instant::now() + self.envelope, + to_send: (r.0, Some(r.1)), + needs_report: true, + }; + } else { + r.1.ok(r.0.clone()); + *needs_reaffirm = true; + } + } + None => break None, } - } else { - reply.1.ok(reply.0.clone()); - State::SyncedTrans { value: reply.0 } - }; + } + _ = tokio::time::sleep(delay) => { + if last_seen == setting { + debug!("TIMEOUT : last seen matches setting, transitioning to Synced"); + self.state = State::Synced { value: setting.clone() }; + } else { + // Same as above: report the real + // reading before committing to + // `Overridden`. + self.reporter.report_value(last_seen.clone().into()).await; + info!("TIMEOUT : last seen differs from setting, transitioning to Overridden"); + self.state = State::Overridden { + setting: setting.clone(), + r#override: last_seen.clone(), + deadline: tokio::time::Instant::now() + self.override_duration.unwrap_or_default(), + }; + } + } } - None => break None, - }, - - State::SettingTrans { value: (val, _) } => { - self.reporter.report_value(val.clone().into()).await; - - let val = val.clone(); + } - if let State::SettingTrans { value: reply } = - std::mem::replace( - &mut self.state, - State::Setting { value: val.clone() }, - ) - { - break Some(reply); - } else { - unreachable!() + // Same as `Applying`, but a value still needs to be + // handed to the driver -- either a brand new target + // (reported here, since the backend hasn't seen it + // yet) or a reassertion after a mismatch (never + // reported; the backend already reflects this + // target). + State::ApplyingTrans { + setting, + last_seen, + deadline, + to_send, + needs_report, + } => { + if *needs_report { + self.reporter + .report_value(to_send.0.clone().into()) + .await; } - } - State::SyncedTrans { value } => { - self.reporter.report_value(value.clone().into()).await; - self.state = State::Synced { - value: value.clone(), + let setting = setting.clone(); + let last_seen = last_seen.clone(); + let deadline = *deadline; + let result = (to_send.0.clone(), to_send.1.take()); + + debug!( + "transitioning to Applying with setting {:?}", + setting + ); + self.state = State::Applying { + setting, + last_seen, + deadline, + needs_reaffirm: false, }; + break Some(result); } State::Overridden { @@ -390,10 +550,16 @@ where if *deadline <= now { self.state = if setting != r#override { - State::SettingTrans { - value: (setting.clone(), None), + debug!("override deadline reached, transitioning to ApplyingTrans"); + State::ApplyingTrans { + setting: setting.clone(), + last_seen: r#override.clone(), + deadline: now + self.envelope, + to_send: (setting.clone(), None), + needs_report: true, } } else { + debug!("override deadline reached, transitioning to Synced"); State::Synced { value: r#override.clone(), } @@ -439,28 +605,34 @@ where // override, we go into the `Synced` // state. If they're different, then // treat it as a new setting and - // return it to the driver so the - // hardware can be adjusted. + // reassert it (protected by a fresh + // envelope, same as any other new + // setting). self.state = if setting != r#override { - State::SettingTrans { - value: (setting.clone(), None) + debug!("override deadline reached, transitioning to ApplyingTrans"); + State::ApplyingTrans { + setting: setting.clone(), + last_seen: r#override.clone(), + deadline: tokio::time::Instant::now() + + self.envelope, + to_send: (setting.clone(), None), + needs_report: true, } } else { + debug!("override deadline reached, transitioning to Synced"); State::Synced { value: r#override.clone() } } + } } } else { - match self.set_stream.next().await { - Some(reply) => { - *setting = reply.0.clone(); - reply.1.ok(reply.0); - } - None => break None, - } + let reply = self.set_stream.next().await?; + + *setting = reply.0.clone(); + reply.1.ok(reply.0) } } } @@ -506,6 +678,7 @@ mod tests { fn mk_device( init: Option, tmo: Option, + envelope: Option, ) -> ( TxDeviceSetting, mpsc::Receiver, @@ -517,14 +690,20 @@ mod tests { ( srtx, rrrx, - OverridableDevice::new(MockReporter(rrtx), srrx, init, tmo), + OverridableDevice::new( + MockReporter(rrtx), + srrx, + init, + tmo, + envelope, + ), ) } #[tokio::test] async fn test_override_deadline_is_absolute() { let (tx_set, mut rx_rdg, mut dev) = - mk_device::(Some(true), Some(Duration::from_secs(60))); + mk_device::(Some(true), Some(Duration::from_secs(60)), None); let deadline = tokio::time::Instant::now() + Duration::from_millis(25); dev.state = State::Overridden { @@ -552,7 +731,8 @@ mod tests { #[tokio::test] async fn test_initialized_shared_device() { - let (tx_set, mut rx_rdg, mut sh_dev) = mk_device::(Some(1), None); + let (tx_set, mut rx_rdg, mut sh_dev) = + mk_device::(Some(1), None, None); // A initialized shared device should assert the initial value. @@ -703,7 +883,8 @@ mod tests { #[test] fn test_uninitialized_shared_device() { - let (tx_set, mut rx_rdg, mut sh_dev) = mk_device::(None, None); + let (tx_set, mut rx_rdg, mut sh_dev) = + mk_device::(None, None, None); let waker = noop_waker(); let mut context = Context::from_waker(&waker); @@ -1013,7 +1194,7 @@ mod tests { #[tokio::test] async fn test_timed_overrides() { let (tx_set, mut rx_rdg, mut sh_dev) = - mk_device::(Some(true), Some(Duration::from_secs(60))); + mk_device::(Some(true), Some(Duration::from_secs(60)), None); // Was created with an initial value, so that should be // presented as the first setting. @@ -1147,4 +1328,566 @@ mod tests { std::mem::drop(tx_set) } + + #[tokio::test] + async fn test_envelope_holds_after_immediate_match() { + let (tx_set, mut rx_rdg, mut sh_dev) = mk_device::( + Some(true), + None, + Some(Duration::from_millis(100)), + ); + + // The initial desired value is delivered to the driver and + // reported to the backend, same as without an envelope. + assert!(matches!( + timeout(Duration::from_secs(0), sh_dev.next_setting()).await, + Ok(Some((true, None))) + )); + assert!(matches!( + timeout(Duration::from_secs(0), rx_rdg.recv()).await, + Ok(Some(device::Value::Bool(true))) + )); + + // A matching poll arrives right away. With the bridge-staleness + // bug this feature fixes, this is exactly the situation that + // must NOT be trusted immediately. + sh_dev.report_update(true).await; + + // Nothing new should be reported, and the device must not have + // committed to `Synced` yet. + assert!(timeout(Duration::from_secs(0), rx_rdg.recv()) + .await + .is_err()); + assert!(matches!(&sh_dev.state, State::Applying { .. })); + + // `.next_setting()` must stay pending -- the envelope is still + // open and no client setting has arrived. + assert!(timeout(Duration::from_millis(20), sh_dev.next_setting()) + .await + .is_err()); + + std::mem::drop(tx_set); + } + + #[tokio::test] + async fn test_envelope_reassert_on_mismatch() { + let (tx_set, mut rx_rdg, mut sh_dev) = mk_device::( + Some(true), + None, + Some(Duration::from_millis(200)), + ); + + assert!(matches!( + timeout(Duration::from_secs(0), sh_dev.next_setting()).await, + Ok(Some((true, None))) + )); + assert!(matches!( + timeout(Duration::from_secs(0), rx_rdg.recv()).await, + Ok(Some(device::Value::Bool(true))) + )); + + // A mismatched poll arrives. It must be remembered, but never + // pushed to the backend. + sh_dev.report_update(false).await; + assert!(timeout(Duration::from_secs(0), rx_rdg.recv()) + .await + .is_err()); + + // The reassert is handed straight back to the driver (no + // responder, since it's not a fresh client request) with no + // extra backend report. + assert!(matches!( + timeout(Duration::from_secs(0), sh_dev.next_setting()).await, + Ok(Some((true, None))) + )); + assert!(timeout(Duration::from_secs(0), rx_rdg.recv()) + .await + .is_err()); + + std::mem::drop(tx_set); + } + + #[tokio::test] + async fn test_envelope_reassert_extends_deadline() { + let (tx_set, mut rx_rdg, mut sh_dev) = mk_device::( + Some(true), + None, + Some(Duration::from_millis(150)), + ); + + assert!(matches!( + timeout(Duration::from_secs(0), sh_dev.next_setting()).await, + Ok(Some((true, None))) + )); + assert!(matches!( + timeout(Duration::from_secs(0), rx_rdg.recv()).await, + Ok(Some(device::Value::Bool(true))) + )); + + // Wait most of the way through the original envelope, then + // cause a mismatch. The deadline must be pushed out by a + // fresh envelope measured from now, not from the start. + tokio::time::sleep(Duration::from_millis(100)).await; + sh_dev.report_update(false).await; + assert!(matches!( + timeout(Duration::from_secs(0), sh_dev.next_setting()).await, + Ok(Some((true, None))) + )); + + // At the *original* deadline (~50ms from here) the device must + // still be waiting -- proving the mismatch reset the timer. + assert!(timeout(Duration::from_millis(60), sh_dev.next_setting()) + .await + .is_err()); + + // Let the hardware settle on the right value so the extended + // envelope can resolve cleanly once it elapses. + sh_dev.report_update(true).await; + assert!(timeout(Duration::from_millis(200), sh_dev.next_setting()) + .await + .is_err()); + assert!(matches!(&sh_dev.state, State::Synced { value: true })); + + std::mem::drop(tx_set); + } + + #[tokio::test] + async fn test_envelope_expiry_resolves_to_synced() { + let (tx_set, _rx_rdg, mut sh_dev) = + mk_device::(None, None, Some(Duration::from_secs(60))); + + sh_dev.state = State::Applying { + setting: true, + last_seen: true, + deadline: tokio::time::Instant::now(), + needs_reaffirm: false, + }; + + // Deadline already elapsed and the last reading matches, so + // this must resolve to `Synced` (and then just wait for a + // setting that never comes). + assert!(timeout(Duration::from_millis(50), sh_dev.next_setting()) + .await + .is_err()); + assert!(matches!(&sh_dev.state, State::Synced { value: true })); + + std::mem::drop(tx_set); + } + + #[tokio::test] + async fn test_envelope_expiry_resolves_to_overridden() { + let (tx_set, _rx_rdg, mut sh_dev) = mk_device::( + None, + Some(Duration::from_secs(60)), + Some(Duration::from_secs(60)), + ); + + sh_dev.state = State::Applying { + setting: true, + last_seen: false, + deadline: tokio::time::Instant::now(), + needs_reaffirm: false, + }; + + // Deadline already elapsed and the last reading still doesn't + // match, so we must give up and call it an override. + assert!(timeout(Duration::from_millis(50), sh_dev.next_setting()) + .await + .is_err()); + assert!(matches!( + &sh_dev.state, + State::Overridden { + setting: true, + r#override: false, + .. + } + )); + + std::mem::drop(tx_set); + } + + #[tokio::test] + async fn test_envelope_expiry_to_overridden_reports_value() { + let (tx_set, mut rx_rdg, mut sh_dev) = mk_device::( + None, + Some(Duration::from_secs(60)), + Some(Duration::from_secs(60)), + ); + + sh_dev.state = State::Applying { + setting: true, + last_seen: false, + deadline: tokio::time::Instant::now(), + needs_reaffirm: false, + }; + + // Deadline already elapsed and the last reading still doesn't + // match, so we give up and call it an override. The real + // (mismatched) value must be reported to the backend right + // then -- otherwise clients are stuck seeing the stale + // setting forever, since a later poll of the *same* value + // never triggers `report_update()`'s own Overridden-mismatch + // report. + assert!(timeout(Duration::from_millis(50), sh_dev.next_setting()) + .await + .is_err()); + assert!(matches!( + &sh_dev.state, + State::Overridden { + setting: true, + r#override: false, + .. + } + )); + assert!(matches!( + timeout(Duration::from_secs(0), rx_rdg.recv()).await, + Ok(Some(device::Value::Bool(false))) + )); + + std::mem::drop(tx_set); + } + + #[tokio::test] + async fn test_envelope_expiry_via_sleep_reports_value() { + let (tx_set, mut rx_rdg, mut sh_dev) = mk_device::( + None, + Some(Duration::from_secs(60)), + Some(Duration::from_millis(30)), + ); + + sh_dev.state = State::Applying { + setting: true, + last_seen: false, + deadline: tokio::time::Instant::now() + Duration::from_millis(30), + needs_reaffirm: false, + }; + + // Same as above, but let the envelope elapse naturally through + // the `tokio::select!` sleep arm instead of the "already + // expired" fast path. + assert!(timeout(Duration::from_millis(200), sh_dev.next_setting()) + .await + .is_err()); + assert!(matches!( + &sh_dev.state, + State::Overridden { + setting: true, + r#override: false, + .. + } + )); + assert!(matches!( + timeout(Duration::from_secs(0), rx_rdg.recv()).await, + Ok(Some(device::Value::Bool(false))) + )); + + std::mem::drop(tx_set); + } + + #[tokio::test] + async fn test_envelope_multiple_reasserts_then_settles() { + let (tx_set, mut rx_rdg, mut sh_dev) = mk_device::( + Some(true), + None, + Some(Duration::from_millis(80)), + ); + + assert!(matches!( + timeout(Duration::from_secs(0), sh_dev.next_setting()).await, + Ok(Some((true, None))) + )); + assert!(matches!( + timeout(Duration::from_secs(0), rx_rdg.recv()).await, + Ok(Some(device::Value::Bool(true))) + )); + + // Several mismatch/reassert cycles in a row. Each one extends + // the envelope and none of them touch the backend. + for _ in 0..3 { + sh_dev.report_update(false).await; + assert!(matches!( + timeout(Duration::from_secs(0), sh_dev.next_setting()).await, + Ok(Some((true, None))) + )); + assert!(timeout(Duration::from_secs(0), rx_rdg.recv()) + .await + .is_err()); + } + + // Finally the hardware settles on the right value. Once the + // full (most recent) envelope elapses with no further + // mismatch, we commit to `Synced`. + sh_dev.report_update(true).await; + assert!(timeout(Duration::from_millis(300), sh_dev.next_setting()) + .await + .is_err()); + assert!(matches!(&sh_dev.state, State::Synced { value: true })); + + std::mem::drop(tx_set); + } + + #[tokio::test] + async fn test_new_setting_arrives_mid_envelope() { + let (tx_set, mut rx_rdg, mut sh_dev) = + mk_device::(Some(1), None, Some(Duration::from_millis(200))); + + assert!(matches!( + timeout(Duration::from_secs(0), sh_dev.next_setting()).await, + Ok(Some((1, None))) + )); + assert!(matches!( + timeout(Duration::from_secs(0), rx_rdg.recv()).await, + Ok(Some(device::Value::Int(1))) + )); + + // A client sends a different setting (2) while we're still + // waiting out the envelope for 1. + let (os_tx, mut os_rx) = oneshot::channel(); + assert!(matches!(tx_set.send((2.into(), os_tx)).await, Ok(()))); + + // The driver is handed the new target, along with the + // responder -- it's the driver's job to ack the client once + // the hardware is actually set. + assert!(matches!( + timeout(Duration::from_secs(0), sh_dev.next_setting()).await, + Ok(Some((2, Some(_)))) + )); + assert!(os_rx.try_recv().is_err()); + assert!(matches!( + timeout(Duration::from_secs(0), rx_rdg.recv()).await, + Ok(Some(device::Value::Int(2))) + )); + assert!(matches!(sh_dev.get_last(), Some(2))); + + std::mem::drop(tx_set); + } + + #[tokio::test] + async fn test_duplicate_setting_mid_envelope_acks_immediately() { + let (tx_set, mut rx_rdg, mut sh_dev) = + mk_device::(Some(1), None, Some(Duration::from_millis(200))); + + assert!(matches!( + timeout(Duration::from_secs(0), sh_dev.next_setting()).await, + Ok(Some((1, None))) + )); + assert!(matches!( + timeout(Duration::from_secs(0), rx_rdg.recv()).await, + Ok(Some(device::Value::Int(1))) + )); + + // A client re-sends the value we're already applying. + let (os_tx, mut os_rx) = oneshot::channel(); + assert!(matches!(tx_set.send((1.into(), os_tx)).await, Ok(()))); + + // It's ack'd immediately without bothering the driver again... + assert!(timeout(Duration::from_secs(0), sh_dev.next_setting()) + .await + .is_err()); + assert_eq!(os_rx.try_recv(), Ok(Ok(device::Value::Int(1)))); + + // ...but it does get re-reported to the backend. + assert!(matches!( + timeout(Duration::from_secs(0), rx_rdg.recv()).await, + Ok(Some(device::Value::Int(1))) + )); + + std::mem::drop(tx_set); + } + + #[tokio::test] + async fn test_next_setting_cancel_safe_during_unknown_trans_report() { + let (rrtx, mut rrrx) = mpsc::channel::(1); + let (srtx, srrx) = mpsc::channel(20); + + // Pre-fill the backend channel so the pending report blocks. + assert!(matches!(rrtx.send(device::Value::Int(0)).await, Ok(()))); + + let mut sh_dev: OverridableDevice = + OverridableDevice::new(MockReporter(rrtx), srrx, None, None, None); + + let (os_tx, mut os_rx) = oneshot::channel(); + assert!(matches!(srtx.send((1.into(), os_tx)).await, Ok(()))); + + let waker = noop_waker(); + let mut cx = Context::from_waker(&waker); + + // First poll: Unknown -> UnknownTrans. The setting is consumed + // from the stream, but no report has been attempted yet. + { + let fut = sh_dev.next_setting(); + tokio::pin!(fut); + assert!(fut.poll(&mut cx).is_pending()); + } + assert!(matches!(&sh_dev.state, State::UnknownTrans { .. })); + + // Second poll: tries to report, but the backend channel is + // full, so this must return `Pending` without mutating state + // any further. Dropping this future (end of the block) + // simulates a cancellation. + { + let fut = sh_dev.next_setting(); + tokio::pin!(fut); + assert!(fut.poll(&mut cx).is_pending()); + } + assert!(matches!(&sh_dev.state, State::UnknownTrans { .. })); + assert!(os_rx.try_recv().is_err()); + + // Drain the dummy value so the retried report can proceed. + assert_eq!(rrrx.try_recv(), Ok(device::Value::Int(0))); + + // Retry: the setting is handed to the driver and exactly one + // (correct) value was reported -- nothing lost, nothing + // duplicated. + assert!(matches!( + timeout(Duration::from_secs(0), sh_dev.next_setting()).await, + Ok(Some((1, Some(_)))) + )); + assert_eq!(rrrx.try_recv(), Ok(device::Value::Int(1))); + assert_eq!(rrrx.try_recv(), Err(mpsc::error::TryRecvError::Empty)); + + std::mem::drop(srtx); + } + + #[tokio::test] + async fn test_next_setting_cancel_safe_during_applying_trans_report() { + let (rrtx, mut rrrx) = mpsc::channel::(1); + let (srtx, srrx) = mpsc::channel(20); + + assert!(matches!(rrtx.send(device::Value::Int(0)).await, Ok(()))); + + let mut sh_dev: OverridableDevice = + OverridableDevice::new(MockReporter(rrtx), srrx, None, None, None); + + sh_dev.state = State::Synced { value: 1 }; + + let (os_tx, mut os_rx) = oneshot::channel(); + assert!(matches!(srtx.send((2.into(), os_tx)).await, Ok(()))); + + let waker = noop_waker(); + let mut cx = Context::from_waker(&waker); + + // First poll: Synced -> ApplyingTrans. The report is attempted + // inside `ApplyingTrans`, so it hasn't happened yet. + { + let fut = sh_dev.next_setting(); + tokio::pin!(fut); + assert!(fut.poll(&mut cx).is_pending()); + } + assert!(matches!(&sh_dev.state, State::ApplyingTrans { .. })); + + // Second poll: the report is attempted, but the channel is + // full. Dropping this future simulates a cancellation; state + // must be untouched. + { + let fut = sh_dev.next_setting(); + tokio::pin!(fut); + assert!(fut.poll(&mut cx).is_pending()); + } + assert!(matches!(&sh_dev.state, State::ApplyingTrans { .. })); + assert!(os_rx.try_recv().is_err()); + + assert_eq!(rrrx.try_recv(), Ok(device::Value::Int(0))); + + assert!(matches!( + timeout(Duration::from_secs(0), sh_dev.next_setting()).await, + Ok(Some((2, Some(_)))) + )); + assert_eq!(rrrx.try_recv(), Ok(device::Value::Int(2))); + assert_eq!(rrrx.try_recv(), Err(mpsc::error::TryRecvError::Empty)); + + std::mem::drop(srtx); + } + + #[tokio::test] + async fn test_next_setting_cancel_safe_during_envelope_select() { + let (rrtx, _rrrx) = mpsc::channel::(20); + let (srtx, srrx) = mpsc::channel(20); + let mut sh_dev: OverridableDevice = + OverridableDevice::new( + MockReporter(rrtx), + srrx, + None, + None, + Some(Duration::from_millis(100)), + ); + + let deadline = tokio::time::Instant::now() + Duration::from_millis(100); + + sh_dev.state = State::Applying { + setting: true, + last_seen: true, + deadline, + needs_reaffirm: false, + }; + + // Nothing is ready yet (no incoming setting, envelope not + // elapsed). Poll once and drop the future mid-`select!` to + // simulate a cancellation. + let waker = noop_waker(); + let mut cx = Context::from_waker(&waker); + { + let fut = sh_dev.next_setting(); + tokio::pin!(fut); + assert!(fut.poll(&mut cx).is_pending()); + } + + // The deadline (and the rest of the state) must be untouched. + assert!(matches!( + &sh_dev.state, + State::Applying { deadline: d, .. } if *d == deadline + )); + + // Re-poll: it must still resolve against the *original* + // deadline (proving the cancellation didn't reset the timer), + // eventually committing to `Synced`. + assert!(timeout(Duration::from_millis(300), sh_dev.next_setting()) + .await + .is_err()); + assert!(matches!(&sh_dev.state, State::Synced { value: true })); + + std::mem::drop(srtx); + } + + #[tokio::test] + async fn test_report_update_cancel_safe() { + let (rrtx, mut rrrx) = mpsc::channel::(1); + let (_srtx, srrx) = mpsc::channel(20); + + assert!(matches!(rrtx.send(device::Value::Int(0)).await, Ok(()))); + + let mut sh_dev: OverridableDevice = + OverridableDevice::new(MockReporter(rrtx), srrx, None, None, None); + + sh_dev.state = State::Synced { value: 1 }; + + let waker = noop_waker(); + let mut cx = Context::from_waker(&waker); + + // The channel is full, so the report inside `report_update()` + // can't complete. Dropping this future simulates a + // cancellation; state must stay untouched. + { + let fut = sh_dev.report_update(2); + tokio::pin!(fut); + assert!(fut.poll(&mut cx).is_pending()); + } + assert!(matches!(&sh_dev.state, State::Synced { value: 1 })); + + // Drain the dummy and retry -- the end state must match an + // uninterrupted call, with exactly one reported value. + assert_eq!(rrrx.try_recv(), Ok(device::Value::Int(0))); + sh_dev.report_update(2).await; + + assert!(matches!( + &sh_dev.state, + State::Overridden { + setting: 1, + r#override: 2, + .. + } + )); + assert_eq!(rrrx.try_recv(), Ok(device::Value::Int(2))); + assert_eq!(rrrx.try_recv(), Err(mpsc::error::TryRecvError::Empty)); + } } diff --git a/drmem-api/src/types/device/value.rs b/drmem-api/src/types/device/value.rs index 4e69858..6231bb3 100644 --- a/drmem-api/src/types/device/value.rs +++ b/drmem-api/src/types/device/value.rs @@ -280,6 +280,28 @@ impl ReadCompat for palette::LinSrgba {} impl ReadWriteCompat for palette::LinSrgba {} +impl From for Value { + fn from(value: ColorType) -> Self { + Value::Color(value) + } +} + +impl TryFrom for ColorType { + type Error = Error; + + fn try_from(value: Value) -> Result { + if let Value::Color(color) = value { + Ok(color) + } else { + Err(Error::TypeError) + } + } +} + +impl ReadCompat for ColorType {} + +impl ReadWriteCompat for ColorType {} + // Parses a color from a string. The only forms currently supported // are "#RRGGBB" and "#RRGGBBAA" where the red, green, blue, and alpha // portions are two hex digits. Even though this function takes a diff --git a/drmemd/Cargo.toml b/drmemd/Cargo.toml index 8b47b05..a389b62 100644 --- a/drmemd/Cargo.toml +++ b/drmemd/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "drmemd" -version = "0.7.1" +version = "0.8.0" authors = ["Rich Neswold "] edition = "2021" description = "Main process of the DrMem control system"