airtouch5 0.2.0

//! A library to comumicate with AirTouch 5 air conditioning control consoles.
//!
//! # Connecting
//!
//! Use the [`AirTouch5::with_ipaddr()`] constructor to establish a connection
//! with an AirTouch 5 console. You can use the [`discovery`] module to
//! discover the address of a console on your local network.
//!
//! ```no_run
//! # use airtouch5::{discovery::{discover, Console, DiscoveryError}, AirTouch5};
//! # #[tokio::main(flavor = "current_thread")]
//! # async fn main() -> Result<(), DiscoveryError> {
//! let console_addr = discover().await?.address;
//! let at5 = AirTouch5::with_ipaddr(console_addr).await?;
//! # Ok(()) }
//! ```
//!
//! # AC units
//!
//! An AirTouch 5 system controls one or more air conditioning units. Each AC
//! unit is identified by an integer index from `0`–`15` (inclusive). Use the
//! [`ac_capabilities()`] method to list the AC units and their [capabilities],
//! or the [`ac_capability()`] method to fetch the capabilities of a single AC
//! unit by its index.
//!
//! ```no_run
//! # use airtouch5::AirTouch5;
//! # #[tokio::main(flavor = "current_thread")]
//! # async fn main() -> std::io::Result<()> {
//! # let console_addr = std::net::IpAddr::V4(std::net::Ipv4Addr::new(192, 0, 2, 123));
//! let at5 = AirTouch5::with_ipaddr(console_addr).await?;
//! for (_, ac) in at5.ac_capabilities().await?.by_index() {
//!     println!(
//!         "AC unit {} supports modes {} with fan speeds {}",
//!         ac.name,
//!         ac.supported_modes.iter_names().map(|(n, _)| n).collect::<Vec<&str>>().join(", "),
//!         ac.supported_fan_speeds.iter_names().map(|(n, _)| n).collect::<Vec<&str>>().join(", "),
//!     );
//! }
//! # Ok(()) }
//! ```
//!
//! Use the [`ac_status()`] method to get the current [status][ac_status] of
//! all AC units, including the current operating mode, current and target
//! temperatures.
//!
//! ```no_run
//! # use airtouch5::AirTouch5;
//! # #[tokio::main(flavor = "current_thread")]
//! # async fn main() -> std::io::Result<()> {
//! # let console_addr = std::net::IpAddr::V4(std::net::Ipv4Addr::new(192, 0, 2, 123));
//! let at5 = AirTouch5::with_ipaddr(console_addr).await?;
//! for (idx, ac) in at5.ac_status().await?.acs {
//!     println!(
//!         "AC unit {} is {} targetting {}",
//!         idx,
//!         ac.power.map_or("unknown".to_string(), |x| format!("{}", x)),
//!         ac.setpoint.map_or("unknown".to_string(), |x| format!("{:#}", x)),
//!     );
//! }
//! # Ok(()) }
//! ```
//!
//! When the `control` feature is enabled, use the [`control_ac()`] method to
//! change the [desired state][ac_control] of the AC unit (for example, to turn
//! the unit on, change between heating and cooling, or to adjust fan speed).
//!
//! ```no_run
//! # #[cfg(feature = "control")]
//! # use airtouch5::{AirTouch5, types::control::{AcControl, AcMode, AcPower, FanSpeed}};
//! # #[cfg(feature = "control")]
//! # #[tokio::main(flavor = "current_thread")]
//! # async fn main() -> std::io::Result<()> {
//! # let console_addr = std::net::IpAddr::V4(std::net::Ipv4Addr::new(192, 0, 2, 123));
//! let at5 = AirTouch5::with_ipaddr(console_addr).await?;
//! at5.control_ac(
//!     0,
//!     AcControl {
//!         power: Some(AcPower::On),
//!         mode: Some(AcMode::Cool),
//!         fan_speed: Some(FanSpeed::Auto),
//!         setpoint: None,
//!     },
//! )
//! .await?;
//! # Ok(()) }
//! # #[cfg(not(feature = "control"))]
//! # fn main() -> () { () }
//! ```
//!
//! # Zones
//!
//! Each AC unit's output may be divided into one or more zones; a zone is
//! usually a room or set of rooms to be air conditioned as a group. Each zone
//! is identified by an integer index. Each AC unit serves a contiguous set of
//! zones, identified by the starting zone index and number of zones.
//!
//! Use the [`zone_names()`] method to fetch the names of all zones in the
//! system, or the [`zone_name()`] method to fetch the name of a single zone by
//! its index.
//!
//! ```no_run
//! # use futures_util::future;
//! # use airtouch5::AirTouch5;
//! # #[tokio::main(flavor = "current_thread")]
//! # async fn main() -> std::io::Result<()> {
//! # let console_addr = std::net::IpAddr::V4(std::net::Ipv4Addr::new(192, 0, 2, 123));
//! let at5 = AirTouch5::with_ipaddr(console_addr).await?;
//! for (_, ac) in at5.ac_capabilities().await?.by_index() {
//!     let zones = future::try_join_all(
//!         (ac.zone_start_index..ac.zone_start_index+ac.zone_count)
//!             .map(|idx| at5.zone_name(idx))
//!         ).await?;
//!     println!(
//!         "AC unit {} serves zones {}",
//!         ac.name,
//!         zones.join(", "),
//!     );
//! }
//! # Ok(()) }
//! ```
//!
//! <div class="warning" id="bugs-zone-name-encoding">
//!
//! The protcol does not specify an encoding for non-ASCII characters in zone
//! names, and the console software appears to handle them poorly. From some
//! experimentaion, it appears that any non-ASCII characters are first encoded
//! as UTF-8, as one would expect, but then the encoded bytestring is truncated
//! to the same number of *bytes* as there are *characters* in the original
//! string. This means that, for example, a zone name with beginning with a
//! character whose UTF-8 encoding is three bytes long, but is otherwise ASCII,
//! will be missing the last two characters when decoded. Worse, a zone name
//! that *ends* with a character whose UTF-8 encoding is more than one byte
//! will have only its first byte present, which is invalid.
//!
//! </div>
//!
//! Use the [`zone_status()`] method to get the current [status][zone_status]
//! of all zones, including whether the zone is active, the amount of airflow
//! to the zone, and, if the zone has a temperature sensor, the current and
//! target temperatures.
//!
//! ```no_run
//! # use airtouch5::{AirTouch5, types::status::ZoneControl};
//! # #[tokio::main(flavor = "current_thread")]
//! # async fn main() -> std::io::Result<()> {
//! # let console_addr = std::net::IpAddr::V4(std::net::Ipv4Addr::new(192, 0, 2, 123));
//! let at5 = AirTouch5::with_ipaddr(console_addr).await?;
//! for (idx, zone) in at5.zone_status().await?.zones {
//!     println!(
//!         "Zone {} is {} targetting {}",
//!         idx,
//!         zone.power,
//!         match zone.control {
//!             ZoneControl::Airflow(pct) => format!("{}%", pct),
//!             ZoneControl::Temperature(_, t) => format!("{:#}", t),
//!         },
//!     );
//! }
//! # Ok(()) }
//! ```
//!
//! When the `control` feature is enabled, use the [`control_zone()`] method
//! to change the [desired state][zone_control] of the zone (for example, to
//! turn the zone off or to adjust the setpoint or airflow).
//!
//! ```no_run
//! # #[cfg(feature = "control")]
//! # use airtouch5::{AirTouch5, types::control::{ZoneControl, ZoneControlValue, ZonePower}};
//! # #[cfg(feature = "control")]
//! # #[tokio::main(flavor = "current_thread")]
//! # async fn main() -> std::io::Result<()> {
//! # let console_addr = std::net::IpAddr::V4(std::net::Ipv4Addr::new(192, 0, 2, 123));
//! let at5 = AirTouch5::with_ipaddr(console_addr).await?;
//! at5.control_zone(
//!     0,
//!     ZoneControl {
//!         power: Some(ZonePower::On),
//!         control: None,
//!         value: Some(ZoneControlValue::Increment),
//!     }
//! )
//! .await?;
//! # Ok(()) }
//! # #[cfg(not(feature = "control"))]
//! # fn main() -> () { () }
//! ```
//!
//! # Subscribing to asynchronous updates
//!
//! The AirTouch 5 console sends asynchronous updates when the state of the
//! system changes (for example, a user turns the air conditioner on from the
//! console, or the current temperature of a zone changes). Use the
//! [`subscribe_changes()`] or [`subscribe_status()`] methods to arrange to
//! receive such updates.
//!
//! [ac_status]: types::status::AcStatus
//! [zone_status]: types::status::ZoneStatus
//! [capabilities]: types::info::AcCapability
//! [ac_control]: types::control::AcControl
//! [zone_control]: types::control::ZoneControl
//! [`ac_status()`]: AirTouch5::ac_status
//! [`zone_status()`]: AirTouch5::zone_status
//! [`ac_capabilities()`]: AirTouch5::ac_capabilities
//! [`ac_capability()`]: AirTouch5::ac_capability
//! [`zone_name()`]: AirTouch5::zone_name
//! [`zone_names()`]: AirTouch5::zone_names
//! [`control_ac()`]: AirTouch5::control_ac
//! [`control_zone()`]: AirTouch5::control_zone
//! [`subscribe_changes()`]: AirTouch5::subscribe_changes
//! [`subscribe_status()`]: AirTouch5::subscribe_status

