Konfigurationen ausgelagert, modbus-Thread wird gerufen

src/config.rs

@@ -0,0 +1,97 @@
+use serde::{Deserialize, Serialize};
+use std::collections::HashMap;
+
+#[derive(Debug, Serialize, Deserialize, Clone)]
+pub struct MqttConfig {
+    pub broker: String,
+    pub port: u16,
+    pub user: Option<String>,
+    pub password: Option<String>,
+    pub button_circulation: Option<String>,
+}
+
+#[derive(Debug, Serialize, Deserialize, Clone)]
+pub struct InfluxConfig {
+    pub bucket: String,
+    pub org: String,
+    pub token: String,
+    pub url: String,
+    pub location: Option<String>,
+    pub measurement: Option<String>,
+}
+
+#[derive(Debug, Serialize, Deserialize, Clone)]
+pub struct ModbusRegisterConfig {
+    pub addr: u16,
+    pub r#type: Option<String>,
+    pub factor: Option<f64>,
+    pub mqtt: Option<bool>,
+    pub influxdb: Option<bool>,
+    pub comment: Option<String>,
+}
+
+#[derive(Debug, Serialize, Deserialize, Clone)]
+pub struct ModbusConfig {
+    pub host: String,
+    pub port: u16,
+    pub max_coils_addr: Option<u16>,
+    pub max_input_addr: Option<u16>,
+    pub max_holding_addr: Option<u16>,
+}
+
+#[derive(Debug, Serialize, Deserialize, Clone)]
+pub struct DefaultConfig {
+    pub loglevel: Option<String>,
+}
+
+#[derive(Debug, Serialize, Deserialize, Clone)]
+pub struct AppConfig {
+    pub default: DefaultConfig,
+    pub mqtt: MqttConfig,
+    pub influxdb: InfluxConfig,
+    pub modbus: ModbusConfig,
+    pub modbus_coils: Option<Vec<HashMap<String, ModbusRegisterConfig>>>,
+    pub modbus_input_register: Option<Vec<HashMap<String, ModbusRegisterConfig>>>,
+    pub modbus_holding_register: Option<Vec<HashMap<String, ModbusRegisterConfig>>>,
+}
+
+#[derive(Debug, Default, Clone)]
+pub struct ModbusValueMaps {
+    pub modbus_coils_values: HashMap<String, f64>,
+    pub modbus_input_register_values: HashMap<String, f64>,
+    pub modbus_holding_register_values: HashMap<String, f64>,
+}
+
+impl ModbusValueMaps {
+    pub fn from_config(config: &AppConfig) -> Self {
+        let mut coils = HashMap::new();
+        if let Some(ref vec) = config.modbus_coils {
+            for map in vec {
+                for key in map.keys() {
+                    coils.entry(key.clone()).or_insert(0.0);
+                }
+            }
+        }
+        let mut input = HashMap::new();
+        if let Some(ref vec) = config.modbus_input_register {
+            for map in vec {
+                for key in map.keys() {
+                    input.entry(key.clone()).or_insert(0.0);
+                }
+            }
+        }
+        let mut holding = HashMap::new();
+        if let Some(ref vec) = config.modbus_holding_register {
+            for map in vec {
+                for key in map.keys() {
+                    holding.entry(key.clone()).or_insert(0.0);
+                }
+            }
+        }
+        ModbusValueMaps {
+            modbus_coils_values: coils,
+            modbus_input_register_values: input,
+            modbus_holding_register_values: holding,
+        }
+    }
+}

src/main.rs

@@ -1,79 +1,23 @@
+
 use actix_web::{web, App, HttpResponse, HttpServer, Result};
 use actix_files as actix_fs;
-use serde::{Deserialize, Serialize};
+use std::sync::{Mutex, Arc};
-use std::sync::Mutex;
 use tera::{Context, Tera};
-// use config::{Config, File};
 use std::fs;
 use serde_yaml;
 use std::collections::HashMap;
