|
|
@@ -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()
|