#![cfg_attr(docsrs, feature(doc_cfg))]

mod conn;
pub mod discovery;
mod io_loop;
mod message;
pub mod types;

#[cfg(feature = "timeout")]
use std::time::Duration;
use std::{io, net::IpAddr};

use crate::{
    message::{
        ac_capability::{AcCapability, AcCapabilityResponse},
        ac_status::AcStatusMessage,
        console_version::ConsoleVersionResponse,
        zone_name::ZoneNameResponse,
        zone_status::ZoneStatusMessage,
    },
    types::status::{CurrentStatus, StatusChange},
};

#[cfg(feature = "control")]
use crate::message::{
    ac_control::{AcControl, AcControlMessage},
    zone_control::{ZoneControl, ZoneControlMessage},
};

/// An AirTouch 5 console that we are interacting with.
///
/// This is the primary structure for interacting with the console. See the
/// [crate-level docs](index.html) for usage and examples.
pub struct AirTouch5 {
    conn_loop: Option<(
        tokio::task::JoinHandle<()>,
        tokio::sync::oneshot::Sender<()>,
    )>,

    /// Channel to send requests to the connection loop on
    channel_req: tokio::sync::mpsc::Sender<io_loop::Request>,

    /// Clone of the connection loop's status watch channel, only used for clients
    /// to `subscribe()` to.
    // TODO: this being a clone prevents the channel from being closed if the io_loop's
    // Sender is dropped (e.g. if the loop terminates on a disconnect). But there is
    // [currently no `watch::WeakSender`](https://github.com/tokio-rs/tokio/issues/7517).
    // Should this be a weak reference to an Rc<Sender<_>> instead?
    status: tokio::sync::watch::Sender<CurrentStatus>,