+use serde::{Serialize, Deserialize};
+mod config;
+mod modbus;
+use crate::config::{AppConfig, ModbusConfig, ModbusRegisterConfig, DefaultConfig, MqttConfig, InfluxConfig, ModbusValueMaps};
 
 
-#[derive(Debug, Serialize, Deserialize, Clone)]
-pub struct MqttConfig {
-    pub broker: String,
-    pub port: u16,
-    pub user: Option<String>,
-    pub password: Option<String>,
-    pub button_circulation: Option<String>,
-}
 
-#[derive(Debug, Serialize, Deserialize, Clone)]
+// ...existing code...
-pub struct InfluxConfig {
-    pub bucket: String,
-    pub org: String,
-    pub token: String,
-    pub url: String,
-    pub location: Option<String>,
-    pub measurement: Option<String>,
-}
-
-#[derive(Debug, Serialize, Deserialize, Clone)]
-pub struct ModbusRegisterConfig {
-    pub addr: u16,
-    pub r#type: Option<String>,
-    pub factor: Option<f64>,
-    pub mqtt: Option<bool>,
-    pub influxdb: Option<bool>,
-    pub comment: Option<String>,
-}
-
-#[derive(Debug, Serialize, Deserialize, Clone)]
-pub struct ModbusConfig {
-    pub host: String,
-    pub port: u16,
-    pub max_coils_addr: Option<u16>,
-    pub max_input_addr: Option<u16>,
-    pub max_holding_addr: Option<u16>,
-}
-
-#[derive(Debug, Serialize, Deserialize, Clone)]
-pub struct DefaultConfig {
-    pub loglevel: Option<String>,
-}
-
-#[derive(Debug, Serialize, Deserialize, Clone)]
-pub struct AppConfig {
-    pub default: DefaultConfig,
-    pub mqtt: MqttConfig,
-    pub influxdb: InfluxConfig,
-    pub modbus: ModbusConfig,
-    pub modbus_coils: Option<Vec<HashMap<String, ModbusRegisterConfig>>>,
-    pub modbus_input_register: Option<Vec<HashMap<String, ModbusRegisterConfig>>>,
-    pub modbus_holding_register: Option<Vec<HashMap<String, ModbusRegisterConfig>>>,
-}
-
-/// Werthaltung für berechnete Modbus-Werte
-#[derive(Debug, Default, Clone)]
-pub struct ModbusValueMaps {
-    pub modbus_coils_values: std::collections::HashMap<String, f64>,
-    pub modbus_input_register_values: std::collections::HashMap<String, f64>,
-    pub modbus_holding_register_values: std::collections::HashMap<String, f64>,
-}
 
 struct AppState {
     config: Mutex<AppConfig>,
-    value_maps: Mutex<ModbusValueMaps>,
+    value_maps: Arc<Mutex<ModbusValueMaps>>,
     templates: Tera,
 }
 
