【功能】ModbusCode重构

src/ems/bms/bms.rs

@@ -41,7 +41,7 @@ impl Bms {
 
 #[async_trait::async_trait]
 impl Service for Bms {
-    async fn read_task(&self) {
+    async fn south(&self) {
         let _producer = self.producer.clone();
         let can_socket = Arc::clone(&self.can_socket);
         loop {
@@ -50,7 +50,7 @@ impl Service for Bms {
         }
     }
 
-    async fn write_task(&self) {
+    async fn north(&self) {
         let consumer = self.consumer.clone();
         loop {
             let string = consumer.lock().await.recv().await.unwrap();

src/ems/bms/gold/gold_bms.rs

@@ -40,7 +40,7 @@ impl GoldBms {
         self.ctx
             .lock()
             .await
-            .read_holding_registers(0, 10)
+            .read_discrete_inputs(0, 10)
             .await??;
         Ok(())
     }
@@ -48,7 +48,7 @@ impl GoldBms {
 
 #[async_trait]
 impl Service for GoldBms {
-    async fn read_task(&self) {
+    async fn south(&self) {
         loop {
             tokio::time::sleep(Duration::from_millis(app_config().emu.bms.interval)).await;
             match self.read_input_register().await {
@@ -64,7 +64,8 @@ impl Service for GoldBms {
         }
     }
 
-    async fn write_task(&self) {}
+    async fn north(&self) {
+    }
 }
 
 async fn connect_modbus_tcp() -> anyhow::Result<tokio_modbus::client::Context> {

src/ems/mod.rs

@@ -11,12 +11,12 @@ pub type Consumer = tokio::sync::broadcast::Receiver<BroadcastMessage>;
 #[async_trait::async_trait]
 pub trait Service: Send + Sync {
     async fn serve(&self) {
-        let read_task = self.read_task();
-        let write_task = self.write_task();
+        let read_task = self.south();
+        let write_task = self.north();
         let _ = join!(read_task, write_task);
     }
 
-    async fn read_task(&self);
+    async fn south(&self);
 
-    async fn write_task(&self);
+    async fn north(&self);
 }

src/ems/pcs/pcs.rs

@@ -1,7 +1,7 @@
 use crate::cmd::config::app_config;
 use crate::connect_or_retry;
 use crate::ems::{Consumer, Producer, Service};
-use crate::internal::modbus::code::ModbusCode;
+use crate::internal::modbus::code::{ModbusCode, RegisterType};
 use crate::internal::modbus::{read_csv_to_code, slice_sequential};
 use crate::internal::utils;
 use anyhow::bail;
@@ -10,7 +10,6 @@ use std::net::SocketAddr;
 use std::sync::Arc;
 use tokio::sync::Mutex;
 use tokio_modbus::client::Reader;
-use tokio_modbus::FunctionCode;
 
 pub struct Pcs {
     pub id: String,
@@ -55,37 +54,31 @@ impl Pcs {
         // 从整个配置中过滤出输入寄存器集合
         let mut input_registers: Vec<_> = modbus_guard
             .iter_mut()
-            .filter(|it| {
-                it.addr.is_some()
-                    && it.name.is_some()
-                    && it.code == Some(FunctionCode::ReadInputRegisters.value())
-            })
+            .filter(|it| it.code == RegisterType::InputRegister)
             .collect();
         //连续子序列分组
         let seq_slice = slice_sequential(input_registers.as_mut_slice());
         //遍历分组
         for codes in seq_slice.into_iter() {
             if let Some(start_code) = codes.first() {
-                if let Some(start_addr) = start_code.addr {
-                    let words = self
-                        .ctx
-                        .lock()
-                        .await
-                        //读取输入寄存器
-                        .read_input_registers(start_addr, codes.len() as u16)
-                        .await??;
-                    if words.len() == codes.len() {
-                        for (i, word) in words.chunks(2).enumerate() {
-                            //大端序
-                            codes[i].data = Some([word[0] as u8, word[1] as u8]);
-                        }
-                    } else {
-                        bail!(
-                            "返回的数据长度不正确, register长度:{}, data长度:{}",
-                            input_registers.len(),
-                            words.len()
-                        )
+                let words = self
+                    .ctx
+                    .lock()
+                    .await
+                    //读取输入寄存器
+                    .read_input_registers(start_code.addr, codes.len() as u16)
+                    .await??;
+                if words.len() == codes.len() {
+                    for (i, word) in words.chunks(2).enumerate() {
+                        //大端序
+                        codes[i].data = Some([word[0] as u8, word[1] as u8]);
                     }
+                } else {
+                    bail!(
+                        "返回的数据长度不正确, register长度:{}, data长度:{}",
+                        input_registers.len(),
+                        words.len()
+                    )
                 }
             }
         }
@@ -99,7 +92,7 @@ impl Pcs {
 
 #[async_trait::async_trait]
 impl Service for Pcs {
-    async fn read_task(&self) {
+    async fn south(&self) {
         let _producer = self.producer.clone();
         loop {
             tokio::time::sleep(tokio::time::Duration::from_millis(
@@ -118,7 +111,7 @@ impl Service for Pcs {
         }
     }
 
-    async fn write_task(&self) {
+    async fn north(&self) {
         let consumer = self.consumer.clone();
         let _ctx = self.ctx.clone();
         loop {

src/internal/modbus/code.rs

@@ -1,14 +1,92 @@
-use crate::internal::modbus::{BaseModbusCode, CsvData, ModbusDataType};
+use crate::internal::modbus::{BaseModbusCode, CsvData};
+use anyhow::anyhow;
 use std::collections::HashMap;
 
+/// 寄存器类型
+#[derive(Debug, Eq, PartialEq)]
+pub enum RegisterType {
+    //线圈寄存器 0
+    Coil,
+    //离散输入寄存器 1
+    DiscreteInput,
+    //保持寄存器 2
+    HoldingRegister,
+    //输入寄存器 3
+    InputRegister,
+}
+
+impl TryFrom<u8> for RegisterType {
+    type Error = anyhow::Error;
+
+    fn try_from(value: u8) -> Result<Self, Self::Error> {
+        match value {
+            0 => Ok(RegisterType::Coil),
+            1 => Ok(RegisterType::DiscreteInput),
+            2 => Ok(RegisterType::HoldingRegister),
+            3 => Ok(RegisterType::InputRegister),
+            _ => anyhow::bail!("无效的寄存器类型！"),
+        }
+    }
+}
+
+/// 数据类型
+#[derive(Debug)]
+pub enum ModbusDataType {
+    Bit,
+    U16,
+    I16,
+    U32,
+    I32,
+    F32,
+    F64,
+}
+
+impl TryFrom<String> for ModbusDataType {
+    type Error = anyhow::Error;
+
+    fn try_from(value: String) -> Result<Self, Self::Error> {
+        match value.as_str() {
+            "bit" => Ok(ModbusDataType::Bit),
+            "u16" => Ok(ModbusDataType::U16),
+            "i16" => Ok(ModbusDataType::I16),
+            "u32" => Ok(ModbusDataType::U32),
+            "i32" => Ok(ModbusDataType::I32),
+            "f32" => Ok(ModbusDataType::F32),
+            "f64" => Ok(ModbusDataType::F64),
+            _ => anyhow::bail!("无效的数据类型！"),
+        }
+    }
+}
+
+#[derive(Debug)]
+pub enum Rw {
+    R,
+    W,
+    RW,
+}
+
+impl TryFrom<String> for Rw {
+    type Error = anyhow::Error;
+
+    fn try_from(value: String) -> Result<Self, Self::Error> {
+        match value.as_str() {
+            "r" => Ok(Rw::R),
+            "w" => Ok(Rw::W),
+            "rw" => Ok(Rw::RW),
+            _ => anyhow::bail!("无效的读写！"),
+        }
+    }
+}
+
 #[derive(Debug)]
 pub struct ModbusCode {
-    pub addr: Option<u16>,
-    pub code: Option<u8>,
+    pub addr: u16,
+    pub code: RegisterType,
     pub data: Option<[u8; 2]>,
-    pub name: Option<String>,
+    pub name: String,
     pub data_type: Option<ModbusDataType>,
     pub factor: Option<f32>,
+    pub rw: Option<Rw>,
     pub unit: Option<String>,
     pub note: Option<HashMap<i32, String>>,
     pub desc: Option<String>,
@@ -17,13 +95,13 @@ pub struct ModbusCode {
 impl ModbusCode {}
 
 impl BaseModbusCode for ModbusCode {
-    fn addr(&self) -> Option<u16> {
+    fn addr(&self) -> u16 {
         self.addr
     }
 }
 
 impl BaseModbusCode for &mut ModbusCode {
-    fn addr(&self) -> Option<u16> {
+    fn addr(&self) -> u16 {
         self.addr
     }
 }
@@ -32,20 +110,16 @@ impl TryFrom<CsvData> for ModbusCode {
     type Error = anyhow::Error;
 
     fn try_from(value: CsvData) -> Result<Self, Self::Error> {
-        let addr = value.address;
-        let code = value.code;
-        let name = value.name;
-        let mut data_type = None;
-        if let Some(ts) = value.t {
-            data_type = if ts == "bool" {
-                Some(ModbusDataType::BOOL)
-            } else if ts == "u16" {
-                Some(ModbusDataType::U16)
-            } else if ts == "s16" {
-                Some(ModbusDataType::S16)
-            } else {
-                None
-            }
+        let addr = value.address.ok_or(anyhow!("寄存器地址为空！"))?;
+        let code = value.code.ok_or(anyhow!("寄存器类型为空！"))?.try_into()?;
+        let name = value.name.ok_or(anyhow!("寄存器名称为空！"))?;
+        let data_type = match value.t {
+            None => None,
+            Some(it) => Some(it.try_into()?),
+        };
+        let rw = match value.rw {
+            None => None,
+            Some(it) => Some(it.try_into()?),
         };
         let factor = value.factor;
         let unit = value.unit;
@@ -75,6 +149,7 @@ impl TryFrom<CsvData> for ModbusCode {
             name,
             data_type,
             factor,
+            rw,
             unit,
             note,
             desc,

src/internal/modbus/mod.rs

@@ -9,9 +9,9 @@ pub struct CsvData {
     pub address: Option<u16>,
     pub code: Option<u8>,
     pub name: Option<String>,
-    pub rw: Option<u8>,
     #[serde(rename = "type")]
     pub t: Option<String>,
+    pub rw: Option<String>,
     pub factor: Option<f32>,
     pub unit: Option<String>,
     pub note: Option<String>,
@@ -26,26 +26,20 @@ pub fn read_csv_to_code(path: &str) -> anyhow::Result<Vec<ModbusCode>> {
     let mut vec = Vec::new();
     for result in rdr.deserialize() {
         let csv_data: CsvData = result?;
-        let code: ModbusCode = csv_data.try_into()?;
-        vec.push(code);
+        if let Ok(code) = csv_data.try_into() {
+            vec.push(code);
+        }
     }
     Ok(vec)
 }
 
 /// 描述了基础ModbusCode
 pub trait BaseModbusCode {
-    fn addr(&self) -> Option<u16>;
+    fn addr(&self) -> u16;
 }
 
 const MAX_REGISTER_COUNT: usize = 120;
 
-#[derive(Debug)]
-pub enum ModbusDataType {
-    BOOL,
-    U16,
-    S16,
-}
-
 /// 将序列按照连续子序列分组
 /// 如: [1,3,2,7,8,11,10] 分割为 [[1,2,3],[7,8],[10,11]]
 pub fn slice_sequential<T>(codes: &mut [T]) -> Vec<&mut [T]>
@@ -58,14 +52,12 @@ where
     while !remaining.is_empty() {
         let mut len = 1;
         while len < remaining.len() {
-            if let (Some(prev_addr), Some(curr_addr)) =
-                (remaining[len - 1].addr(), remaining[len].addr())
-            {
-                if prev_addr + 1 == curr_addr && len < MAX_REGISTER_COUNT {
-                    len += 1;
-                } else {
-                    break;
-                }
+            let prev_addr = remaining[len - 1].addr();
+            let curr_addr = remaining[len].addr();
+            if prev_addr + 1 == curr_addr && len < MAX_REGISTER_COUNT {
+                len += 1;
+            } else {
+                break;
             }
         }
         let (group, rest) = remaining.split_at_mut(len);
@@ -90,20 +82,18 @@ pub fn sequential<T: BaseModbusCode>(mut codes: Vec<T>) -> Vec<Vec<T>> {
         current_group.push(prev);
         for current in iter {
             if let Some(prev) = current_group.last() {
-                if let Some(prev_addr) = prev.addr() {
-                    if let Some(curr_addr) = current.addr() {
-                        if prev_addr + 1 == curr_addr {
-                            if current_group.len() >= MAX_REGISTER_COUNT {
-                                vec.push(current_group);
-                                current_group = Vec::new();
-                            }
-                            current_group.push(current);
-                        } else {
-                            vec.push(current_group);
-                            current_group = Vec::new();
-                            current_group.push(current)
-                        }
+                let prev_addr = prev.addr();
+                let curr_addr = current.addr();
+                if prev_addr + 1 == curr_addr {
+                    if current_group.len() >= MAX_REGISTER_COUNT {
+                        vec.push(current_group);
+                        current_group = Vec::new();
                     }
+                    current_group.push(current);
+                } else {
+                    vec.push(current_group);
+                    current_group = Vec::new();
+                    current_group.push(current)
                 }
             }
         }
@@ -116,30 +106,30 @@ pub fn sequential<T: BaseModbusCode>(mut codes: Vec<T>) -> Vec<Vec<T>> {
 mod test {
     use super::*;
 
-    struct TestModbusCode(Option<u16>);
+    struct TestModbusCode(u16);
 
     impl BaseModbusCode for TestModbusCode {
-        fn addr(&self) -> Option<u16> {
+        fn addr(&self) -> u16 {
             self.0
         }
     }
 
     #[test]
     fn test_sequential() {
-        let base = TestModbusCode { 0: Some(1u16) };
+        let base = TestModbusCode { 0: 1u16 };
         let mut codes = vec![
             base,
-            TestModbusCode { 0: Some(3u16) },
-            TestModbusCode { 0: Some(2u16) },
-            TestModbusCode { 0: Some(7u16) },
-            TestModbusCode { 0: Some(8u16) },
-            TestModbusCode { 0: Some(11u16) },
-            TestModbusCode { 0: Some(10u16) },
+            TestModbusCode { 0: 3u16 },
+            TestModbusCode { 0: 2u16 },
+            TestModbusCode { 0: 7u16 },
+            TestModbusCode { 0: 8u16 },
+            TestModbusCode { 0: 11u16 },
+            TestModbusCode { 0: 10u16 },
         ];
         let vec = slice_sequential(codes.as_mut_slice());
         let vec = vec
             .into_iter()
-            .map(|it| it.into_iter().map(|i| i.0.unwrap()).collect::<Vec<u16>>())
+            .map(|it| it.into_iter().map(|i| i.0).collect::<Vec<u16>>())
             .collect::<Vec<Vec<u16>>>();
         assert_eq!(vec, vec![vec![1, 2, 3], vec![7, 8], vec![10, 11],]);
     }