    /// Weak reference to the connection loop's change broadcast, only used for
    /// clients to `subscribe()` to.
    changes: tokio::sync::broadcast::WeakSender<StatusChange>,
}

impl AirTouch5 {
    /// Connect to an AirTouch 5 console at the given IP address and port.
    ///
    /// This constructor requires the port number to be specified. Since the
    /// Airtouch 5 console always accepts connections on a known port, it is
    /// often more convenient to use one of the `with_` constructors that
    /// only require the IP address or hostname, and use the known port.
    /// However it may sometimes be necessary to connect to a differeny port
    /// number, such as when tunnelling a connection though SSH or some other
    /// proxy or port mapper, and this constructor is provided for those
    /// occasions.
    ///
    /// ```no_run
    /// # use std::net::{IpAddr, Ipv4Addr};
    /// # use airtouch5::AirTouch5;
    /// # #[tokio::main(flavor = "current_thread")]
    /// # async fn main() -> std::io::Result<()> {
    /// let console_addr = IpAddr::V4(Ipv4Addr::new(192, 0, 2, 123));
    /// let console_port = 12345;
    /// let at5 = AirTouch5::new((console_addr, console_port)).await?;
    /// # Ok(()) }
    /// ```
    pub async fn new<A: tokio::net::ToSocketAddrs>(addr: A) -> io::Result<Self> {
        let connection = conn::Connection::new(addr).await?;
        Self::with_connection(connection).await
    }

