Beta (not stable)

optimized_ecg.py

@@ -0,0 +1,409 @@
+import customtkinter as ctk
+import time
+import math
+import random
+import csv
+import tkinter as tk
+import customtkinter as ctk
+import time
+
+
+## cc maximilian scheinast-peter
+## last update 11.04.2024
+## dieses programm funktioniert ähnlich wie ein digitales Oszi zur Darstellung von Vitalkurven
+## Attribution-NonCommercial license
+
+def get_data1():
+    import serial
+    import time
+
+    # Configure serial port (adjust port name as needed)
+    ser = serial.Serial('COM10', 9600)
+
+    # Wait for the serial connection to establish
+    ser.timeout = 2
+
+    #ser.write(b'testcom.py\n')  # Send command to start the script
+    while True:
+        # Read data from serial port
+        data = ser.readline().decode().strip()
+
+        # Check if data is not empty
+        #print(data)
+        if data:
+            try:
+                # Attempt to convert data to float
+                value = float(data)
+                timestamp = time.time()
+                return timestamp, value
+            except ValueError:
+                print("Invalid data format:", data)
+        else:
+            print("Empty data received")
+        
+        # Add a small delay to prevent rapid looping
+        time.sleep(0.001)
+
+def get_data():  # example heartfunction
+    global i
+    i=i+1
+    ###print(i)
+    timestamp = time.time()  ## dont change
+
+    ## logic for get one datapoint like a read_value function
+    t = (timestamp % 1) * 2 * math.pi
+    p_wave = 5 * math.sin(t)
+    qrs_complex = 40 * math.sin(1.5 * t) * math.exp(-0.25 * t ** 2)
+    t_wave = 10 * math.sin(2 * t) * math.exp(-0.5 * t ** 2)
+    value = 50 + p_wave + qrs_complex + t_wave
+    value = max(0, min(value, 100))
+    wait_for_next_millisecond()
+
+    return timestamp, value
+
+
+def wait_for_next_millisecond():
+    """Waits until the next full millisecond."""
+    current_time = time.time()
+    next_millisecond = (int(current_time * 1000) + 1) / 1000
+    time.sleep(next_millisecond - current_time)
+
+
+# Function to generate realistic ECG data
+
+
+class EKGApp:
+    def __init__(self, master):
+        self.master = master
+        master.title("EKG Visualization")
+
+        # Main Frame
+        main_frame = ctk.CTkFrame(master)
+        main_frame.pack(fill=ctk.BOTH, expand=True)
+
+        # Single Sweep Canvas
+        self.single_canvas = ctk.CTkCanvas(main_frame, width=600, height=300)
+        self.single_canvas.pack(side=ctk.LEFT, fill=ctk.BOTH, expand=True)
+
+        # Multi Sweep Canvas
+        self.multi_canvas = ctk.CTkCanvas(main_frame, width=600, height=300)
+        self.multi_canvas.pack(side=ctk.LEFT, fill=ctk.BOTH, expand=True)
+
+        # Input Frame
+        input_frame = ctk.CTkFrame(main_frame)
+        input_frame.pack(side=ctk.RIGHT, fill=ctk.Y)
+
+        # Data Structures
+        self.single_data = []
+        self.multi_data = []
+        self.logged_data = []
+
+        # Time Variables
+        self.start_time = time.time()
+        self.multi_sweep_duration = 10  # Initial Laufbanddauer
+
+        # Trigger Variables
+        self.trigger_level = 50
+        self.trigger_armed = False
+        self.trigger_paused = False
+        self.trigger_timestamp = None
+        self.last_trigger_timestamp = None
+        self.last_value = 0
+        self.trigger_count = 0
+
+        # Cooldown Variables
+        self.cooldown_time = 0
+        self.cooldown_active = False
+
+        # Logging Variables
+        self.logging_active = False
+
+        # Batching Variables
+        self.batch_size = 5
+        self.current_batch = []
+
+        # Input Fields and Buttons (with consistent size)
+        self.create_input_fields(input_frame)
+
+        # Logging Frame
+        logging_frame = ctk.CTkFrame(master)
+        logging_frame.pack(fill=ctk.X)
+        self.create_logging_buttons(logging_frame)
+
+        # Update Data Periodically
+        self.update_data()
+
+        # Resize Handling
+        self.single_canvas.bind("<Configure>", self.on_resize)
+        self.multi_canvas.bind("<Configure>", self.on_resize)
+
+    def create_input_fields(self, frame):
+        # Consistent width for all input elements
+        input_width = 150
+
+        # Trigger Level
+        ctk.CTkLabel(frame, text="Trigger Level:").pack()
+        self.trigger_entry = ctk.CTkEntry(frame, width=input_width)
+        self.trigger_entry.insert(0, str(self.trigger_level))
+        self.trigger_entry.pack()
+
+        # Flank Selection
+        ctk.CTkLabel(frame, text="Flanke:").pack()
+        self.flank_var = ctk.StringVar(value="Steigende Flanke")
+        self.flank_combobox = ctk.CTkComboBox(frame, variable=self.flank_var,
+                                              values=["Steigende Flanke", "Fallende Flanke"], width=input_width)
+        self.flank_combobox.pack()
+
+        # Cooldown
+        ctk.CTkLabel(frame, text="Cooldown (Sekunden):").pack()
+        self.cooldown_entry = ctk.CTkEntry(frame, width=input_width)
+        self.cooldown_entry.insert(0, "0")
+        self.cooldown_entry.pack()
+        self.cooldown_button = ctk.CTkButton(frame, text="Cooldown setzen", command=self.set_cooldown,
+                                             width=input_width)
+        self.cooldown_button.pack()
+
+        # Ax Time (X-Achsen-Zeit)
+        ctk.CTkLabel(frame, text="Ax-Zeit (Sekunden):").pack()
+        self.ax_time_entry = ctk.CTkEntry(frame, width=input_width)
+        self.ax_time_entry.insert(0, str(self.multi_sweep_duration))  # Initial value
+        self.ax_time_entry.pack()
+        self.ax_time_button = ctk.CTkButton(frame, text="Ax-Zeit setzen", command=self.set_ax_time, width=input_width)
+        self.ax_time_button.pack()
+
+        # Trigger Buttons
+        self.trigger_button = ctk.CTkButton(frame, text="Trigger starten", command=self.toggle_trigger,
+                                            width=input_width)
+        self.trigger_button.pack()
+        self.pause_button = ctk.CTkButton(frame, text="Trigger pausieren", command=self.pause_trigger,
+                                          state=ctk.DISABLED, width=input_width)
+        self.pause_button.pack()
+
+        # Multi Sweep Duration (Laufbanddauer)
+        ctk.CTkLabel(frame, text="Laufbanddauer (Sekunden):").pack()
+        self.duration_entry = ctk.CTkEntry(frame, width=input_width)
+        self.duration_entry.insert(0, "30")
+        self.duration_entry.pack()
+        self.duration_button = ctk.CTkButton(frame, text="Dauer setzen", command=self.set_duration, width=input_width)
+        self.duration_button.pack()
+
+    def set_ax_time(self):
+
+        try:
+            self.multi_sweep_duration = float(self.ax_time_entry.get())
+        except ValueError:
+            pass
+
+    def create_logging_buttons(self, frame):
+        # Consistent width for logging buttons
+        button_width = 120
+
+        self.start_logging_button = ctk.CTkButton(frame, text="Start Logging", command=self.start_logging,
+                                                  width=button_width)
+        self.start_logging_button.pack(side=ctk.LEFT)
+        self.end_logging_button = ctk.CTkButton(frame, text="End Logging", command=self.end_logging, state=ctk.DISABLED,
+                                                width=button_width)
+        self.end_logging_button.pack(side=ctk.LEFT)
+        self.save_data_button = ctk.CTkButton(frame, text="Save Data", command=self.save_data, state=ctk.DISABLED,
+                                              width=button_width)
+        self.save_data_button.pack(side=ctk.LEFT)
+
+    def set_cooldown(self):
+        try:
+            self.cooldown_time = float(self.cooldown_entry.get())
+        except ValueError:
+            pass
+
+    def set_duration(self):
+        try:
+            self.multi_sweep_duration = float(self.duration_entry.get())
+        except ValueError:
+            pass
+
+    def toggle_trigger(self):
+        if self.trigger_armed:
+            self.trigger_armed = False
+            self.trigger_button.configure(text="Trigger starten")
+            self.pause_button.configure(state=ctk.DISABLED)
+            self.single_data = []
+            self.draw_single_canvas()
+        else:
+            try:
+                self.trigger_level = float(self.trigger_entry.get())
+            except ValueError:
+                pass
+            self.trigger_armed = True
+            self.trigger_button.configure(text="Trigger stoppen")
+            self.pause_button.configure(state=ctk.NORMAL)
+            self.trigger_count = 0
+
+    def pause_trigger(self):
+        self.trigger_paused = not self.trigger_paused
+        if self.trigger_paused:
+            self.pause_button.configure(text="Trigger fortsetzen")
+        else:
+            self.pause_button.configure(text="Trigger pausieren")
+
+    def start_logging(self):
+        self.logging_active = True
+        self.logged_data = []
+        self.start_logging_button.configure(state=ctk.DISABLED)
+        self.end_logging_button.configure(state=ctk.NORMAL)
+
+    def end_logging(self):
+        self.logging_active = False
+        self.start_logging_button.configure(state=ctk.NORMAL)
+        self.end_logging_button.configure(state=ctk.DISABLED)
+        self.save_data_button.configure(state=ctk.NORMAL)
+
+    def save_data(self):
+        # Popup for file name
+        ####print(123)
+        file_name = ctk.CTkInputDialog(text="Enter file name:", title="Save Data").get_input()
+        if file_name:
+            try:
+                with open(f"{file_name}.csv", "w", newline="", encoding='utf-8') as csvfile:
+                    writer = csv.writer(csvfile)
+                    writer.writerow(["Timestamp", "Value"])
+                    writer.writerows(self.logged_data)
+                self.save_data_button.configure(state=ctk.DISABLED)
+            except Exception as e:
+                print(f"Error saving data: {e}")
+
+    def update_data(self):
+
+        # Get EKG Data
+        timestamp, value = get_data()
+
+        # Collect data points into batches
+        self.current_batch.append((timestamp, value))
+
+        if len(self.current_batch) == self.batch_size:
+            # Process the batch (e.g., plot it)
+            #self.multi_data.append((timestamp, value))
+            self.process_batch(self.current_batch)
+            
+            # Reset the current batch
+            self.current_batch = []
+
+
+
+        # Update Multi Sweep Data
+        #self.multi_data.append((timestamp, value))
+        try:
+            if timestamp - self.multi_data[0][0] > self.multi_sweep_duration:
+                self.multi_data.pop(0)
+        except:
+            pass
+        # Trigger Logic
+        if self.trigger_armed and not self.trigger_paused and not self.cooldown_active:
+            condition = (
+                        value >= self.trigger_level and self.last_value < self.trigger_level) if self.flank_var.get() == "Steigende Flanke" else (
+                        value <= self.trigger_level and self.last_value > self.trigger_level)
+            if condition:
+                self.last_trigger_timestamp = timestamp
+                self.single_data = []
+                self.trigger_count = 0
+                if self.cooldown_time > 0:
+                    self.cooldown_active = True
+                    self.master.after(int(self.cooldown_time * 1000), self.end_cooldown)
+
+        # Update Single Sweep Data
+        if self.last_trigger_timestamp is not None and self.trigger_armed and not self.trigger_paused:
+            self.single_data.append((timestamp, value))
+
+        # Update Logged Data
+        if self.logging_active:
+            self.logged_data.append((timestamp, value))
+
+        
+        
+
+        # Store last value and update trigger count
+        self.last_value = value
+        if self.trigger_armed and not self.trigger_paused:
+            self.trigger_count += 1
+
+    
+
+        # Call again after 1ms
+        self.master.after(1,self.update_data)
+
+    def end_cooldown(self):
+        self.cooldown_active = False
+
+    def process_batch(self, batch):
+        global i
+        i=i+1
+        #print(i)
+        # Example: Calculate average value for the batch
+        total_value = sum(value for _, value in batch)
+        average_value = total_value / len(batch)
+        for timestamp, value in batch:
+            self.multi_data.append((timestamp, value))
+        ###print(f"Average value for batch: {average_value}")
+        self.draw_multi_canvas()
+        # Draw Canvases
+        self.draw_single_canvas()
+        # You can implement your own logic here, such as plotting the batch
+        # on a separate canvas or performing other calculations.
+
+    def draw_single_canvas(self):
+        self.single_canvas.delete("all")
+        if not self.single_data or self.last_trigger_timestamp is None:
+            return
+        
+        # Prepare coordinates for create_line
+        coords = []
+        x_scale = self.single_canvas.winfo_width() / self.multi_sweep_duration
+        y_scale = self.single_canvas.winfo_height() / 100
+        last_x, last_y = None, None
+        
+        for timestamp, value in self.single_data:
+            x = (timestamp - self.last_trigger_timestamp) * x_scale
+            y = self.single_canvas.winfo_height() - (value * self.single_canvas.winfo_height() / 100)
+            if last_x is not None:
+                coords.extend([last_x, last_y, x, y])
+            last_x, last_y = x, y
+
+        # Draw the lines
+        if coords:
+            self.single_canvas.create_line(*coords, fill="blue")
+
+
+    def draw_multi_canvas(self):
+        self.multi_canvas.delete("all")
+        if not self.multi_data:
+            return
+
+        duration = max(self.multi_data[-1][0] - self.multi_data[0][0], 0.001)
+        x_scale = self.multi_canvas.winfo_width() / duration
+        y_scale = self.multi_canvas.winfo_height() / 100
+
+        coords = []
+        for timestamp, value in self.multi_data:
+            x = (timestamp - self.multi_data[0][0]) * x_scale
+            y = self.multi_canvas.winfo_height() - (value * y_scale)
+            coords.extend([x, y])
+
+        self.multi_canvas.create_line(*coords, fill="blue")
+        trigger_y = self.multi_canvas.winfo_height() - (self.trigger_level * y_scale)
+        self.multi_canvas.create_line(self.multi_canvas.winfo_width() - 10, trigger_y, self.multi_canvas.winfo_width(), trigger_y + 5, fill="red")
+        self.multi_canvas.create_line(self.multi_canvas.winfo_width() - 10, trigger_y, self.multi_canvas.winfo_width(), trigger_y - 5, fill="red")
+
+    def draw_time_axis(self, canvas, start_time, duration):
+        for i in range(int(duration) + 1):
+            x = i * canvas.winfo_width() / duration
+            canvas.create_line(x, 0, x, canvas.winfo_height(), fill="gray", dash=(2, 2))
+            canvas.create_text(x, canvas.winfo_height() - 10, text=f"{i:.1f}s", anchor=ctk.N)
+
+    def on_resize(self, event):
+        self.draw_single_canvas()
+        self.draw_multi_canvas()
+i=0
+if __name__ == "__main__":
+    ctk.set_appearance_mode("dark")  # Modes: "System" (standard), "Dark", "Light"
+    ctk.set_default_color_theme("blue")  # Themes: "blue" (standard), "green", "dark-blue"
+    root = ctk.CTk()
+    app = EKGApp(root)
+    root.mainloop()