huawei_inverter/

client.rs

use crate::{
    error::ReadError, helpers::request_failure_actions, lock::get_connection,
    pointrange::PointRanges, Config, DynValue, ModbusError, Model, Point, PointDefinition,
    PointRange, Value, Word, WriteError, WriteQueue,
};
use itertools::Itertools;
use std::{net::SocketAddr, time::Duration};
use tokio::time::sleep;
use tokio_modbus::{Slave, SlaveId};

#[derive(Debug, Clone, Copy)]
pub struct Client {
    config: Config,
}

#[derive(Debug, thiserror::Error)]
pub enum ClientError {
    #[error(transparent)]
    Modbus(#[from] ModbusError),
    #[error(transparent)]
    Read(#[from] ReadError),
    #[error(transparent)]
    Write(#[from] WriteError),
    #[error("IO error: {0}")]
    IO(#[from] std::io::Error),
}

impl Client {
    pub fn new(config: Config) -> Self {
        Self { config }
    }

    pub fn socket_addr(&self) -> SocketAddr {
        self.config.socket_addr
    }

    pub fn slave(&self) -> Slave {
        self.config.slave
    }

    pub fn slave_id(&self) -> SlaveId {
        self.config.slave.0
    }

    pub fn connect_timeout(&self) -> Duration {
        self.config.connect_timeout
    }

    pub fn read_timeout(&self) -> Duration {
        self.config.read_timeout
    }

    pub fn write_timeout(&self) -> Duration {
        self.config.write_timeout
    }

    pub fn num_read_retries(&self) -> usize {
        self.config.num_read_retries
    }

    pub fn num_write_retries(&self) -> usize {
        self.config.num_write_retries
    }

    pub async fn read_model<M: Model>(&self) -> Result<M, ClientError> {
        // "125" seems to be a hard limit for how much can be read from Huawei inverter
        // at the same time.
        let max_read_length = 125;
        let point_ranges = M::point_ranges();
        let ranges = if max_read_length == point_ranges.max_read_words() {
            point_ranges
        } else {
            PointRanges::new(M::READABLE_POINT_DEFINITIONS.to_owned(), max_read_length)
        };
        // First read gets timeout without waiting at least one second after connect.
        let raw = self.read_raw(&ranges).await?;
        Ok(M::parse(&raw)?)
    }
    pub async fn read_point<V: Value>(&self, point: &Point<V>) -> Result<V, ClientError> {
        // First read gets timeout without waiting at least one second after connect.
        let ranges = PointRanges::new(vec![point.def], 125);
        let data = self.read_raw(&ranges).await?;
        Ok(ranges
            .parse_one(point.to_owned(), &data)
            .map_err(ReadError::from)?)
    }

    pub async fn read_point_dyn(&self, def: PointDefinition) -> Result<DynValue, ClientError> {
        let ranges = PointRanges::new(vec![def], 125);
        let data = self.read_raw(&ranges).await?;
        Ok(def
            .decode_dyn(&data[0..def.quantity as usize])
            .map_err(ReadError::from)?)
    }

    pub async fn write_point<V: Value + std::fmt::Debug>(
        &self,
        point: Point<V>,
        value: V,
    ) -> Result<(), ClientError> {
        // Validate that value is correct before trying to insert
        value.validate().map_err(WriteError::from)?;

        let raw_value = value.encode();
        self.write_raw(&[(point.def, raw_value)]).await?;
        Ok(())
    }

    pub async fn write_multiple(&self, writer: &mut WriteQueue) -> Result<(), ClientError> {
        let point_values = writer
            .dequeue_all()
            .into_iter()
            .map(|w| (w.point_def(), w.raw_value().to_owned()))
            .collect_vec();
        self.write_raw(&point_values).await?;
        Ok(())
    }

    /// Read raw data from modbus
    pub async fn read_raw(&self, ranges: &PointRanges) -> Result<Vec<u16>, ClientError> {
        let mut data: Vec<u16> = Vec::with_capacity(ranges.address_len());
        let num_retries = self.num_read_retries();

        let connection = get_connection(self.config).await?;
        let mut conn_guard = connection.lock().await;

        for range in ranges.ranges() {
            match range {
                PointRange::Gap(range) => {
                    tracing::trace!("Filling range {}..{} with zeroes…", range.start, range.end);
                    data.extend((range.start..range.end).map(|_| 0));
                }
                PointRange::Definitions { range, .. } => {
                    let start = range.start;
                    let quantity = range.end - range.start;
                    for retry_no in 0..=num_retries {
                        let last_retry = retry_no == num_retries;
                        match conn_guard.read_registers(start, quantity).await {
                            Ok(chunk) => {
                                data.extend(chunk);
                                break;
                            }
                            Err(error) => {
                                if last_retry {
                                    return Err(error.into());
                                }
                                tracing::debug!(
                                    "Modbus request failed with error: {:?}, retrying...",
                                    error
                                );
                                let actions = request_failure_actions(error)?;
                                for action in actions {
                                    match action {
                                        crate::helpers::RequestFailureAction::Sleep(secs) => {
                                            sleep(Duration::from_secs(secs.into())).await
                                        }
                                        crate::helpers::RequestFailureAction::NewContext => {
                                            conn_guard.setup().await?;
                                        }
                                    }
                                }
                            }
                        };
                    }
                }
            }
        }
        Ok(data)
    }

    pub async fn write_raw(
        &self,
        point_values: &[(PointDefinition, Box<[Word]>)],
    ) -> Result<(), ClientError> {
        let num_retries = self.num_write_retries();

        for (def, _) in point_values {
            if !def.is_writable() {
                return Err(WriteError::NotWritable.into());
            }
        }

        let connection = get_connection(self.config).await?;
        let mut conn_guard = connection.lock().await;

        for (point_def, raw_value) in point_values {
            if raw_value.len() > point_def.quantity as usize {
                return Err(WriteError::ValueTooLarge.into());
            }

            for retry_no in 0..=num_retries {
                let last_retry = retry_no == num_retries;
                match conn_guard
                    .write_registers(point_def.address, raw_value)
                    .await
                {
                    Ok(_) => {
                        tracing::debug!(
                            "Wrote {:?} word(s) to point {:?}",
                            raw_value.len(),
                            point_def.address
                        );
                        break;
                    }
                    Err(error) => {
                        if last_retry {
                            return Err(error.into());
                        }
                        tracing::debug!(
                            "Modbus request failed with error: {:?}, retrying...",
                            error
                        );
                        let actions = request_failure_actions(error)?;
                        for action in actions {
                            match action {
                                crate::helpers::RequestFailureAction::Sleep(secs) => {
                                    sleep(Duration::from_secs(secs.into())).await
                                }
                                crate::helpers::RequestFailureAction::NewContext => {
                                    conn_guard.setup().await?;
                                }
                            }
                        }
                    }
                };
            }
        }
        Ok(())
    }
}