    /// Connect to an AirTouch 5 console at the given IP address.
    ///
    /// ```no_run
    /// # use std::net::{IpAddr, Ipv4Addr};
    /// # use airtouch5::AirTouch5;
    /// # #[tokio::main(flavor = "current_thread")]
    /// # async fn main() -> std::io::Result<()> {
    /// let console_addr = IpAddr::V4(Ipv4Addr::new(192, 0, 2, 123));
    /// let at5 = AirTouch5::with_ipaddr(console_addr).await?;
    /// # Ok(()) }
    /// ```
    pub async fn with_ipaddr(addr: IpAddr) -> io::Result<Self> {
        let connection = conn::Connection::with_ipaddr(addr).await?;
        Self::with_connection(connection).await
    }

    /// Connect to an AirTouch 5 console at any of the given IP address.
    /// The connection will be made to the first address that sucessfully
    /// connects.
    ///
    /// ```no_run
    /// # use std::net::{IpAddr, Ipv4Addr};
    /// # use airtouch5::AirTouch5;
    /// # #[tokio::main(flavor = "current_thread")]
    /// # async fn main() -> std::io::Result<()> {
    /// let console_addrs = vec![
    ///     IpAddr::V4(Ipv4Addr::new(192, 0, 2, 123)),
    ///     IpAddr::V4(Ipv4Addr::new(192, 0, 2, 234))
    /// ];
    /// let at5 = AirTouch5::with_ipaddrs(&console_addrs).await?;
    /// # Ok(()) }
    /// ```
    pub async fn with_ipaddrs(addrs: &[IpAddr]) -> io::Result<Self> {
        let connection = conn::Connection::with_ipaddrs(addrs).await?;
        Self::with_connection(connection).await
    }

    /// Connect to an AirTouch 5 console at the given IP address.
    ///
    /// ```no_run
    /// # use airtouch5::AirTouch5;
    /// # #[tokio::main(flavor = "current_thread")]
    /// # async fn main() -> std::io::Result<()> {
    /// let console_addr = "192.0.2.123";
    /// let at5 = AirTouch5::with_str(console_addr).await?;
    /// # Ok(()) }
    /// ```
    pub async fn with_str(addr: &str) -> io::Result<Self> {
        let connection = conn::Connection::with_str(addr).await?;
        Self::with_connection(connection).await
    }

    /// Spawn the I/O loop and construct this instance around the given
    /// [`Connection`][conn::Connection].
    async fn with_connection(connection: conn::Connection) -> io::Result<Self> {
        let (kill_tx, kill_rx) = tokio::sync::oneshot::channel();
        let (cast_tx, _) = tokio::sync::broadcast::channel(32);
        let (watch_tx, _) = tokio::sync::watch::channel(CurrentStatus::default());
        let (req_tx, req_rx) = tokio::sync::mpsc::channel(4);
        let changes = cast_tx.downgrade();
        let status = watch_tx.clone();
        let h = io_loop::spawn(connection, kill_rx, req_rx, cast_tx, watch_tx);
        Ok(Self {
            conn_loop: Some((h, kill_tx)),
            channel_req: req_tx,
            status,
            changes,
        })
    }

    /// Returns a receiver for a [`broadcast`] channel, whose values represent
    /// asynchronous updates to the current system state. Each update may be for
    /// or more AC units, or for one or more zones.
    ///
    /// ```no_run
    /// # use std::net::{IpAddr, Ipv4Addr};
    /// # use airtouch5::{AirTouch5, types::status::StatusChange};
    /// # #[tokio::main(flavor = "current_thread")]
    /// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
    /// # let console_addr = IpAddr::V4(Ipv4Addr::new(192, 0, 2, 123));
    /// let at5 = AirTouch5::with_ipaddr(console_addr).await?;
    /// if let Some(mut changes) = at5.subscribe_changes() {
    ///     let change = changes.recv().await?;
    ///     match change {
    ///         StatusChange::AcStatusChange(acs) => {
    ///             // handle the new AC state(s)
    ///         },
    ///         StatusChange::ZoneStatusChange(zones) => {
    ///             // handle the new zone state(s)
    ///         },
    ///         _ => {
    ///             // a change other than AC or zone state;
    ///             // reserved for future use
    ///         },
    ///     }
    /// }
    /// # Ok(()) }
    /// ```
    ///
    /// `None` may be returned if the channel has already been closed (for example,
    /// if the connection has been lost).
    ///
    /// [`broadcast`]: tokio::sync::broadcast
    pub fn subscribe_changes(&self) -> Option<tokio::sync::broadcast::Receiver<StatusChange>> {
        self.changes.upgrade().map(|c| c.subscribe())
    }