@@ -82,7 +26,7 @@ impl AppState {
         let conf_str = std::fs::read_to_string(conf_path).expect("Config-Datei konnte nicht gelesen werden");
         let config: AppConfig = serde_yaml::from_str(&conf_str).expect("Config-Deserialisierung fehlgeschlagen");
 
-        let value_maps = Mutex::new(ModbusValueMaps::from_config(&config));
+        let value_maps = Arc::new(Mutex::new(ModbusValueMaps::from_config(&config)));
 
         let tera = match Tera::new("templates/**/*") {
             Ok(t) => t,
@@ -100,39 +44,8 @@ impl AppState {
     }
 }
 
-impl ModbusValueMaps {
+
-    pub fn from_config(config: &AppConfig) -> Self {
+// ...existing code...
-        let mut coils = HashMap::new();
-        if let Some(ref vec) = config.modbus_coils {
-            for map in vec {
-                for key in map.keys() {
-                    coils.entry(key.clone()).or_insert(0.0);
-                }
-            }
-        }
-        let mut input = HashMap::new();
-        if let Some(ref vec) = config.modbus_input_register {
-            for map in vec {
-                for key in map.keys() {
-                    input.entry(key.clone()).or_insert(0.0);
-                }
-            }
-        }
-        let mut holding = HashMap::new();
-        if let Some(ref vec) = config.modbus_holding_register {
-            for map in vec {
-                for key in map.keys() {
-                    holding.entry(key.clone()).or_insert(0.0);
-                }
-            }
-        }
-        ModbusValueMaps {
-            modbus_coils_values: coils,
-            modbus_input_register_values: input,
-            modbus_holding_register_values: holding,
-        }
-    }
-}
 
 async fn index(data: web::Data<AppState>) -> Result<HttpResponse> {
     let config = data.config.lock().unwrap();
@@ -274,9 +187,24 @@ async fn save_settings(
 #[actix_web::main]
 async fn main() -> std::io::Result<()> {
 
+
     let config_path = "paramod.yaml";
     let app_state = web::Data::new(AppState::load_from_conf(config_path));
 
+    // Starte Modbus-Polling-Thread
+    {
+        let config = app_state.config.lock().unwrap().clone();
+        let value_maps = Arc::clone(&app_state.value_maps);
+        modbus::start_modbus_polling_thread(
+            &config.modbus,
+            &config.modbus_input_register,
+            &config.modbus_holding_register,
+            &config.modbus_coils,
+            value_maps,
+            std::time::Duration::from_secs(2), // Poll-Intervall
+        );
+    }
+
     println!("Server läuft auf http://0.0.0.0:8080");
 
     HttpServer::new(move || {

src/modbus.rs

@@ -0,0 +1,69 @@
+use crate::config::{ModbusValueMaps, ModbusRegisterConfig, ModbusConfig};
+use std::collections::HashMap;
+use std::sync::{Arc, Mutex};
+use std::thread;
+use std::time::Duration;
+
+/// Startet einen Thread, der zyklisch Modbus-Register abfragt und ModbusValueMaps aktualisiert
+pub fn start_modbus_polling_thread(
+    modbus_config: &ModbusConfig,
+    input_registers: &Option<Vec<HashMap<String, ModbusRegisterConfig>>>,
+    holding_registers: &Option<Vec<HashMap<String, ModbusRegisterConfig>>>,
+    coils: &Option<Vec<HashMap<String, ModbusRegisterConfig>>>,
+    value_maps: Arc<Mutex<ModbusValueMaps>>,
+    poll_interval: Duration,
+) {
+    let input_registers = input_registers.clone();
+    let holding_registers = holding_registers.clone();
+    let coils = coils.clone();
+    thread::spawn(move || {
+        loop {
+            // Input Register
+            if let Some(ref vec) = input_registers {
+                for map in vec {
+                    for (key, reg) in map {
+                        let typ = reg.r#type.as_deref().unwrap_or("");
+                        let value = match typ {
+                            "INT16" => { let raw: i16 = 0; raw as f64 },
+                            "UINT16" => { let raw: u16 = 0; raw as f64 },
+                            "UINT32" => { let raw1: u16 = 0; let raw2: u16 = 0; let raw32: u32 = ((raw1 as u32) << 16) | (raw2 as u32); raw32 as f64 },
+                            _ => 0.0,
+                        };
+                        if let Ok(mut maps) = value_maps.lock() {
+                            maps.modbus_input_register_values.insert(key.clone(), value);
+                        }
+                    }
+                }
+            }
+            // Holding Register
+            if let Some(ref vec) = holding_registers {
+                for map in vec {
+                    for (key, reg) in map {
+                        let typ = reg.r#type.as_deref().unwrap_or("");
+                        let value = match typ {
+                            "INT16" => { let raw: i16 = 0; raw as f64 },
+                            "UINT16" => { let raw: u16 = 0; raw as f64 },
+                            "UINT32" => { let raw1: u16 = 0; let raw2: u16 = 0; let raw32: u32 = ((raw1 as u32) << 16) | (raw2 as u32); raw32 as f64 },
+                            _ => 0.0,
+                        };
+                        if let Ok(mut maps) = value_maps.lock() {
+                            maps.modbus_holding_register_values.insert(key.clone(), value);
+                        }
+                    }
+                }
+            }
+            // Coils
+            if let Some(ref vec) = coils {
+                for map in vec {
+                    for (key, _reg) in map {
+                        let value = 0.0; // Hier Coil-Wert abfragen
+                        if let Ok(mut maps) = value_maps.lock() {
+                            maps.modbus_coils_values.insert(key.clone(), value);
+                        }
+                    }
+                }
+            }
+            thread::sleep(poll_interval);
+        }
+    });
+}

src/modbus_thread.rs

@@ -1,100 +0,0 @@
-use std::collections::HashMap;
-use std::sync::{Arc, Mutex};
-use std::time::Duration;
-use tokio::sync::Notify;
-use tokio::task::JoinHandle;
-use tokio_modbus::client::tcp;
-use tokio_modbus::prelude::*;
-use log::{info, error, debug};
-
-// Platzhalter für Konfiguration
-#[derive(Clone)]
-pub struct ModbusConfig {
-    pub modbus_host: String,
-    pub modbus_port: u16,
-    pub modbus_max_holding_addr: u16,
-    pub modbus_max_input_addr: u16,
-    // Register-Konfigurationen als Map: key -> (addr, type, factor, mqtt, influxdb)
-    pub modbus_input_register: HashMap<String, RegisterConfig>,
-    pub modbus_holding_register: HashMap<String, RegisterConfig>,
-}
-
-#[derive(Clone)]
-pub struct RegisterConfig {
-    pub addr: u16,
-    pub reg_type: String, // z.B. "INT16", "UINT16"
-    pub factor: f64,
-    pub mqtt: bool,
-    pub influxdb: bool,
-}
-
-const NONSET_INT16: i16 = -32768;
-const NONSET_UINT16: u16 = 65535;
-
-pub struct ModbusThread {
-    config: ModbusConfig,
-    timeout: Duration,
-    polling_interval: Duration,
-    min_write_interval: Duration,
-    running: Arc<Notify>,
-    stop_notify: Arc<Notify>,
-    // Register-Cache
-    input_register: Arc<Mutex<Vec<u16>>>,
-    holding_register: Arc<Mutex<Vec<u16>>>,
-    // Schreib-Queue
-    write_holding_register: Arc<Mutex<HashMap<u16, u16>>>,
-    // Async-Task-Handle
-    handle: Option<JoinHandle<()>>,
-}
-
-impl ModbusThread {
-    pub fn new(config: ModbusConfig, timeout: f64, polling_interval: f64, min_write_interval: f64) -> Self {
-        Self {
-            config,
-            timeout: Duration::from_secs_f64(timeout),
-            polling_interval: Duration::from_secs_f64(polling_interval),
-            min_write_interval: Duration::from_secs_f64(min_write_interval),
-            running: Arc::new(Notify::new()),
-            stop_notify: Arc::new(Notify::new()),
-            input_register: Arc::new(Mutex::new(Vec::new())),
-            holding_register: Arc::new(Mutex::new(Vec::new())),
-            write_holding_register: Arc::new(Mutex::new(HashMap::new())),
-            handle: None,
-        }
-    }
-
-    pub fn start(&mut self) {
-        let config = self.config.clone();
-        let timeout = self.timeout;
-        let polling_interval = self.polling_interval;
-        let min_write_interval = self.min_write_interval;
-        let running = self.running.clone();
-        let stop_notify = self.stop_notify.clone();
-        let input_register = self.input_register.clone();
-        let holding_register = self.holding_register.clone();
-        let write_holding_register = self.write_holding_register.clone();
-
-        self.handle = Some(tokio::spawn(async move {
-            let mut write_countdown = min_write_interval;
-            let socket_addr = format!("{}:{}", config.modbus_host, config.modbus_port);
-            loop {
-                // Stop-Check
-                if stop_notify.is_notified() {
-                    break;
-                }
-                // Connect
-                let mut ctx = match tcp::connect(socket_addr.clone()).await {
-                    Ok(c) => {
-                        info!("Modbus: Verbindung hergestellt");
-                        c
-                    }
-                    Err(e) => {
-                        error!("Modbus: Verbindungsfehler: {:?}", e);
-                        tokio::time::sleep(Duration::from_secs(2)).await;
-                        continue;
-                    }
-                };
-                // Polling-Loop
-                loop {
-                    if stop_notify.is_notified() {
-                        let _ =