diff --git a/.justfile b/.justfile index c1d9b251..5e9021ee 100644 --- a/.justfile +++ b/.justfile @@ -1,7 +1,7 @@ # This section has all the testing commands. The first rule is the # default and it tests every subcrate. -test-all: test-api _test-drivers test-drmemd +test-all: format test-api _test-drivers test-drmemd @echo "All tests passed successfuly!" test-api: @@ -40,10 +40,10 @@ _test-redis-graphql: test-api test-drmemd: _test-simple _test-simple-graphql _test-redis-graphql -release : test-all - @echo "Building release" - time nice cargo build \ - --release \ +build mode="dev": + @echo "Building {{ mode }}" + nice cargo build \ + {{ if mode == "rel" { "--release" } else { "" } }} \ --features simple-backend,graphql,all-drivers # This section has the targets for checking the syntax and @@ -70,6 +70,11 @@ check: _check-simple _check-simple-graphql _check-redis-graphql publish: test-all @echo "DrMem Project published successfully!" +# Formats the project. + +format: + nice cargo fmt --all + # Local variables: # mode: makefile # End: diff --git a/Cargo.lock b/Cargo.lock index 84618a31..625ca693 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -189,18 +189,18 @@ dependencies = [ [[package]] name = "clap" -version = "4.5.53" +version = "4.5.54" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c9e340e012a1bf4935f5282ed1436d1489548e8f72308207ea5df0e23d2d03f8" +checksum = "c6e6ff9dcd79cff5cd969a17a545d79e84ab086e444102a591e288a8aa3ce394" dependencies = [ "clap_builder", ] [[package]] name = "clap_builder" -version = "4.5.53" +version = "4.5.54" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d76b5d13eaa18c901fd2f7fca939fefe3a0727a953561fefdf3b2922b8569d00" +checksum = "fa42cf4d2b7a41bc8f663a7cab4031ebafa1bf3875705bfaf8466dc60ab52c00" dependencies = [ "anstyle", "clap_lex", @@ -364,7 +364,7 @@ dependencies = [ [[package]] name = "drmem-api" -version = "0.6.0" +version = "0.6.1" dependencies = [ "chrono", "noop-waker", @@ -402,7 +402,7 @@ dependencies = [ [[package]] name = "drmem-drv-tplink" -version = "0.6.0" +version = "0.6.1" dependencies = [ "drmem-api", "futures", @@ -430,7 +430,7 @@ dependencies = [ [[package]] name = "drmemd" -version = "0.6.1" +version = "0.6.2" dependencies = [ "cfgrammar", "chrono", @@ -1047,9 +1047,9 @@ checksum = "bbd2bcb4c963f2ddae06a2efc7e9f3591312473c50c6685e1f298068316e66fe" [[package]] name = "libc" -version = "0.2.178" +version = "0.2.179" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "37c93d8daa9d8a012fd8ab92f088405fb202ea0b6ab73ee2482ae66af4f42091" +checksum = "c5a2d376baa530d1238d133232d15e239abad80d05838b4b59354e5268af431f" [[package]] name = "libm" @@ -1408,9 +1408,9 @@ dependencies = [ [[package]] name = "proc-macro2" -version = "1.0.104" +version = "1.0.105" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9695f8df41bb4f3d222c95a67532365f569318332d03d5f3f67f37b20e6ebdf0" +checksum = "535d180e0ecab6268a3e718bb9fd44db66bbbc256257165fc699dadf70d16fe7" dependencies = [ "unicode-ident", ] @@ -1433,9 +1433,9 @@ dependencies = [ [[package]] name = "quote" -version = "1.0.42" +version = "1.0.43" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a338cc41d27e6cc6dce6cefc13a0729dfbb81c262b1f519331575dd80ef3067f" +checksum = "dc74d9a594b72ae6656596548f56f667211f8a97b3d4c3d467150794690dc40a" dependencies = [ "proc-macro2", ] @@ -1737,9 +1737,9 @@ dependencies = [ [[package]] name = "serde_json" -version = "1.0.148" +version = "1.0.149" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3084b546a1dd6289475996f182a22aba973866ea8e8b02c51d9f46b1336a22da" +checksum = "83fc039473c5595ace860d8c4fafa220ff474b3fc6bfdb4293327f1a37e94d86" dependencies = [ "itoa", "memchr", @@ -1888,9 +1888,9 @@ checksum = "13c2bddecc57b384dee18652358fb23172facb8a2c51ccc10d74c157bdea3292" [[package]] name = "syn" -version = "2.0.112" +version = "2.0.114" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "21f182278bf2d2bcb3c88b1b08a37df029d71ce3d3ae26168e3c653b213b99d4" +checksum = "d4d107df263a3013ef9b1879b0df87d706ff80f65a86ea879bd9c31f9b307c2a" dependencies = [ "proc-macro2", "quote", @@ -2024,9 +2024,9 @@ checksum = "1f3ccbac311fea05f86f61904b462b55fb3df8837a366dfc601a0161d0532f20" [[package]] name = "tokio" -version = "1.48.0" +version = "1.49.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ff360e02eab121e0bc37a2d3b4d4dc622e6eda3a8e5253d5435ecf5bd4c68408" +checksum = "72a2903cd7736441aac9df9d7688bd0ce48edccaadf181c3b90be801e81d3d86" dependencies = [ "bytes", "libc", @@ -2071,9 +2071,9 @@ dependencies = [ [[package]] name = "tokio-stream" -version = "0.1.17" +version = "0.1.18" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "eca58d7bba4a75707817a2c44174253f9236b2d5fbd055602e9d5c07c139a047" +checksum = "32da49809aab5c3bc678af03902d4ccddea2a87d028d86392a4b1560c6906c70" dependencies = [ "futures-core", "pin-project-lite", @@ -2095,9 +2095,9 @@ dependencies = [ [[package]] name = "tokio-util" -version = "0.7.17" +version = "0.7.18" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2efa149fe76073d6e8fd97ef4f4eca7b67f599660115591483572e406e165594" +checksum = "9ae9cec805b01e8fc3fd2fe289f89149a9b66dd16786abd8b19cfa7b48cb0098" dependencies = [ "bytes", "futures-core", @@ -2223,9 +2223,9 @@ checksum = "562d481066bde0658276a35467c4af00bdc6ee726305698a55b86e61d7ad82bb" [[package]] name = "unicase" -version = "2.8.1" +version = "2.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "75b844d17643ee918803943289730bec8aac480150456169e647ed0b576ba539" +checksum = "dbc4bc3a9f746d862c45cb89d705aa10f187bb96c76001afab07a0d35ce60142" [[package]] name = "unicode-bidi" @@ -2268,9 +2268,9 @@ checksum = "8ecb6da28b8a351d773b68d5825ac39017e680750f980f3a1a85cd8dd28a47c1" [[package]] name = "url" -version = "2.5.7" +version = "2.5.8" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "08bc136a29a3d1758e07a9cca267be308aeebf5cfd5a10f3f67ab2097683ef5b" +checksum = "ff67a8a4397373c3ef660812acab3268222035010ab8680ec4215f38ba3d0eed" dependencies = [ "form_urlencoded", "idna 1.1.0", @@ -2805,18 +2805,18 @@ dependencies = [ [[package]] name = "zerocopy" -version = "0.8.31" +version = "0.8.32" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fd74ec98b9250adb3ca554bdde269adf631549f51d8a8f8f0a10b50f1cb298c3" +checksum = "1fabae64378cb18147bb18bca364e63bdbe72a0ffe4adf0addfec8aa166b2c56" dependencies = [ "zerocopy-derive", ] [[package]] name = "zerocopy-derive" -version = "0.8.31" +version = "0.8.32" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d8a8d209fdf45cf5138cbb5a506f6b52522a25afccc534d1475dad8e31105c6a" +checksum = "c9c2d862265a8bb4471d87e033e730f536e2a285cc7cb05dbce09a2a97075f90" dependencies = [ "proc-macro2", "quote", @@ -2885,6 +2885,6 @@ dependencies = [ [[package]] name = "zmij" -version = "1.0.7" +version = "1.0.12" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "de9211a9f64b825911bdf0240f58b7a8dac217fe260fc61f080a07f61372fbd5" +checksum = "2fc5a66a20078bf1251bde995aa2fdcc4b800c70b5d92dd2c62abc5c60f679f8" diff --git a/drivers/drmem-drv-ntp/src/lib.rs b/drivers/drmem-drv-ntp/src/lib.rs index 211e1b06..30f3e475 100644 --- a/drivers/drmem-drv-ntp/src/lib.rs +++ b/drivers/drmem-drv-ntp/src/lib.rs @@ -214,7 +214,7 @@ impl Instance { Err(e) => error!("couldn't receive data -> {}", e), } }, - _ = tokio::time::sleep(std::time::Duration::from_millis(1_000)) => { + _ = tokio::time::sleep(Duration::from_millis(1_000)) => { warn!("timed-out waiting for reply to \"get synced host\" request") } } @@ -346,7 +346,7 @@ impl Instance { } } }, - _ = tokio::time::sleep(std::time::Duration::from_millis(1_000)) => { + _ = tokio::time::sleep(Duration::from_millis(1_000)) => { warn!("timed-out waiting for reply to \"get host info\" request") } } @@ -366,6 +366,7 @@ impl driver::Registrator for Devices { fn register_devices<'a>( core: &'a mut driver::RequestChan, _: &DriverConfig, + _override_timeout: Option, max_history: Option, ) -> impl Future> + Send + 'a { // It's safe to use `.unwrap()` for these names because, in a diff --git a/drivers/drmem-drv-sump/src/lib.rs b/drivers/drmem-drv-sump/src/lib.rs index 422ad0b7..f274baae 100644 --- a/drivers/drmem-drv-sump/src/lib.rs +++ b/drivers/drmem-drv-sump/src/lib.rs @@ -12,7 +12,7 @@ use tokio::{ tcp::{OwnedReadHalf, OwnedWriteHalf}, TcpStream, }, - time, + time::Duration, }; use tracing::{debug, error, info, warn, Span}; @@ -225,8 +225,8 @@ impl Instance { use socket2::{Domain, Socket, TcpKeepalive, Type}; let keepalive = TcpKeepalive::new() - .with_time(time::Duration::from_secs(5)) - .with_interval(time::Duration::from_secs(5)); + .with_time(Duration::from_secs(5)) + .with_interval(Duration::from_secs(5)); let socket = Socket::new(Domain::IPV4, Type::STREAM, None) .expect("couldn't create socket"); @@ -236,7 +236,7 @@ impl Instance { match socket.connect_timeout( &>::into(*addr), - time::Duration::from_millis(100), + Duration::from_millis(100), ) { Ok(()) => { info!("connected"); @@ -284,6 +284,7 @@ impl driver::Registrator for Devices { fn register_devices<'a>( core: &'a mut driver::RequestChan, _: &DriverConfig, + _override_timeout: Option, max_history: Option, ) -> impl Future> + Send + 'a { let service_name = "service".parse::().unwrap(); diff --git a/drivers/drmem-drv-tplink/Cargo.toml b/drivers/drmem-drv-tplink/Cargo.toml index dca689f2..7a629d1c 100644 --- a/drivers/drmem-drv-tplink/Cargo.toml +++ b/drivers/drmem-drv-tplink/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "drmem-drv-tplink" -version = "0.6.0" +version = "0.6.1" authors = ["Rich Neswold "] edition = "2021" homepage = "https://github.com/DrMemCS/drmem" @@ -42,7 +42,7 @@ serde_json.workspace = true serde_json.default-features = false serde_json.features = ["std"] -drmem-api = { path = "../../drmem-api", version = "0.6" } +drmem-api = { path = "../../drmem-api", version = "0.6.1" } [dev-dependencies] tokio.workspace = true diff --git a/drivers/drmem-drv-tplink/README.md b/drivers/drmem-drv-tplink/README.md index 63807433..676d2fba 100644 --- a/drivers/drmem-drv-tplink/README.md +++ b/drivers/drmem-drv-tplink/README.md @@ -14,6 +14,7 @@ This table contains products that are known to work with this driver. | Device Model | Vendor | Description | |--------------|--------|---------------| +| HS103 | Kasa | On/off outlet | | HS220 | Kasa | Dimmer switch. Note: it takes around 200ms for this module to respond to a command. In some instances, it took ~1s! So don't try controlling it rapidly. | ## Configuration @@ -23,16 +24,26 @@ The driver needs to know where to access the device. - `addr` is a string containing the host name, or IP address, and port number of the TP-Link device (in **"hostname:#"** or **"\#.#.#.#:#"** format.) The port is almost always 9999. +- `type` is a string that indicates what type of device to use. It can + be "switch", "outlet", or "dimmer". ## Devices -The driver creates these devices: +The driver creates these devices for switch devices: | Base Name | Type | Units | Comment | |--------------|----------|-------|----------------------------------------| | `error` | bool, RO | | If true, there is an error communicating with the device. | -| `brightness` | f64 , RW | % | Accepts 0 - 100 for percent brightness. If the value is out of range, it will be brought back in range. | -| `led` | bool, RW | | `true` and `false` turn the LED indicator on and off, respectively. | +| `state` | bool, RW | | `true` and `false` turn it on and off, respectively. | +| `indicator` | bool, RW | | `true` and `false` turn the LED indicator on and off, respectively. | + +The driver creates these devices for dimmer devices: + +| Base Name | Type | Units | Comment | +|--------------|----------|-------|----------------------------------------| +| `error` | bool, RO | | If true, there is an error communicating with the device. | +| `brightness` | f64, RW | % | Accepts 0 - 100 for percent brightness. If the value is out of range, it will be brought back in range. | +| `indicator` | bool, RW | | `true` and `false` turn the LED indicator on and off, respectively. | ## History diff --git a/drivers/drmem-drv-tplink/src/lib.rs b/drivers/drmem-drv-tplink/src/lib.rs index c1c56daf..7f6aea44 100644 --- a/drivers/drmem-drv-tplink/src/lib.rs +++ b/drivers/drmem-drv-tplink/src/lib.rs @@ -30,7 +30,9 @@ // Received: {"system":{"set_bright":{"err_code":-2,"err_msg":"member not support"}}} use drmem_api::{ - driver::{self, classes, DriverConfig}, + driver::{ + self, classes, DriverConfig, Registrator, RequestChan, ResettableState, + }, Error, Result, }; use futures::{Future, FutureExt}; @@ -39,18 +41,99 @@ use std::net::SocketAddrV4; use tokio::{ io::{AsyncReadExt, AsyncWriteExt}, net::TcpStream, - time, + task::JoinHandle, + time::{self, Duration}, }; -use tracing::{debug, error, warn, Span}; +use tracing::{debug, error, info, Span}; -mod tplink_api; +mod tplink; const BUF_TOTAL: usize = 4_096; +// Holds the state of the device received by an info request. Switch +// and outlet devices will always return None for the brightness. +struct DeviceState { + indicator: Option, + brightness: Option, + relay: Option, +} + +// 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 DevType { + Switch(classes::Switch), + Dimmer(classes::Dimmer), +} + +// A set of devices must be resettable (in case the device gets +// rebooted.) This implementation simply resets the devices in the set +// used by the driver instance. +impl ResettableState for DevType { + fn reset_state(&mut self) { + match self { + DevType::Switch(dev) => { + dev.state.reset_state(); + dev.indicator.reset_state(); + } + DevType::Dimmer(dev) => { + dev.brightness.reset_state(); + dev.indicator.reset_state(); + } + } + } +} + +impl Registrator for DevType { + // Defines the registration interface for the device set. + fn register_devices<'a>( + drc: &'a mut RequestChan, + cfg: &'a DriverConfig, + override_timeout: Option, + max_history: Option, + ) -> impl Future> + Send + 'a { + async move { + match cfg.get("type") { + Some(toml::value::Value::String(dtype)) => match dtype.as_str() + { + "outlet" | "switch" => Ok(DevType::Switch( + classes::Switch::register_devices( + drc, + cfg, + override_timeout, + max_history, + ) + .await?, + )), + "dimmer" => Ok(DevType::Dimmer( + classes::Dimmer::register_devices( + drc, + cfg, + override_timeout, + max_history, + ) + .await?, + )), + _ => Err(Error::ConfigError(String::from( + "'type' must be \"dimmer\", \"outlet\", or \"switch\"", + ))), + }, + Some(_) => Err(Error::ConfigError(String::from( + "'type' config parameter should be a string", + ))), + None => Err(Error::ConfigError(String::from( + "missing 'type' parameter in config", + ))), + } + } + } +} + pub struct Instance { addr: SocketAddrV4, reported_error: Option, buf: [u8; BUF_TOTAL], + poll_timeout: Duration, } impl Instance { @@ -83,11 +166,11 @@ impl Instance { } } - // Attempts to read a `tplink_api::Reply` type from the socket. + // Attempts to read a `tplink::Reply` type from the socket. // All replies have a 4-byte length header so we know how much // data to read. - async fn read_reply(&mut self, s: &mut R) -> Result + async fn read_reply(&mut self, s: &mut R) -> Result where R: AsyncReadExt + std::marker::Unpin, { @@ -100,7 +183,7 @@ impl Instance { if let Err(e) = s.read_exact(filled).await { Err(Error::MissingPeer(e.to_string())) } else { - tplink_api::Reply::decode(filled).ok_or_else(|| { + tplink::Reply::decode(filled).ok_or_else(|| { Error::ParseError(format!( "bad reply : {}", String::from_utf8_lossy(filled) @@ -119,7 +202,7 @@ impl Instance { // Attempts to send a command to the socket. - async fn send_cmd(s: &mut S, cmd: tplink_api::Cmd) -> Result<()> + async fn send_cmd(s: &mut S, cmd: tplink::Cmd) -> Result<()> where S: AsyncWriteExt + std::marker::Unpin, { @@ -135,7 +218,7 @@ impl Instance { Err(e) => Err(ERR_F(e)) } } - _ = time::sleep(time::Duration::from_millis(500)) => + _ = time::sleep(Duration::from_millis(500)) => Err(Error::TimeoutError) } } @@ -147,8 +230,8 @@ impl Instance { &mut self, rx: &mut R, tx: &mut S, - cmd: tplink_api::Cmd, - ) -> Result + cmd: tplink::Cmd, + ) -> Result where R: AsyncReadExt + std::marker::Unpin, S: AsyncWriteExt + std::marker::Unpin, @@ -160,7 +243,7 @@ impl Instance { #[rustfmt::skip] tokio::select! { result = self.read_reply(rx) => result, - _ = time::sleep(time::Duration::from_millis(500)) => + _ = time::sleep(Duration::from_millis(500)) => Err(Error::TimeoutError) } } @@ -177,11 +260,14 @@ impl Instance { s: &mut TcpStream, v: bool, ) -> Result<()> { - use tplink_api::{active_cmd, ErrorStatus, Reply}; + use tplink::{Cmd, ErrorStatus, Reply}; let (mut rx, mut tx) = s.split(); - match self.rpc(&mut rx, &mut tx, active_cmd(v as u8)).await? { + match self + .rpc(&mut rx, &mut tx, Cmd::mk_active_cmd(v as u8)) + .await? + { Reply::System { set_relay_state: Some(ErrorStatus { err_code: 0, .. }), .. @@ -209,14 +295,14 @@ impl Instance { s: &mut TcpStream, v: bool, ) -> Result<()> { - use tplink_api::{led_cmd, ErrorStatus, Reply}; + use tplink::{Cmd, ErrorStatus, Reply}; let (mut rx, mut tx) = s.split(); // Send the request and receive the reply. Use pattern // matching to determine the return value of the function. - match self.rpc(&mut rx, &mut tx, led_cmd(v)).await? { + match self.rpc(&mut rx, &mut tx, Cmd::mk_led_cmd(v)).await? { Reply::System { set_led_off: Some(ErrorStatus { err_code: 0, .. }), .. @@ -239,26 +325,42 @@ impl Instance { // Retrieves info. - async fn info_rpc(&mut self, s: &mut TcpStream) -> Result<(bool, u8)> { - use tplink_api::{info_cmd, Reply}; + async fn info_rpc(&mut self, s: &mut TcpStream) -> Result { + use tplink::{Cmd, Reply}; let (mut rx, mut tx) = s.split(); // Send the request and receive the reply. Use pattern // matching to determine the return value of the function. - match self.rpc(&mut rx, &mut tx, info_cmd()).await? { + match self.rpc(&mut rx, &mut tx, Cmd::mk_info_cmd()).await? { Reply::System { get_sysinfo: Some(info), .. } => { - let led = info.led_off.unwrap_or(1) == 0; + let indicator = info.led_off.unwrap_or(1) == 0; match (info.relay_state.map(|v| v != 0), info.brightness) { - (None, None) => Ok((led, 0)), - (None, Some(br)) => Ok((led, br)), - (Some(false), _) => Ok((led, 0)), - (Some(true), br) => Ok((led, br.unwrap_or(100))), + (None, None) => Ok(DeviceState { + indicator: Some(indicator), + relay: None, + brightness: Some(0.0), + }), + (None, Some(_)) => Ok(DeviceState { + indicator: Some(indicator), + relay: None, + brightness: Some(0.0), + }), + (rs @ Some(false), _) => Ok(DeviceState { + indicator: Some(indicator), + relay: rs, + brightness: Some(0.0), + }), + (rs @ Some(true), br) => Ok(DeviceState { + indicator: Some(indicator), + relay: rs, + brightness: Some(br.unwrap_or(100).into()), + }), } } @@ -273,11 +375,14 @@ impl Instance { // argument. async fn brightness_rpc(&mut self, s: &mut TcpStream, v: u8) -> Result<()> { - use tplink_api::{brightness_cmd, ErrorStatus, Reply}; + use tplink::{Cmd, ErrorStatus, Reply}; let (mut rx, mut tx) = s.split(); - match self.rpc(&mut rx, &mut tx, brightness_cmd(v)).await? { + match self + .rpc(&mut rx, &mut tx, Cmd::mk_brightness_cmd(v)) + .await? + { Reply::Dimmer { set_brightness: Some(ErrorStatus { err_code: 0, .. }), .. @@ -321,25 +426,44 @@ impl Instance { // Connects to the address. Sets a timeout of 1 second for the // connection. - async fn connect(addr: &SocketAddrV4) -> Result { - use tokio::net::TcpSocket; + async fn connect(addr: SocketAddrV4) -> Result { + use tokio::{net::TcpSocket, time}; - let fut = time::timeout(time::Duration::from_secs(1), async { + loop { match TcpSocket::new_v4() { Ok(s) => { - s.set_recv_buffer_size((BUF_TOTAL * 2) as u32)?; - s.set_nodelay(true)?; - - Ok(s.connect((*addr).into()).await?) + s.set_recv_buffer_size((BUF_TOTAL * 2) as u32).map_err( + |e| { + Error::OperationError(format!( + "couldn't set socket's buffer size -- {}", + e + )) + }, + )?; + s.set_nodelay(true).map_err(|e| { + Error::OperationError(format!( + "couldn't set NO DELAY on socket -- {}", + e + )) + })?; + + if let Ok(Ok(s)) = time::timeout( + Duration::from_secs(2), + s.connect(addr.into()), + ) + .await + { + return Ok(s); + } + time::sleep(time::Duration::from_secs(60)).await + } + Err(e) => { + return Err(Error::OperationError(format!( + "couldn't create socket -- {}", + e + ))) } - Err(e) => Err(e), } - }); - - match fut.await { - Ok(Ok(s)) => Ok(s), - Ok(Err(e)) => Err(Error::MissingPeer(e.to_string())), - Err(_) => Err(Error::MissingPeer("timeout".into())), } } @@ -349,9 +473,8 @@ impl Instance { &mut self, s: &'a mut TcpStream, v: f64, - reply: driver::SettingReply, - report: &'a mut driver::ReadWriteDevice, - ) -> Result> { + reply: Option>, + ) -> Result<()> { if !v.is_nan() { // Clip incoming settings to the range 0.0..=100.0. Handle // infinities, too. @@ -366,25 +489,22 @@ impl Instance { // Send an OK reply to the client with the updated value. - reply(Ok(v)); + if let Some(reply) = reply { + reply(Ok(v)); + } // Always log incoming settings. Let the client know there // was a successful setting, and include the value that // was used. - match self.set_brightness(s, v).await { - Ok(()) => { - report.report_update(v).await; - Ok(Some(v)) - } - Err(e) => { - error!("setting brightness : {}", &e); - Err(e) - } - } + self.set_brightness(s, v).await } else { - reply(Err(Error::InvArgument("device doesn't accept NaN".into()))); - Ok(None) + if let Some(reply) = reply { + reply(Err(Error::InvArgument( + "device doesn't accept NaN".into(), + ))); + } + Ok(()) } } @@ -394,160 +514,339 @@ impl Instance { &mut self, s: &'a mut TcpStream, v: bool, - reply: driver::SettingReply, - report: &'a mut driver::ReadWriteDevice, + reply: Option>, ) -> Result<()> { - reply(Ok(v)); - match self.led_state_rpc(s, v).await { - Ok(()) => { - report.report_update(v).await; - Ok(()) - } - Err(e) => { - error!("setting LED : {}", &e); - Err(e) - } + if let Some(reply) = reply { + reply(Ok(v)); } + self.led_state_rpc(s, v).await } // Checks to see if the current error state ('value') matches the // previosuly reported error state. If not, it saves the current // state and sends the updated value to the backend. - async fn sync_error_state( - &mut self, - report: &mut driver::ReadOnlyDevice, - value: bool, - ) { + async fn sync_error_state(&mut self, dev: &mut DevType, value: bool) { if self.reported_error != Some(value) { self.reported_error = Some(value); - report.report_update(value).await; + match dev { + DevType::Switch(classes::Switch { error, .. }) => { + error.report_update(value).await + } + DevType::Dimmer(classes::Dimmer { error, .. }) => { + error.report_update(value).await + } + } } } - async fn main_loop( - &mut self, - s: &mut TcpStream, - devices: &mut ::HardwareType, - ) { - // Create a 5-second interval timer which will be used to poll - // the device to see if its state was changed by some outside - // mechanism. + async fn pause_as_dimmer(&mut self, dev: &mut classes::Dimmer) { + let classes::Dimmer { + brightness: ref mut d_b, + indicator: ref mut d_i, + .. + } = dev; - let mut timer = - tokio::time::interval(tokio::time::Duration::from_secs(5)); - let mut current_led = false; - let mut current_brightness = -1.0f64; + let timeout = time::sleep(Duration::from_secs(60)); + tokio::pin!(timeout); + + loop { + #[rustfmt::skip] + tokio::select! { + _ = &mut timeout => { + break; + } + Some((v, reply)) = d_b.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)); + } + } + } + } + } - // Main loop of the driver. This loop never ends. + async fn pause_as_switch(&mut self, dev: &mut classes::Switch) { + let classes::Switch { + state: ref mut d_r, + indicator: ref mut d_i, + .. + } = dev; - 'main: loop { - self.sync_error_state(&mut devices.error, false).await; + let timeout = time::sleep(Duration::from_secs(60)); + tokio::pin!(timeout); - // Get mutable references to the setting channels. + loop { + #[rustfmt::skip] + tokio::select! { + _ = &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)); + } + } + } + } + } - let classes::Dimmer { - brightness: ref mut d_b, - indicator: ref mut d_l, - .. - } = *devices; + async fn manage_connect_as_switch( + &mut self, + task: JoinHandle>, + dev: &mut classes::Switch, + ) -> Result { + let classes::Switch { + state: ref mut d_r, + indicator: ref mut d_i, + .. + } = dev; - // Now wait for one of three events to occur. + tokio::pin!(task); + loop { #[rustfmt::skip] tokio::select! { - // If the timer tick expires, it's time to get the - // latest state of the device. Since external apps can - // modify the device outside of DrMem's control, we - // have to periodically poll it to stay in sync. - - _ = timer.tick() => { - if let Ok((led, br)) = self.info_rpc(s).await { - let br = br as f64; - - // If the LED state has changed outside of the - // driver, update the local state. - - if current_led != led { - debug!("external LED update: {}", led); - current_led = led; - d_l.report_update(led).await; - } + task_result = &mut task => { + match task_result { + Ok(result) => return result, + Err(e) => { + return Err(Error::OperationError(format!( + "driver panicked while connecting -- {}", + e + ))) + } + } + } + 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)); + } + } + } + } + } - // If the brightness state has changed outside - // of the driver, update the local state. + async fn manage_connect_as_dimmer( + &mut self, + task: JoinHandle>, + dev: &mut classes::Dimmer, + ) -> Result { + let classes::Dimmer { + brightness: ref mut d_b, + indicator: ref mut d_i, + .. + } = dev; - if current_brightness != br { - debug!("external brightness update: {}", br); - current_brightness = br; - devices.brightness.report_update(br).await; - } - } else { - break 'main - } + tokio::pin!(task); + + loop { + #[rustfmt::skip] + tokio::select! { + task_result = &mut task => { + match task_result { + Ok(result) => return result, + Err(e) => { + return Err(Error::OperationError(format!( + "driver panicked while connecting -- {}", + e + ))) + } + } + } + Some((v, reply)) = d_b.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)); + } + } + } + } + } - // Handle settings to the brightness device. + async fn manage_switch<'a>( + &mut self, + s: &'a mut TcpStream, + dev: &mut classes::Switch, + ) -> bool { + // Get mutable references to the setting channels. - Some((v, reply)) = d_b.next_setting() => { + let classes::Switch { + state: ref mut d_r, + indicator: ref mut d_i, + .. + } = dev; - // If the settings matches the current state, then - // don't actually control the hardware. + // Now wait for one of three events to occur. - if current_brightness != v { - match self.handle_brightness_setting( - s, v, reply, d_b - ).await { - Ok(Some(v)) => current_brightness = v, - Ok(None) => (), - Err(_) => break 'main - } - } else { - debug!("don't need to apply brightness setting"); + #[rustfmt::skip] + tokio::select! { + // If our poll timeout expires, we need to request the + // current state of the hardware (because it could be + // changed by external apps or a person using the manual + // switch.) + + _ = tokio::time::sleep(self.poll_timeout) => { + + // Reset the next timeout to be 5 seconds. + + self.poll_timeout = Duration::from_secs(5); - // Hardware wasn't updated, but we still need - // to log the setting and return a reply to - // the client. + // Request the hardware state. - d_b.report_update(v).await; - reply(Ok(v)) - } + 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 + } + } + Err(e) => { + error!("error getting state -- {e}"); + return false + } } + } - // Handle settings to the LED indicator device. + // Handle settings to the brightness device. - Some((v, reply)) = d_l.next_setting() => { - debug!("led setting -> {}", &v); + Some((v, reply)) = d_r.next_setting() => { + if let Some(reply) = reply { + reply(Ok(v)); + } + 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); + } - // If the settings matches the current state, then - // don't actually control the hardware. + // Handle settings to the LED indicator device. - if current_led != v { - if self.handle_led_setting( - s, v, reply, &mut devices.indicator - ).await == Ok(()) { - current_led = v; - } else { - break 'main - } - } else { - debug!("don't need to apply led setting"); + 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 + } + self.poll_timeout = Duration::from_secs(0); + } + } + return true; + } + + async fn manage_dimmer<'a>( + &mut self, + s: &'a mut TcpStream, + dev: &mut classes::Dimmer, + ) -> 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. + + #[rustfmt::skip] + tokio::select! { + // If our poll timeout expires, we need to request the + // current state of the hardware (because it could be + // changed by external apps or a person using the manual + // switch.) + + _ = tokio::time::sleep(self.poll_timeout) => { + + // Reset the next timeout to be 5 seconds. - // Hardware wasn't updated, but we still need - // to log the setting and return a reply to - // the client. + self.poll_timeout = Duration::from_secs(5); - d_l.report_update(v).await; - reply(Ok(v)) - } + // Request the hardware state. + + match self.info_rpc(s).await { + Ok(info) => { + if let Some(indicator) = info.indicator { + d_i.report_update(indicator).await + } + + if let Some(brightness) = info.brightness { + d_b.report_update(brightness).await + } + } + Err(e) => { + error!("error getting state -- {e}"); + return false + } } } + + // 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 + } + self.poll_timeout = Duration::from_secs(0); + } + } + return true; + } + + async fn main_loop( + &mut self, + mut s: TcpStream, + devices: &mut ::HardwareType, + ) { + self.sync_error_state(&mut *devices, false).await; + loop { + if !match &mut *devices { + DevType::Switch(dev) => self.manage_switch(&mut s, dev).await, + DevType::Dimmer(dev) => self.manage_dimmer(&mut s, dev).await, + } { + break; + } } + self.sync_error_state(&mut *devices, true).await; } } impl driver::API for Instance { - type HardwareType = classes::Dimmer; + type HardwareType = DevType; // This driver doesn't store any data in its instance; it's all // stored in local variables in the `.run()` method. @@ -562,6 +861,7 @@ impl driver::API for Instance { addr: cfg_addr?, reported_error: None, buf: [0; BUF_TOTAL], + poll_timeout: Duration::from_secs(0), })) } } @@ -573,40 +873,42 @@ impl driver::API for Instance { // span. Span::current().record("cfg", self.addr.to_string()); - + self.sync_error_state(&mut *devices, true).await; loop { - // First, connect to the device. We'll leave the TCP - // connection open so we're ready for the next - // transaction. Tests have shown that the HS220 handles - // multiple client connections. - - match Instance::connect(&self.addr).await { - Ok(mut s) => { - self.main_loop(&mut s, devices).await; + info!("connecting to hardware at {}", &self.addr); + + let conn_task = tokio::spawn(Instance::connect(self.addr)); + let conn_result = match &mut *devices { + DevType::Switch(ref mut dev) => { + self.manage_connect_as_switch(conn_task, dev).await } - Err(e) => { - warn!("couldn't connect : '{}'", e); + DevType::Dimmer(ref mut dev) => { + self.manage_connect_as_dimmer(conn_task, dev).await } - } + }; - self.sync_error_state(&mut devices.error, true).await; + match conn_result { + Ok(s) => self.main_loop(s, devices).await, + Err(e) => error!("connection failed -- {}", e), + } - // Log the error and then sleep for 10 seconds. Hopefully + // Log the error and then sleep for 60 seconds. Hopefully // the device will be available then. - tokio::time::sleep(tokio::time::Duration::from_secs(10)).await + match &mut *devices { + DevType::Switch(ref mut dev) => self.pause_as_switch(dev).await, + DevType::Dimmer(ref mut dev) => self.pause_as_dimmer(dev).await, + }; } } } #[cfg(test)] mod test { - use super::{tplink_api, Instance}; + use super::{tplink, Instance}; use crate::BUF_TOTAL; - use std::{ - io::Write, - net::{Ipv4Addr, SocketAddrV4}, - }; + use std::net::{Ipv4Addr, SocketAddrV4}; + use tokio::time::Duration; #[tokio::test] async fn test_read_reply() { @@ -618,6 +920,7 @@ mod test { addr: SocketAddrV4::new(Ipv4Addr::new(0, 0, 0, 0), 0), reported_error: None, buf: [0u8; BUF_TOTAL], + poll_timeout: Duration::from_secs(5), }; assert!(inst.read_reply(&mut &buf[0..=0]).await.is_err()); @@ -632,11 +935,8 @@ mod test { let mut buf = vec![0, 0, 0, REPLY.len() as u8]; - { - let mut wr = tplink_api::CmdWriter::create(&mut buf); - - assert_eq!(wr.write(REPLY).unwrap(), REPLY.len()); - } + buf.extend_from_slice(&REPLY[..]); + tplink::crypt::encode(&mut buf[4..]); assert!(buf.len() == 45); assert!(buf.as_slice().len() == 45); @@ -645,6 +945,7 @@ mod test { addr: SocketAddrV4::new(Ipv4Addr::new(0, 0, 0, 0), 0), reported_error: None, buf: [0u8; BUF_TOTAL], + poll_timeout: Duration::from_secs(5), }; assert!(inst.read_reply(&mut &buf[0..4]).await.is_err()); diff --git a/drivers/drmem-drv-tplink/src/tplink/cmd.rs b/drivers/drmem-drv-tplink/src/tplink/cmd.rs new file mode 100644 index 00000000..835f61b9 --- /dev/null +++ b/drivers/drmem-drv-tplink/src/tplink/cmd.rs @@ -0,0 +1,109 @@ +use crate::tplink::crypt; +use serde::Serialize; + +// This module holds the types that represent values used in the +// request fields. + +pub mod value { + use serde::Serialize; + + // Defines the internal value used by the `set_relay_state` + // command. Needs to convert to `{"state":value}`. + + #[derive(Serialize, PartialEq, Debug)] + pub struct Active { + pub state: u8, + } + + // Defines the internal value used by the `set_led_off` + // command. Needs to convert to `{"off":value}`. + + #[derive(Serialize, PartialEq, Debug)] + pub struct Led { + pub off: u8, + } + + // Defines the internal value used by the `get_sysinfo` + // command. Needs to convert to `{}`. + + #[derive(Serialize, PartialEq, Debug)] + pub struct Info {} + + // Defines the internal value used by the `Brightness` + // command. Needs to convert to `{"brightness":value}`. + + #[derive(Serialize, PartialEq, Debug)] + pub struct Brightness { + pub brightness: u8, + } +} + +#[derive(Serialize, PartialEq, Debug)] +pub enum Cmd { + #[serde(rename = "system")] + System { + #[serde(skip_serializing_if = "Option::is_none")] + set_relay_state: Option, + #[serde(skip_serializing_if = "Option::is_none")] + get_sysinfo: Option, + #[serde(skip_serializing_if = "Option::is_none")] + set_led_off: Option, + }, + + #[serde(rename = "smartlife.iot.dimmer")] + Dimmer { set_brightness: value::Brightness }, +} + +impl Cmd { + pub fn encode(&self) -> Vec { + let mut buf = Vec::with_capacity(100); + + buf.push(0u8); + buf.push(0u8); + buf.push(0u8); + buf.push(0u8); + + serde_json::to_writer(&mut buf, &self).unwrap(); + + let sz = buf.len() - 4; + + buf[0] = (sz >> 24) as u8; + buf[1] = (sz >> 16) as u8; + buf[2] = (sz >> 8) as u8; + buf[3] = sz as u8; + + crypt::encode(&mut buf[4..]); + + buf + } + + pub fn mk_active_cmd(v: u8) -> Self { + Self::System { + set_relay_state: Some(value::Active { state: v }), + get_sysinfo: None, + set_led_off: None, + } + } + + pub fn mk_brightness_cmd(v: u8) -> Self { + Self::Dimmer { + set_brightness: value::Brightness { brightness: v }, + } + } + + pub fn mk_info_cmd() -> Self { + Self::System { + set_relay_state: None, + get_sysinfo: Some(value::Info {}), + set_led_off: None, + } + } + + pub fn mk_led_cmd(v: bool) -> Self { + Self::System { + set_relay_state: None, + get_sysinfo: None, + set_led_off: Some(value::Led { off: (!v) as u8 }), + } + } +} diff --git a/drivers/drmem-drv-tplink/src/tplink/crypt.rs b/drivers/drmem-drv-tplink/src/tplink/crypt.rs new file mode 100644 index 00000000..0b7afa57 --- /dev/null +++ b/drivers/drmem-drv-tplink/src/tplink/crypt.rs @@ -0,0 +1,39 @@ +// This is the decryption algorithm used by TP-Link. + +pub fn decode(buf: &mut [u8]) { + let mut key = 171u8; + + for b in buf.iter_mut() { + let tmp = *b; + + *b ^= key; + key = tmp; + } +} + +// This is the encryption algorithm used by TP-Link. + +pub fn encode(buf: &mut [u8]) { + let mut key = 171u8; + + for b in buf.iter_mut() { + key ^= *b; + *b = key; + } +} + +#[cfg(test)] +mod test { + use super::*; + + #[test] + fn test_crypt() { + let buf = [1u8, 2u8, 3u8, 4u8, 5u8]; + let mut enc = vec![]; + + enc.extend_from_slice(&buf); + encode(&mut enc); + decode(&mut enc); + assert_eq!(&buf, &enc[..]); + } +} diff --git a/drivers/drmem-drv-tplink/src/tplink_api.rs b/drivers/drmem-drv-tplink/src/tplink/mod.rs similarity index 51% rename from drivers/drmem-drv-tplink/src/tplink_api.rs rename to drivers/drmem-drv-tplink/src/tplink/mod.rs index 77778f5f..90f02394 100644 --- a/drivers/drmem-drv-tplink/src/tplink_api.rs +++ b/drivers/drmem-drv-tplink/src/tplink/mod.rs @@ -2,132 +2,12 @@ // device. It also configures the `serde` crate so these commands are // converted to the expected JSON layout. -use serde::{Deserialize, Serialize}; -use std::{io::Write, marker::PhantomData}; +use serde::Deserialize; -// This type allows us to write a TP-Link command into a vector in one -// pass. When it is created, it contains the initial key. As data is -// written, it is "encrypted" and the key is updated. +mod cmd; +pub mod crypt; -pub struct CmdWriter<'a> { - key: u8, - buf: &'a mut Vec, -} - -impl<'a> CmdWriter<'a> { - // Creates a new, initialized writer. The parameter is the vector - // that is to receive the encrypted data. - - pub fn create(b: &'a mut Vec) -> Self { - CmdWriter { key: 171u8, buf: b } - } -} - -impl Write for CmdWriter<'_> { - // This is a mandatory method, but it doesn't do anything. - - fn flush(&mut self) -> std::io::Result<()> { - Ok(()) - } - - // Writes a buffer of data to the vector. As the data is - // transferred, it is "encrypted". Returns the number of bytes - // written (which is always the number passed in.) - - fn write(&mut self, b: &[u8]) -> std::io::Result { - let sz = b.len(); - - for ii in b.iter() { - self.key ^= *ii; - self.buf.push(self.key); - } - Ok(sz) - } -} - -// This is the decryption algorithm. - -fn decrypt(buf: &mut [u8]) { - let mut key = 171u8; - - for b in buf.iter_mut() { - let tmp = *b; - - *b ^= key; - key = tmp; - } -} - -// Defines the internal value used by the `set_relay_state` command. -// Needs to convert to `{"state":value}`. - -#[derive(Serialize, PartialEq, Debug)] -pub struct ActiveValue { - pub state: u8, -} - -// Defines the internal value used by the `set_led_off` command. Needs -// to convert to `{"off":value}`. - -#[derive(Serialize, PartialEq, Debug)] -pub struct LedValue { - pub off: u8, -} - -// Defines the internal value used by the `get_sysinfo` command. Needs -// to convert to `{}`. - -#[derive(Serialize, PartialEq, Debug)] -pub struct InfoValue { - #[serde(skip)] - pub nothing: PhantomData<()>, -} - -// Defines the internal value used by the `Brightness` command. Needs -// to convert to `{"brightness":value}`. - -#[derive(Serialize, PartialEq, Debug)] -pub struct BrightnessValue { - pub brightness: u8, -} - -#[derive(Serialize, PartialEq, Debug)] -pub enum Cmd { - #[serde(rename = "system")] - System { - #[serde(skip_serializing_if = "Option::is_none")] - set_relay_state: Option, - #[serde(skip_serializing_if = "Option::is_none")] - get_sysinfo: Option, - #[serde(skip_serializing_if = "Option::is_none")] - set_led_off: Option, - }, - - #[serde(rename = "smartlife.iot.dimmer")] - Dimmer { set_brightness: BrightnessValue }, -} - -impl Cmd { - pub fn encode(&self) -> Vec { - let mut buf = Vec::with_capacity(100); - - buf.push(0u8); - buf.push(0u8); - buf.push(0u8); - buf.push(0u8); - - serde_json::to_writer(CmdWriter::create(&mut buf), &self).unwrap(); - - let sz = buf.len() - 4; - - buf[0] = (sz >> 24) as u8; - buf[1] = (sz >> 16) as u8; - buf[2] = (sz >> 8) as u8; - buf[3] = sz as u8; - - buf - } -} +pub use cmd::Cmd; #[derive(Deserialize, PartialEq, Debug)] pub struct ErrorStatus { @@ -171,91 +51,44 @@ pub enum Reply { impl Reply { pub fn decode(buf: &mut [u8]) -> Option { - decrypt(buf); + crypt::decode(buf); serde_json::from_slice(buf).ok() } } -pub fn active_cmd(v: u8) -> Cmd { - Cmd::System { - set_relay_state: Some(ActiveValue { state: v }), - get_sysinfo: None, - set_led_off: None, - } -} - -pub fn brightness_cmd(v: u8) -> Cmd { - Cmd::Dimmer { - set_brightness: BrightnessValue { brightness: v }, - } -} - -pub fn info_cmd() -> Cmd { - Cmd::System { - set_relay_state: None, - get_sysinfo: Some(InfoValue { - nothing: PhantomData, - }), - set_led_off: None, - } -} - -pub fn led_cmd(v: bool) -> Cmd { - Cmd::System { - set_relay_state: None, - get_sysinfo: None, - set_led_off: Some(LedValue { off: (!v) as u8 }), - } -} - #[cfg(test)] mod tests { use super::*; use serde_json; - #[test] - fn test_crypt() { - let buf = [1u8, 2u8, 3u8, 4u8, 5u8]; - let mut v: Vec = Vec::new(); - - { - let mut wr = CmdWriter::create(&mut v); - - assert_eq!(wr.write(&buf).unwrap(), 5); - } - - decrypt(&mut v[..]); - assert_eq!(&buf, &v[..]); - } - #[test] fn test_cmds() { assert_eq!( - serde_json::to_string(&active_cmd(1)).unwrap(), + serde_json::to_string(&Cmd::mk_active_cmd(1)).unwrap(), "{\"system\":{\"set_relay_state\":{\"state\":1}}}" ); assert_eq!( - serde_json::to_string(&led_cmd(false)).unwrap(), + serde_json::to_string(&Cmd::mk_led_cmd(false)).unwrap(), "{\"system\":{\"set_led_off\":{\"off\":1}}}" ); assert_eq!( - serde_json::to_string(&led_cmd(true)).unwrap(), + serde_json::to_string(&Cmd::mk_led_cmd(true)).unwrap(), "{\"system\":{\"set_led_off\":{\"off\":0}}}" ); assert_eq!( - serde_json::to_string(&info_cmd()).unwrap(), + serde_json::to_string(&Cmd::mk_info_cmd()).unwrap(), "{\"system\":{\"get_sysinfo\":{}}}" ); assert_eq!( - serde_json::to_string(&brightness_cmd(0)).unwrap(), + serde_json::to_string(&Cmd::mk_brightness_cmd(0)).unwrap(), "{\"smartlife.iot.dimmer\":{\"set_brightness\":{\"brightness\":0}}}" ); assert_eq!( - serde_json::to_string(&brightness_cmd(50)).unwrap(), + serde_json::to_string(&Cmd::mk_brightness_cmd(50)).unwrap(), "{\"smartlife.iot.dimmer\":{\"set_brightness\":{\"brightness\":50}}}" ); assert_eq!( - serde_json::to_string(&brightness_cmd(100)).unwrap(), + serde_json::to_string(&Cmd::mk_brightness_cmd(100)).unwrap(), "{\"smartlife.iot.dimmer\":{\"set_brightness\":{\"brightness\":100}}}" ); } diff --git a/drmem-api/Cargo.toml b/drmem-api/Cargo.toml index 87ca5840..20f9bc65 100644 --- a/drmem-api/Cargo.toml +++ b/drmem-api/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "drmem-api" -version = "0.6.0" +version = "0.6.1" authors = ["Rich Neswold "] edition = "2021" description = "Traits and types used internally by the DrMem control system" diff --git a/drmem-api/src/driver/classes/dimmer_type.rs b/drmem-api/src/driver/classes/dimmer_type.rs index d4559437..7a30304f 100644 --- a/drmem-api/src/driver/classes/dimmer_type.rs +++ b/drmem-api/src/driver/classes/dimmer_type.rs @@ -17,10 +17,11 @@ //! ``` use crate::driver::{ - ro_device::ReadOnlyDevice, rw_device::ReadWriteDevice, DriverConfig, - Registrator, RequestChan, Result, + ro_device::ReadOnlyDevice, shared_rw_device::SharedReadWriteDevice, + DriverConfig, Registrator, RequestChan, Result, }; use std::future::Future; +use tokio::time::Duration; /// Defines the common API used by Dimmers. pub struct Dimmer { @@ -29,16 +30,17 @@ pub struct Dimmer { pub error: ReadOnlyDevice, /// Controls the brightness setting of the dimmer. Off is 0.0 and /// full-on is 100.0. - pub brightness: ReadWriteDevice, + pub brightness: SharedReadWriteDevice, /// A product might include an indicator. If the hardware does, /// this device can turn it on and off. - pub indicator: ReadWriteDevice, + pub indicator: SharedReadWriteDevice, } impl Registrator for Dimmer { fn register_devices<'a>( drc: &'a mut RequestChan, _cfg: &DriverConfig, + override_timeout: Option, max_history: Option, ) -> impl Future> + Send + 'a { let nm_error = "error".parse(); @@ -59,10 +61,20 @@ impl Registrator for Dimmer { .add_ro_device::(nm_error, None, max_history) .await?, brightness: drc - .add_rw_device::(nm_brightness, Some("%"), max_history) + .add_shared_rw_device::( + nm_brightness, + Some("%"), + override_timeout, + max_history, + ) .await?, indicator: drc - .add_rw_device::(nm_indicator, None, max_history) + .add_shared_rw_device::( + nm_indicator, + None, + override_timeout, + max_history, + ) .await?, }) } diff --git a/drmem-api/src/driver/classes/switch_type.rs b/drmem-api/src/driver/classes/switch_type.rs index 167d5c06..19211090 100644 --- a/drmem-api/src/driver/classes/switch_type.rs +++ b/drmem-api/src/driver/classes/switch_type.rs @@ -17,31 +17,60 @@ //! ``` use crate::driver::{ - rw_device::ReadWriteDevice, DriverConfig, Registrator, RequestChan, Result, + ro_device::ReadOnlyDevice, shared_rw_device::SharedReadWriteDevice, + DriverConfig, Registrator, RequestChan, Result, }; use std::future::Future; +use tokio::time::Duration; /// 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, /// Indicates the state of the switch. Writing `true` or `false` /// turns the switch on and off, respectively. - pub state: ReadWriteDevice, + pub state: SharedReadWriteDevice, + /// A product might include an indicator. If the hardware does, + /// this device can turn it on and off. + pub indicator: SharedReadWriteDevice, } impl Registrator for Switch { fn register_devices<'a>( drc: &'a mut RequestChan, _cfg: &DriverConfig, + override_timeout: Option, max_history: Option, ) -> impl Future> + Send + 'a { + let nm_error = "error".parse(); let nm_state = "state".parse(); + let nm_indicator = "indicator".parse(); async move { + let nm_error = nm_error?; let nm_state = nm_state?; + let nm_indicator = nm_indicator?; Ok(Switch { + error: drc + .add_ro_device::(nm_error, None, max_history) + .await?, state: drc - .add_rw_device::(nm_state, None, max_history) + .add_shared_rw_device::( + nm_state, + None, + override_timeout, + max_history, + ) + .await?, + indicator: drc + .add_shared_rw_device::( + nm_indicator, + None, + override_timeout, + max_history, + ) .await?, }) } diff --git a/drmem-api/src/driver/classes/weather_type.rs b/drmem-api/src/driver/classes/weather_type.rs index 034a8aa8..609e1b17 100644 --- a/drmem-api/src/driver/classes/weather_type.rs +++ b/drmem-api/src/driver/classes/weather_type.rs @@ -4,6 +4,7 @@ use crate::{ Error, }; use std::future::Future; +use tokio::time::Duration; pub enum Units { English, @@ -67,6 +68,7 @@ impl Registrator for Weather { fn register_devices<'a>( drc: &'a mut RequestChan, cfg: &DriverConfig, + _override_timeout: Option, max_history: Option, ) -> impl Future> + Send + 'a { let dewpoint_name = "dewpoint".parse::().unwrap(); diff --git a/drmem-api/src/driver/mod.rs b/drmem-api/src/driver/mod.rs index 558a2902..c253a232 100644 --- a/drmem-api/src/driver/mod.rs +++ b/drmem-api/src/driver/mod.rs @@ -4,7 +4,10 @@ use crate::types::{device, Error}; use std::future::Future; use std::{convert::Infallible, sync::Arc}; -use tokio::sync::{mpsc, oneshot}; +use tokio::{ + sync::{mpsc, oneshot}, + time::Duration, +}; use toml::value; use super::Result; @@ -220,7 +223,7 @@ impl RequestChan { &self, name: device::Base, units: Option<&str>, - override_duration: Option, + override_duration: Option, max_history: Option, ) -> Result> { let (tx, rx) = oneshot::channel(); @@ -268,7 +271,8 @@ pub trait ResettableState { pub trait Registrator: ResettableState + Sized + Send { fn register_devices<'a>( drc: &'a mut RequestChan, - cfg: &DriverConfig, + cfg: &'a DriverConfig, + override_timeout: Option, max_history: Option, ) -> impl Future> + Send + 'a; } diff --git a/drmemd/Cargo.toml b/drmemd/Cargo.toml index 371e3dff..334fc4b6 100644 --- a/drmemd/Cargo.toml +++ b/drmemd/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "drmemd" -version = "0.6.1" +version = "0.6.2" authors = ["Rich Neswold "] edition = "2021" description = "Main process of the DrMem control system" diff --git a/drmemd/src/driver/drv_cycle.rs b/drmemd/src/driver/drv_cycle.rs index a177a6d7..aad8bd8b 100644 --- a/drmemd/src/driver/drv_cycle.rs +++ b/drmemd/src/driver/drv_cycle.rs @@ -4,7 +4,7 @@ use drmem_api::{ Error, Result, }; use std::{convert::Infallible, future::Future}; -use tokio::time; +use tokio::time::{self, Duration}; use tracing::{self, debug}; // This enum represents the three states in which the device can be. @@ -23,7 +23,7 @@ pub struct Instance { enabled: Vec, state: CycleState, index: usize, - millis: time::Duration, + millis: Duration, } impl Instance { @@ -37,7 +37,7 @@ impl Instance { /// Creates a new, idle `Instance`. pub fn new( enabled_at_boot: bool, - millis: time::Duration, + millis: Duration, disabled: device::Value, enabled: Vec, ) -> Instance { @@ -53,7 +53,7 @@ impl Instance { // Validates the time duration from the driver configuration. - fn get_cfg_millis(cfg: &DriverConfig) -> Result { + fn get_cfg_millis(cfg: &DriverConfig) -> Result { match cfg.get("millis") { Some(toml::value::Value::Integer(millis)) => { // DrMem's official sample rate is 20 Hz, so the cycle @@ -65,7 +65,7 @@ impl Instance { // drmem-api crate indicating the max sample rate? if (50..=3_600_000).contains(millis) { - Ok(time::Duration::from_millis(*millis as u64)) + Ok(Duration::from_millis(*millis as u64)) } else { Err(Error::ConfigError(String::from( "'millis' out of range", @@ -202,6 +202,7 @@ impl driver::Registrator for Devices { fn register_devices<'a>( core: &'a mut driver::RequestChan, _cfg: &DriverConfig, + _override_timeout: Option, max_history: Option, ) -> impl Future> + Send + 'a { let output_name = "output".parse::().unwrap(); @@ -321,7 +322,7 @@ impl ResettableState for Devices {} mod tests { use super::*; use drmem_api::driver::API; - use std::time::Duration; + use tokio::time::Duration; #[tokio::test] async fn test_cfg() { @@ -451,7 +452,7 @@ mod tests { fn test_state_changes() { let mut timer = Instance::new( false, - time::Duration::from_millis(1000), + Duration::from_millis(1000), device::Value::Bool(false), vec![device::Value::Bool(true), device::Value::Bool(false)], ); @@ -489,7 +490,7 @@ mod tests { fn test_numeric_cycles() { let mut timer = Instance::new( false, - time::Duration::from_millis(1000), + Duration::from_millis(1000), device::Value::Int(0), vec![ device::Value::Int(1), @@ -536,7 +537,7 @@ mod tests { fn test_duplicate_skips() { let mut timer = Instance::new( false, - time::Duration::from_millis(1000), + Duration::from_millis(1000), device::Value::Bool(false), vec![ device::Value::Bool(true), @@ -580,7 +581,7 @@ mod tests { fn test_transition_skips() { let mut timer = Instance::new( false, - time::Duration::from_millis(1000), + Duration::from_millis(1000), device::Value::Bool(false), vec![ device::Value::Bool(false), diff --git a/drmemd/src/driver/drv_latch.rs b/drmemd/src/driver/drv_latch.rs index 420f9449..015f2cb1 100644 --- a/drmemd/src/driver/drv_latch.rs +++ b/drmemd/src/driver/drv_latch.rs @@ -4,6 +4,7 @@ use drmem_api::{ Error, Result, }; use std::{convert::Infallible, future::Future}; +use tokio::time::Duration; // This enum represents the two states in which the latch can be. @@ -98,6 +99,7 @@ impl driver::Registrator for Devices { fn register_devices<'a>( core: &'a mut driver::RequestChan, _cfg: &DriverConfig, + _override_timeout: Option, max_history: Option, ) -> impl Future> + Send + 'a { let output_name = "output".parse::().unwrap(); diff --git a/drmemd/src/driver/drv_map.rs b/drmemd/src/driver/drv_map.rs index e287d73b..0b961200 100644 --- a/drmemd/src/driver/drv_map.rs +++ b/drmemd/src/driver/drv_map.rs @@ -4,6 +4,7 @@ use drmem_api::{ Error, Result, }; use std::{convert::Infallible, future::Future, ops::RangeInclusive}; +use tokio::time::Duration; #[derive(Debug, PartialEq)] struct Entry(RangeInclusive, device::Value); @@ -200,6 +201,7 @@ impl driver::Registrator for Devices { fn register_devices<'a>( core: &'a mut driver::RequestChan, _cfg: &DriverConfig, + _override_timeout: Option, max_history: Option, ) -> impl Future> + Send + 'a { let output_name = "output".parse::().unwrap(); diff --git a/drmemd/src/driver/drv_memory.rs b/drmemd/src/driver/drv_memory.rs index 39be1033..869733f2 100644 --- a/drmemd/src/driver/drv_memory.rs +++ b/drmemd/src/driver/drv_memory.rs @@ -4,6 +4,7 @@ use drmem_api::{ Error, Result, }; use std::{convert::Infallible, future::Future}; +use tokio::time::Duration; // Defines the signature if a function that validates a // `device::Value`'s type. @@ -176,6 +177,7 @@ impl driver::Registrator for Devices { fn register_devices<'a>( core: &'a mut driver::RequestChan, cfg: &DriverConfig, + _override_timeout: Option, max_history: Option, ) -> impl Future> + Send + 'a { let vars = Self::get_cfg_vars(cfg); diff --git a/drmemd/src/driver/drv_timer.rs b/drmemd/src/driver/drv_timer.rs index 1393b78c..08049106 100644 --- a/drmemd/src/driver/drv_timer.rs +++ b/drmemd/src/driver/drv_timer.rs @@ -4,7 +4,7 @@ use drmem_api::{ Error, Result, }; use std::{convert::Infallible, future::Future}; -use tokio::time; +use tokio::time::{self, Duration}; use tracing::{debug, info}; // This enum represents the four states in which the timer can @@ -23,7 +23,7 @@ pub struct Instance { state: TimerState, active_value: device::Value, inactive_value: device::Value, - millis: time::Duration, + millis: Duration, } impl Instance { @@ -39,7 +39,7 @@ impl Instance { pub fn new( active_value: device::Value, inactive_value: device::Value, - millis: time::Duration, + millis: Duration, ) -> Instance { Instance { state: TimerState::Armed, @@ -69,11 +69,11 @@ impl Instance { // Validates the time duration from the driver configuration. - fn get_cfg_millis(cfg: &DriverConfig) -> Result { + fn get_cfg_millis(cfg: &DriverConfig) -> Result { match cfg.get("millis") { Some(toml::value::Value::Integer(millis)) => { if (50..=3_600_000).contains(millis) { - Ok(time::Duration::from_millis(*millis as u64)) + Ok(Duration::from_millis(*millis as u64)) } else { Err(Error::ConfigError(String::from( "'millis' out of range", @@ -189,6 +189,7 @@ impl driver::Registrator for Devices { fn register_devices<'a>( core: &'a mut driver::RequestChan, _cfg: &DriverConfig, + _override_timeout: Option, max_history: Option, ) -> impl Future> + Send + 'a { let output_name = "output".parse::().unwrap(); @@ -304,14 +305,13 @@ impl ResettableState for Devices {} #[cfg(test)] mod tests { use super::*; - use tokio::time; #[test] fn test_state_changes() { let mut timer = Instance::new( device::Value::Bool(true), device::Value::Bool(false), - time::Duration::from_millis(1000), + Duration::from_millis(1000), ); assert_eq!(timer.state, TimerState::Armed); diff --git a/drmemd/src/driver/mod.rs b/drmemd/src/driver/mod.rs index 64a8d00b..e0e24a5e 100644 --- a/drmemd/src/driver/mod.rs +++ b/drmemd/src/driver/mod.rs @@ -1,10 +1,11 @@ use drmem_api::{ driver::{self, Registrator, ResettableState, API}, - Result, + Error, Result, }; use futures::future::Future; use std::collections::HashMap; use std::{convert::Infallible, pin::Pin, sync::Arc}; +use tokio::{sync::Barrier, time::Duration}; use tracing::{error, field, info, info_span, warn, Instrument}; mod drv_cycle; @@ -16,8 +17,12 @@ mod drv_timer; pub type Fut = Pin + Send>>; pub type MgrTask = Fut>; -pub type Launcher = - fn(driver::DriverConfig, driver::RequestChan, Option) -> MgrTask; +pub type Launcher = fn( + driver::DriverConfig, + driver::RequestChan, + Option, + barrier: Arc, +) -> MgrTask; type DriverInfo = (&'static str, &'static str, Launcher); @@ -37,30 +42,26 @@ where let mut restart_delay = START_DELAY; loop { + info!("initializing"); + // Create a Future that creates an instance of the driver // using the provided configuration parameters. - let result = T::create_instance(&cfg) - .instrument(info_span!("prepping", cfg = field::Empty)); - - match result.await { + match T::create_instance(&cfg).await { Ok(mut instance) => { use futures::FutureExt; use std::panic::AssertUnwindSafe; restart_delay = START_DELAY; + info!("running"); // Drivers are never supposed to exit so // catch_unwind() will only catch panics which means // we need to only look for `Err(_)` values. - let Err(e) = AssertUnwindSafe( - instance - .run(&mut devices) - .instrument(info_span!("running")), - ) - .catch_unwind() - .await; + let Err(e) = AssertUnwindSafe(instance.run(&mut devices)) + .catch_unwind() + .await; error!("exited unexpectedly -- {e:?}") } @@ -85,6 +86,24 @@ where } } +fn get_cfg_override(cfg: &driver::DriverConfig) -> Result> { + match cfg.get("override_timeout") { + Some(toml::value::Value::Integer(secs)) => { + if (1..=(60 * 60 * 24 * 365)).contains(secs) { + Ok(Some(Duration::from_secs(*secs as u64))) + } else { + Err(Error::ConfigError(String::from( + "'override_timeout' out of range", + ))) + } + } + Some(_) => Err(Error::ConfigError(String::from( + "'override_timeout' config parameter should be an integer", + ))), + None => Ok(None), + } +} + // This generic function manages an instance of a specific driver. We // use generics because each driver has a different set of devices // (T::HardwareType), so one function wouldn't be able to handle every @@ -94,6 +113,7 @@ fn manage_instance( cfg: driver::DriverConfig, mut req_chan: driver::RequestChan, max_history: Option, + barrier: Arc, ) -> MgrTask where T: API + Send + 'static, @@ -101,14 +121,32 @@ where Box::pin(async move { // Let the driver API register the necessary devices. - let devices = - T::HardwareType::register_devices(&mut req_chan, &cfg, max_history) - .instrument(info_span!("register", cfg = field::Empty)) - .await?; + let devices = async { + match get_cfg_override(&cfg) { + Ok(tmo) => { + T::HardwareType::register_devices( + &mut req_chan, + &cfg, + tmo, + max_history, + ) + .instrument(info_span!("register", cfg = field::Empty)) + .await + } + Err(e) => Err(e), + } + } + .await; + + // When we reached this location, all devices have been + // registered (or not, if an error occurred). Sync to the + // barrier so the main loop and move on to the next driver. + + barrier.wait().await; // Create a future that manages the instance. - Ok(mgr_body::(devices, cfg).await) + Ok(mgr_body::(devices?, cfg).await) }) } diff --git a/drmemd/src/graphql/mod.rs b/drmemd/src/graphql/mod.rs index 06b3d476..0aa73f6a 100644 --- a/drmemd/src/graphql/mod.rs +++ b/drmemd/src/graphql/mod.rs @@ -10,7 +10,7 @@ use juniper_graphql_ws::ConnectionConfig; use juniper_warp::subscriptions::serve_graphql_ws; use libmdns::Responder; use std::{pin::Pin, result, sync::Arc, time::Duration}; -use tracing::{debug, error, info, info_span, Instrument}; +use tracing::{debug, error, info, info_span, warn, Instrument}; use warp::{http::StatusCode, reject, reply, Filter, Rejection, Reply}; pub mod config; @@ -1062,6 +1062,7 @@ fn build_site( db_logic: &[Logic], ) -> impl Filter + Clone { + warn!("building insecure GraphQL interface"); build_base_site(db, cchan, db_logic).recover(handle_rejection) } @@ -1072,6 +1073,8 @@ fn build_secure_site( db_logic: &[Logic], ) -> impl Filter + Clone { + info!("building secure GraphQL interface"); + // Clone the table of clients that are allowed in to the system. let clients: Arc<[String]> = Arc::clone(&cfg.clients); diff --git a/drmemd/src/logic/mod.rs b/drmemd/src/logic/mod.rs index ad0319fe..1eae033c 100644 --- a/drmemd/src/logic/mod.rs +++ b/drmemd/src/logic/mod.rs @@ -399,7 +399,7 @@ impl Node { let mut time = Arc::new((chrono::Utc::now(), chrono::Local::now())); let mut solar = None; - info!("starting"); + info!("running"); loop { // Create a future that yields the time-of-day using the @@ -509,14 +509,12 @@ impl Node { tokio::spawn( async move { - let name = cfg.name.clone(); - // Create a new instance and let it initialize itself. // Hold onto the result -- success or failure -- and // handle it after the barrier. let node = Node::init(c_req, rx_tod, rx_solar, cfg) - .instrument(info_span!("init", name = &name)) + .instrument(info_span!("init")) .await; // This barrier syncs this tasks with the start-up diff --git a/drmemd/src/logic/solar.rs b/drmemd/src/logic/solar.rs index 8c2ddc23..ab79933f 100644 --- a/drmemd/src/logic/solar.rs +++ b/drmemd/src/logic/solar.rs @@ -144,7 +144,7 @@ pub fn create_task( info!("waiting for clients to register"); barrier.wait().await; - info!("running task"); + info!("running"); run(tx_copy, lat, long).await } .instrument(info_span!("solar")), diff --git a/drmemd/src/logic/tod.rs b/drmemd/src/logic/tod.rs index ca5ca3af..baf1dbfb 100644 --- a/drmemd/src/logic/tod.rs +++ b/drmemd/src/logic/tod.rs @@ -155,7 +155,7 @@ pub fn create_task( info!("waiting for clients to register"); barrier.wait().await; - info!("running task"); + info!("running"); run(tx_copy).await } .instrument(info_span!("tod")), diff --git a/drmemd/src/main.rs b/drmemd/src/main.rs index a56dbd04..c50a95af 100644 --- a/drmemd/src/main.rs +++ b/drmemd/src/main.rs @@ -94,7 +94,7 @@ async fn run() -> Result<()> { // Iterate through the list of drivers specified in the // configuration file. - info!("starting driver instances"); + info!("starting drivers"); for driver in cfg.driver { let driver_name: drmem_api::driver::Name = @@ -110,6 +110,7 @@ async fn run() -> Result<()> { &driver.prefix, &tx_drv_req, ); + let barrier = Arc::new(Barrier::new(2)); // Call the function that manages instances of this // driver. If it returns `Ok()`, the value is a Future @@ -121,6 +122,7 @@ async fn run() -> Result<()> { driver.cfg.unwrap_or_default().clone(), chan, driver.max_history, + barrier.clone(), ); // Push the driver instance at the end of the vector. @@ -131,7 +133,9 @@ async fn run() -> Result<()> { name = driver_name.as_ref(), prefix = driver.prefix.to_string() ), - )))) + )))); + + let _ = barrier.wait().await; } else { error!("no driver named {}", driver.name); return Err(Error::NotFound); @@ -142,7 +146,7 @@ async fn run() -> Result<()> { // freed up. { - let barrier = Arc::new(Barrier::new(3)); + let barrier = Arc::new(Barrier::new(3 + cfg.logic.len())); // Start the time-of-day task. This needs to be done // *before* any logic blocks are started because logic @@ -160,9 +164,7 @@ async fn run() -> Result<()> { barrier.clone(), ); - info!("starting logic instances"); - - let init_barrier = Arc::new(Barrier::new(2)); + info!("starting logic blocks"); // Iterate through the [[logic]] sections of the config. @@ -172,12 +174,11 @@ async fn run() -> Result<()> { tx_tod.subscribe(), tx_solar.subscribe(), logic.clone(), - init_barrier.clone(), + barrier.clone(), ) .await; tasks.push(wrap_task(node_task)); - init_barrier.wait().await; } // Now that all the logic blocks have initialized, we @@ -195,7 +196,7 @@ async fn run() -> Result<()> { { use futures::FutureExt; - info!("starting GraphQL interface"); + info!("starting GraphQL"); // This server should never exit. If it does, report an // `OperationError`, @@ -206,6 +207,7 @@ async fn run() -> Result<()> { drv_tbl.clone(), tx_clnt_req.clone(), ) + .instrument(info_span!("gql")) .then(|_| async { Err(Error::OperationError("graphql server exited".to_owned())) });