    /// Returns a recevier for a [`watch`] channel, whose value represents the
    /// [current known state][CurrentStatus] of the system.
    ///
    /// ```no_run
    /// # use std::net::{IpAddr, Ipv4Addr};
    /// # use airtouch5::{AirTouch5, types::status::{StatusSet, AcPower, ZonePower}};
    /// # #[tokio::main(flavor = "current_thread")]
    /// # async fn main() -> std::io::Result<()> {
    /// # let console_addr = IpAddr::V4(Ipv4Addr::new(192, 0, 2, 123));
    /// let at5 = AirTouch5::with_ipaddr(console_addr).await?;
    /// if let Some(mut watch) = at5.subscribe_status() {
    ///     let current = watch.borrow_and_update();
    ///     println!("{} AC units and {} zones are on",
    ///         current.acs().values()
    ///             .filter(|a| a.power.is_some() && a.power != Some(AcPower::Off))
    ///             .count(),
    ///         current.zones().values()
    ///             .filter(|z| z.power != ZonePower::Off)
    ///             .count(),
    ///     );
    /// }
    /// # Ok(()) }
    /// ```
    ///
    /// `None` may be returned if the channel has already been closed (for example,
    /// if the connection has been lost).
    ///
    /// <div class="warning">
    /// Note that when a connection is first established, until a full set of
    /// status messages has been received, the state is not fully known, and only
    /// the known state will be represented by the channel value.
    ///
    /// Calling both [`ac_status()`] and [`zone_status()`] will ensure that the
    /// state is fully known.
    ///
    /// ```no_run
    /// # use std::net::{IpAddr, Ipv4Addr};
    /// # use airtouch5::AirTouch5;
    /// # #[tokio::main(flavor = "current_thread")]
    /// # async fn main() -> std::io::Result<()> {
    /// # let console_addr = IpAddr::V4(Ipv4Addr::new(192, 0, 2, 123));
    /// let at5 = AirTouch5::with_ipaddr(console_addr).await?;
    /// let _ = at5.ac_status().await?;
    /// let _ = at5.zone_status().await?;
    /// let mut watch = at5.subscribe_status();
    /// # Ok(()) }
    /// ```
    ///
    /// <!-- TODO: have `subscribe_status()` arrange for status request(s) to be
    /// sent if the state isn't already fully known -->
    /// </div>
    ///
    /// [`ac_status()`]: AirTouch5::ac_status
    /// [`zone_status()`]: AirTouch5::zone_status
    /// [`watch`]: tokio::sync::watch
    pub fn subscribe_status(&self) -> Option<tokio::sync::watch::Receiver<CurrentStatus>> {
        Some(self.status.subscribe())
    }

    /// Gracefully shut down the connection to the console.
    pub async fn shutdown(&mut self) -> io::Result<()> {
        if let Some((h, tx)) = self.conn_loop.take() {
            if tx.send(()).is_ok() {
                return Ok(h.await?);
            }
        }
        Ok(())
    }

    // FIXME: we use io::Result/io::Error too much instead of making our own newtype

    /// Send a request frame to the console (via the I/O loop) and wait for a reply.
    async fn send_request(&self, request: conn::frame::Frame) -> io::Result<conn::frame::Frame> {
        let (ret_tx, ret_rx) = tokio::sync::oneshot::channel();
        self.channel_req
            .send((request, ret_tx))
            .await
            .map_err(io::Error::other)?;
        ret_rx.await.map_err(io::Error::other)?
    }

    /// Request the current status of all zones.
    pub async fn zone_status(&self) -> io::Result<ZoneStatusMessage> {
        let req = ZoneStatusMessage::request();
        let f = self
            .send_request(req.try_into().map_err(io::Error::other)?)
            .await?;
        let resp: ZoneStatusMessage = f.try_into()?;
        Ok(resp)
    }

    /// Request the current status of all zones, with timeout.
    #[cfg(feature = "timeout")]
    pub async fn zone_status_timeout(&self, duration: Duration) -> io::Result<ZoneStatusMessage> {
        tokio::time::timeout(duration, self.zone_status()).await?
    }

