Merge branch 'feature_wp_klasse'

data/.~lock.ModBus TCPIP 1.17(1).xlsx#

@@ -1 +0,0 @@
-,nils,nils-ThinkPad-P52,17.04.2025 14:59,file:///home/nils/.config/libreoffice/4;

data_base_csv.py

@@ -0,0 +1,46 @@
+import csv
+import os
+import tempfile
+import shutil
+
+class DataBaseCsv:
+    def __init__(self, filename: str):
+        self.filename = filename
+
+    def store_data(self, data: dict):
+        new_fields = list(data.keys())
+
+        # If file does not exist or is empty → create new file with header
+        if not os.path.exists(self.filename) or os.path.getsize(self.filename) == 0:
+            with open(self.filename, mode='w', newline='') as csv_file:
+                writer = csv.DictWriter(csv_file, fieldnames=new_fields)
+                writer.writeheader()
+                writer.writerow(data)
+            return
+
+        # If file exists → read existing header and data
+        with open(self.filename, mode='r', newline='') as csv_file:
+            reader = csv.DictReader(csv_file)
+            existing_fields = reader.fieldnames
+            existing_data = list(reader)
+
+        # Merge old and new fields (keep original order, add new ones)
+        all_fields = existing_fields.copy()
+        for field in new_fields:
+            if field not in all_fields:
+                all_fields.append(field)
+
+        # Write to a temporary file with updated header
+        with tempfile.NamedTemporaryFile(mode='w', delete=False, newline='', encoding='utf-8') as tmp_file:
+            writer = csv.DictWriter(tmp_file, fieldnames=all_fields)
+            writer.writeheader()
+
+            # Write old rows with updated field list
+            for row in existing_data:
+                writer.writerow({field: row.get(field, '') for field in all_fields})
+
+            # Write new data row
+            writer.writerow({field: data.get(field, '') for field in all_fields})
+
+        # Replace original file with updated temporary file
+        shutil.move(tmp_file.name, self.filename)

heat_pump.py

@@ -0,0 +1,62 @@
+from pymodbus.client import ModbusTcpClient
+import pandas as pd
+import time
+
+class HeatPump:
+    def __init__(self, ip_address: str):
+        self.ip = ip_address
+        self.client = None
+        self.connect_to_modbus()
+        self.registers = None
+        self.get_registers()
+
+    def connect_to_modbus(self):
+        port = 502
+        self.client = ModbusTcpClient(self.ip, port=port)
+        try:
+            if not self.client.connect():
+                print("Verbindung zur Wärmepumpe fehlgeschlagen.")
+                exit(1)
+            print("Verbindung zur Wärmepumpe erfolgreich.")
+        except KeyboardInterrupt:
+            print("Beendet durch Benutzer (Ctrl+C).")
+        finally:
+            self.client.close()
+
+    def get_registers(self):
+        # Excel-Datei mit den Input-Registerinformationen
+        excel_path = "data/ModBus TCPIP 1.17(1).xlsx"
+        xls = pd.ExcelFile(excel_path)
+        df_input_registers = xls.parse('04 Input Register')
+
+        # Relevante Spalten bereinigen
+        df_clean = df_input_registers[['MB Adresse', 'Variable', 'Beschreibung', 'Variabel Typ']].dropna()
+        df_clean['MB Adresse'] = df_clean['MB Adresse'].astype(int)
+
+        # Dictionary aus Excel erzeugen
+        self.registers = {
+            row['MB Adresse']: {
+                'desc': row['Beschreibung'],
+                'type': 'REAL' if row['Variabel Typ'] == 'REAL' else 'INT'
+            }
+            for _, row in df_clean.iterrows()
+        }
+
+    def get_data(self):
+        data = {}
+        data['Zeit'] = time.strftime('%Y-%m-%d %H:%M:%S')
+        for address, info in self.registers.items():
+            reg_type = info['type']
+            result = self.client.read_input_registers(address, count=2 if reg_type == 'REAL' else 1)
+            if result.isError():
+                print(f"Fehler beim Lesen von Adresse {address}: {result}")
+                continue
+
+            if reg_type == 'REAL':
+                value = result.registers[0] / 10.0
+            else:
+                value = result.registers[0]
+
+            print(f"Adresse {address} - {info['desc']}: {value}")
+            data[f"{address} - {info['desc']}"] = value
+        return data

main.py

@@ -1,75 +1,17 @@
-from pymodbus.client import ModbusTcpClient
-import pandas as pd
-import struct
 import time
-import csv
+from datetime import datetime
-import os
+from data_base_csv import DataBaseCsv
+from heat_pump import HeatPump
 
-# Excel-Datei mit den Input-Registerinformationen
+interval = 10  # z.B. alle 10 Sekunden
-excel_path = "data/ModBus TCPIP 1.17(1).xlsx"
-xls = pd.ExcelFile(excel_path)
-df_input_registers = xls.parse('04 Input Register')
 
-# Relevante Spalten bereinigen
+db = DataBaseCsv('modbus_log.csv')
-df_clean = df_input_registers[['MB Adresse', 'Variable', 'Beschreibung', 'Variabel Typ']].dropna()
+hp = HeatPump(ip_address='10.0.0.10')
-df_clean['MB Adresse'] = df_clean['MB Adresse'].astype(int)
-
-# Dictionary aus Excel erzeugen
-registers = {
-    row['MB Adresse']: {
-        'desc': row['Beschreibung'],
-        'type': 'REAL' if row['Variabel Typ'] == 'REAL' else 'INT'
-    }
-    for _, row in df_clean.iterrows()
-}
-
-# CSV-Datei vorbereiten
-csv_filename = "modbus_log.csv"
-csv_exists = os.path.isfile(csv_filename)
-
-# Verbindung zur Wärmepumpe
-ip = '10.0.0.10'
-port = 502
-unit_id = 1
-
-client = ModbusTcpClient(ip, port=port)
-
-try:
-    if not client.connect():
-        print("Verbindung zur Wärmepumpe fehlgeschlagen.")
-        exit(1)
-    print("Verbindung zur Wärmepumpe erfolgreich.")
-
-    with open(csv_filename, mode='a', newline='') as file:
-        writer = csv.writer(file)
-        if not csv_exists:
-            header = ['Zeit'] + [f"{address} - {info['desc']}" for address, info in registers.items()]
-            writer.writerow(header)
 
 while True:
-            print(f"\n--- Neue Abfrage --- {time.strftime('%Y-%m-%d %H:%M:%S')} ---")
+    now = datetime.now()
-            row = [time.strftime('%Y-%m-%d %H:%M:%S')]
+    if now.second % interval == 0 and now.microsecond < 100_000:
-            for address, info in registers.items():
+       db.store_data(hp.get_data())
-                reg_type = info['type']
-                result = client.read_input_registers(address, count=2 if reg_type == 'REAL' else 1)
-                if result.isError():
-                    print(f"Fehler beim Lesen von Adresse {address}: {result}")
-                    row.append('Fehler')
-                    continue
 
-                if reg_type == 'REAL':
+    time.sleep(0.1)
-                    value = result.registers[0] / 10.0
-                else:
-                    value = result.registers[0]
 
-                print(f"Adresse {address} - {info['desc']}: {value}")
-                row.append(value)
-
-            writer.writerow(row)
-            file.flush()
-            time.sleep(10)
-
-except KeyboardInterrupt:
-    print("Beendet durch Benutzer (Ctrl+C).")
-finally:
-    client.close()