    /// Request the current status of all AC units.
    pub async fn ac_status(&self) -> io::Result<AcStatusMessage> {
        let req = AcStatusMessage::request();
        let f = self
            .send_request(req.try_into().map_err(io::Error::other)?)
            .await?;
        let resp: AcStatusMessage = f.try_into()?;
        Ok(resp)
    }

    /// Request the current status of all AC units, with timeout.
    #[cfg(feature = "timeout")]
    pub async fn ac_status_timeout(&self, duration: Duration) -> io::Result<AcStatusMessage> {
        tokio::time::timeout(duration, self.ac_status()).await?
    }

    /// Request the capabilities of all AC units.
    pub async fn ac_capabilities(&self) -> io::Result<AcCapabilityResponse> {
        let req = message::ac_capability::AcCapabilityRequest::new(None);
        let f = self
            .send_request(req.try_into().map_err(io::Error::other)?)
            .await?;
        let resp: AcCapabilityResponse = f.try_into()?;
        Ok(resp)
    }

    /// Request the capabilities of all AC units, with timeout.
    #[cfg(feature = "timeout")]
    pub async fn ac_capabilities_timeout(
        &self,
        duration: Duration,
    ) -> io::Result<AcCapabilityResponse> {
        tokio::time::timeout(duration, self.ac_capabilities()).await?
    }

    /// Request the capabilities of a specific AC unit.
    pub async fn ac_capability(&self, ac_index: u8) -> io::Result<AcCapability> {
        let req = message::ac_capability::AcCapabilityRequest::new(Some(ac_index));
        let f = self
            .send_request(req.try_into().map_err(io::Error::other)?)
            .await?;
        let mut resp: AcCapabilityResponse = f.try_into()?;
        if let Some(ac) = resp.acs.remove(&ac_index) {
            Ok(ac)
        } else {
            Err(io::Error::from(io::ErrorKind::NotFound))
        }
    }

    /// Request the capabilities of a specific AC unit, with timeout.
    #[cfg(feature = "timeout")]
    pub async fn ac_capability_timeout(
        &self,
        ac_index: u8,
        duration: Duration,
    ) -> io::Result<AcCapability> {
        tokio::time::timeout(duration, self.ac_capability(ac_index)).await?
    }

    /// Request the version information of the AirTouch 5 system.
    pub async fn console_version(&self) -> io::Result<ConsoleVersionResponse> {
        let req = message::console_version::ConsoleVersionRequest::new();
        let f = self
            .send_request(req.try_into().map_err(io::Error::other)?)
            .await?;
        let resp: ConsoleVersionResponse = f.try_into()?;
        Ok(resp)
    }

    /// Request the version information of the AirTouch 5 system, with timeout.
    #[cfg(feature = "timeout")]
    pub async fn console_version_timeout(
        &self,
        duration: Duration,
    ) -> io::Result<ConsoleVersionResponse> {
        tokio::time::timeout(duration, self.console_version()).await?
    }

    /// Request the names of all zones.
    pub async fn zone_names(&self) -> io::Result<ZoneNameResponse> {
        let req = message::zone_name::ZoneNameRequest::new(None);
        let f = self
            .send_request(req.try_into().map_err(io::Error::other)?)
            .await?;
        let resp: ZoneNameResponse = f.try_into()?;
        Ok(resp)
    }

    /// Request the names of all zones, with timeout.
    #[cfg(feature = "timeout")]
    pub async fn zone_names_timeout(&self, duration: Duration) -> io::Result<ZoneNameResponse> {
        tokio::time::timeout(duration, self.zone_names()).await?
    }

    /// Request the name of a specific zone.
    pub async fn zone_name(&self, zone_idx: u8) -> io::Result<String> {
        let req = message::zone_name::ZoneNameRequest::new(Some(zone_idx));
        let f = self
            .send_request(req.try_into().map_err(io::Error::other)?)
            .await?;
        let mut resp: ZoneNameResponse = f.try_into()?;
        if let Some(name) = resp.zones.remove(&zone_idx) {
            Ok(name)
        } else {
            Err(io::Error::other(format!("no zone {}", zone_idx)))
        }
    }

    /// Request the name of a specific zone, with timeout.
    #[cfg(feature = "timeout")]
    pub async fn zone_name_timeout(&self, zone_idx: u8, duration: Duration) -> io::Result<String> {
        tokio::time::timeout(duration, self.zone_name(zone_idx)).await?
    }

    /// Request a change in the control values for a specific zone.
    #[cfg(feature = "control")]
    pub async fn control_zone(
        &self,
        zone_idx: u8,
        zone_control: ZoneControl,
    ) -> io::Result<ZoneStatusMessage> {
        self.control_zones([(zone_idx, zone_control)]).await
    }

    /// Request a change in the control values for a specific zone, with timeout.
    #[cfg(all(feature = "control", feature = "timeout"))]
    pub async fn control_zone_timeout(
        &self,
        zone_idx: u8,
        zone_control: ZoneControl,
        duration: Duration,
    ) -> io::Result<ZoneStatusMessage> {
        tokio::time::timeout(duration, self.control_zones([(zone_idx, zone_control)])).await?
    }

    /// Request a change in the control values for a number of zones.
    #[cfg(feature = "control")]
    pub async fn control_zones<
        K: Into<u8>,
        V: Into<ZoneControl>,
        T: IntoIterator<Item = (K, V)>,
    >(
        &self,
        zones: T,
    ) -> io::Result<ZoneStatusMessage> {
        let req = ZoneControlMessage::new(zones);
        let f = self
            .send_request(req.try_into().map_err(io::Error::other)?)
            .await?;
        let resp: ZoneStatusMessage = f.try_into()?;
        Ok(resp)
    }

    /// Request a change in the control values for a number of zones, with timeout.
    #[cfg(all(feature = "control", feature = "timeout"))]
    pub async fn control_zones_timeout<
        K: Into<u8>,
        V: Into<ZoneControl>,
        T: IntoIterator<Item = (K, V)>,
    >(
        &self,
        zones: T,
        duration: Duration,
    ) -> io::Result<ZoneStatusMessage> {
        tokio::time::timeout(duration, self.control_zones(zones)).await?
    }

    /// Request a change in the control values for a specific AC unit.
    #[cfg(feature = "control")]
    pub async fn control_ac(
        &self,
        ac_idx: u8,
        ac_control: AcControl,
    ) -> io::Result<AcStatusMessage> {
        self.control_acs([(ac_idx, ac_control)]).await
    }

    /// Request a change in the control values for a specific AC unit, with timeout.
    #[cfg(all(feature = "control", feature = "timeout"))]
    pub async fn control_ac_timeout(
        &self,
        ac_idx: u8,
        ac_control: AcControl,
        duration: Duration,
    ) -> io::Result<AcStatusMessage> {
        tokio::time::timeout(duration, self.control_acs([(ac_idx, ac_control)])).await?
    }

    /// Request a change in the control values for a number of AC units.
    #[cfg(feature = "control")]
    pub async fn control_acs<K: Into<u8>, V: Into<AcControl>, T: IntoIterator<Item = (K, V)>>(
        &self,
        acs: T,
    ) -> io::Result<AcStatusMessage> {
        let req = AcControlMessage::new(acs);
        let f = self
            .send_request(req.try_into().map_err(io::Error::other)?)
            .await?;
        let resp: AcStatusMessage = f.try_into()?;
        Ok(resp)
    }

    /// Request a change in the control values for a number of AC units, with timeout.
    #[cfg(all(feature = "control", feature = "timeout"))]
    pub async fn control_acs_timeout<
        K: Into<u8>,
        V: Into<AcControl>,
        T: IntoIterator<Item = (K, V)>,
    >(
        &self,
        acs: T,
        duration: Duration,
    ) -> io::Result<AcStatusMessage> {
        tokio::time::timeout(duration, self.control_acs(acs)).await?
    }
}

impl Drop for AirTouch5 {
    // ensure we signal the task to exit
    fn drop(&mut self) {
        if let Some((_, tx)) = self.conn_loop.take() {
            let _ = tx.send(()); /* ignore errors, the task must already be gone */
        }
    }
}

#[cfg(test)]
#[macro_use]
extern crate assert_matches;

#[cfg(test)]
mod tests {
    #[allow(unused_imports)]
    use super::*;

    // No unit tests yet (this is probably more a job for integration tests anyway).
}