Jan
f50a641211
Make all measurements update live in the GUI
Allow the stop button to immediately halt the test at any point in the cycle
Still maintain proper cleanup and data saving when stopped
(Deepseek)
880 lines
38 KiB
Python
880 lines
38 KiB
Python
# -*- coding: utf-8 -*-
|
|
import os
|
|
import time
|
|
import csv
|
|
import threading
|
|
from datetime import datetime
|
|
import tkinter as tk
|
|
from tkinter import ttk, messagebox
|
|
import pysmu
|
|
import numpy as np
|
|
import matplotlib
|
|
matplotlib.use('TkAgg')
|
|
from matplotlib.backends.backend_tkagg import FigureCanvasTkAgg
|
|
from matplotlib.figure import Figure
|
|
from collections import deque
|
|
|
|
class DeviceDisconnectedError(Exception):
|
|
pass
|
|
|
|
class BatteryTester:
|
|
def __init__(self, root):
|
|
# Color scheme
|
|
self.bg_color = "#2E3440"
|
|
self.fg_color = "#D8DEE9"
|
|
self.accent_color = "#5E81AC"
|
|
self.warning_color = "#BF616A"
|
|
self.success_color = "#A3BE8C"
|
|
|
|
# Main window configuration
|
|
self.root = root
|
|
self.root.title("ADALM1000 - Battery Capacity Tester (CC Test)")
|
|
self.root.geometry("1000x800")
|
|
self.root.minsize(800, 700)
|
|
self.root.configure(bg=self.bg_color)
|
|
|
|
# Device and measurement state
|
|
self.session_active = False
|
|
self.measuring = False
|
|
self.test_running = False
|
|
self.continuous_mode = False
|
|
self.request_stop = False
|
|
self.interval = 0.1
|
|
self.log_dir = os.path.expanduser("~/adalm1000/logs")
|
|
os.makedirs(self.log_dir, exist_ok=True)
|
|
|
|
# Battery test parameters
|
|
self.capacity = tk.DoubleVar(value=0.2) # Battery capacity in Ah
|
|
self.charge_cutoff = tk.DoubleVar(value=1.45) # Charge cutoff voltage
|
|
self.discharge_cutoff = tk.DoubleVar(value=0.9) # Discharge cutoff voltage
|
|
self.rest_time = tk.DoubleVar(value=0.1) # Rest time in hours
|
|
self.c_rate = tk.DoubleVar(value=0.1) # C-rate for test (default C/5 = 0.2)
|
|
|
|
# Test progress tracking
|
|
self.test_phase = tk.StringVar(value="Idle")
|
|
self.capacity_ah = tk.DoubleVar(value=0.0)
|
|
self.charge_capacity = tk.DoubleVar(value=0.0) # capacity tracking
|
|
self.coulomb_efficiency = tk.DoubleVar(value=0.0) # efficiency calculation
|
|
self.cycle_count = tk.IntVar(value=0) # cycle counting
|
|
|
|
# Data buffers
|
|
self.time_data = deque()
|
|
self.voltage_data = deque()
|
|
self.current_data = deque()
|
|
self.phase_data = deque()
|
|
|
|
# Initialize UI and device
|
|
self.setup_ui()
|
|
self.init_device()
|
|
|
|
# Ensure proper cleanup
|
|
self.root.protocol("WM_DELETE_WINDOW", self.on_close)
|
|
|
|
def setup_ui(self):
|
|
"""Configure the user interface"""
|
|
self.style = ttk.Style()
|
|
self.style.theme_use('clam')
|
|
|
|
# Configure styles
|
|
self.style.configure('.', background=self.bg_color, foreground=self.fg_color)
|
|
self.style.configure('TFrame', background=self.bg_color)
|
|
self.style.configure('TLabel', background=self.bg_color, foreground=self.fg_color)
|
|
self.style.configure('TButton', background=self.accent_color, foreground=self.fg_color,
|
|
padding=6, font=('Helvetica', 10, 'bold'))
|
|
self.style.map('TButton',
|
|
background=[('active', self.accent_color), ('disabled', '#4C566A')],
|
|
foreground=[('active', self.fg_color), ('disabled', '#D8DEE9')])
|
|
self.style.configure('TEntry', fieldbackground="#3B4252", foreground=self.fg_color)
|
|
self.style.configure('Header.TLabel', font=('Helvetica', 14, 'bold'), foreground=self.accent_color)
|
|
self.style.configure('Value.TLabel', font=('Helvetica', 12, 'bold'))
|
|
self.style.configure('Status.TLabel', font=('Helvetica', 10))
|
|
self.style.configure('Warning.TButton', background=self.warning_color)
|
|
self.style.configure('Success.TButton', background=self.success_color)
|
|
|
|
# Main layout
|
|
self.content_frame = ttk.Frame(self.root)
|
|
self.content_frame.pack(fill=tk.BOTH, expand=True, padx=10, pady=10)
|
|
|
|
# Header area
|
|
header_frame = ttk.Frame(self.content_frame)
|
|
header_frame.pack(fill=tk.X, pady=(0, 20))
|
|
|
|
ttk.Label(header_frame, text="ADALM1000 Battery Capacity Tester (CC Test)", style='Header.TLabel').pack(side=tk.LEFT)
|
|
|
|
# Status indicator
|
|
self.status_light = tk.Canvas(header_frame, width=20, height=20, bg=self.bg_color, bd=0, highlightthickness=0)
|
|
self.status_light.pack(side=tk.RIGHT, padx=10)
|
|
self.status_indicator = self.status_light.create_oval(2, 2, 18, 18, fill='red')
|
|
self.connection_label = ttk.Label(header_frame, text="Disconnected")
|
|
self.connection_label.pack(side=tk.RIGHT)
|
|
|
|
# Reconnect button
|
|
self.reconnect_btn = ttk.Button(header_frame, text="Reconnect", command=self.reconnect_device)
|
|
self.reconnect_btn.pack(side=tk.RIGHT, padx=10)
|
|
|
|
# Measurement display
|
|
display_frame = ttk.LabelFrame(self.content_frame, text=" Live Measurements ", padding=15)
|
|
display_frame.pack(fill=tk.BOTH, expand=False)
|
|
|
|
# Measurement values
|
|
measurement_labels = [
|
|
("Voltage (V)", "V"),
|
|
("Current (A)", "A"),
|
|
("Test Phase", ""),
|
|
("Elapsed Time", "s"),
|
|
("Discharge Capacity", "Ah"),
|
|
("Charge Capacity", "Ah"),
|
|
("Coulomb Eff.", "%"),
|
|
("Cycle Count", ""),
|
|
]
|
|
|
|
for i, (label, unit) in enumerate(measurement_labels):
|
|
ttk.Label(display_frame, text=f"{label}:", font=('Helvetica', 11)).grid(row=i//2, column=(i%2)*2, sticky=tk.W, pady=5)
|
|
value_label = ttk.Label(display_frame, text="0.000", style='Value.TLabel')
|
|
value_label.grid(row=i//2, column=(i%2)*2+1, sticky=tk.W, padx=10)
|
|
if unit:
|
|
ttk.Label(display_frame, text=unit).grid(row=i//2, column=(i%2)*2+2, sticky=tk.W)
|
|
|
|
if i == 0:
|
|
self.voltage_label = value_label
|
|
elif i == 1:
|
|
self.current_label = value_label
|
|
elif i == 2:
|
|
self.phase_label = value_label
|
|
elif i == 3:
|
|
self.time_label = value_label
|
|
elif i == 4:
|
|
self.capacity_label = value_label
|
|
elif i == 5:
|
|
self.charge_capacity_label = value_label
|
|
elif i == 6:
|
|
self.efficiency_label = value_label
|
|
elif i == 7:
|
|
self.cycle_label = value_label
|
|
|
|
# Control area
|
|
controls_frame = ttk.Frame(self.content_frame)
|
|
controls_frame.pack(fill=tk.X, pady=(10, 10), padx=0)
|
|
|
|
# Parameters frame
|
|
params_frame = ttk.LabelFrame(controls_frame, text="Test Parameters", padding=10)
|
|
params_frame.pack(side=tk.LEFT, fill=tk.X, expand=True)
|
|
|
|
# Battery capacity
|
|
ttk.Label(params_frame, text="Battery Capacity (Ah):").grid(row=0, column=0, sticky=tk.W)
|
|
ttk.Entry(params_frame, textvariable=self.capacity, width=6).grid(row=0, column=1, padx=5, sticky=tk.W)
|
|
|
|
# Charge cutoff
|
|
ttk.Label(params_frame, text="Charge Cutoff (V):").grid(row=1, column=0, sticky=tk.W)
|
|
ttk.Entry(params_frame, textvariable=self.charge_cutoff, width=6).grid(row=1, column=1, padx=5, sticky=tk.W)
|
|
|
|
# Discharge cutoff
|
|
ttk.Label(params_frame, text="Discharge Cutoff (V):").grid(row=2, column=0, sticky=tk.W)
|
|
ttk.Entry(params_frame, textvariable=self.discharge_cutoff, width=6).grid(row=2, column=1, padx=5, sticky=tk.W)
|
|
|
|
# Rest time
|
|
ttk.Label(params_frame, text="Rest Time (hours):").grid(row=3, column=0, sticky=tk.W)
|
|
ttk.Entry(params_frame, textvariable=self.rest_time, width=6).grid(row=3, column=1, padx=5, sticky=tk.W)
|
|
|
|
# C-rate for test (C/5 by default)
|
|
ttk.Label(params_frame, text="Test C-rate:").grid(row=0, column=2, sticky=tk.W, padx=(10,0))
|
|
ttk.Entry(params_frame, textvariable=self.c_rate, width=4).grid(row=0, column=3, padx=5, sticky=tk.W)
|
|
ttk.Label(params_frame, text="(e.g., 0.2 for C/5)").grid(row=0, column=4, sticky=tk.W)
|
|
|
|
# Start/Stop buttons
|
|
button_frame = ttk.Frame(controls_frame)
|
|
button_frame.pack(side=tk.RIGHT, padx=10)
|
|
|
|
self.start_button = ttk.Button(button_frame, text="START TEST", command=self.start_test, style='TButton')
|
|
self.start_button.pack(side=tk.TOP, pady=5)
|
|
|
|
self.stop_button = ttk.Button(button_frame, text="STOP TEST", command=self.stop_test, style='Warning.TButton', state=tk.DISABLED)
|
|
self.stop_button.pack(side=tk.TOP, pady=5)
|
|
|
|
# Continuous mode checkbox
|
|
self.continuous_var = tk.BooleanVar(value=True)
|
|
ttk.Checkbutton(button_frame, text="Continuous Mode", variable=self.continuous_var).pack(side=tk.TOP, pady=5)
|
|
|
|
# Plot area
|
|
self.plot_frame = ttk.Frame(self.content_frame)
|
|
self.plot_frame.pack(fill=tk.BOTH, expand=True, padx=20, pady=(0, 5))
|
|
self.setup_plot()
|
|
|
|
# Status bar
|
|
self.status_var = tk.StringVar()
|
|
self.status_var.set("Ready")
|
|
self.status_label = ttk.Label(self.root, textvariable=self.status_var, style='Status.TLabel', padding=(0, 5), anchor=tk.W)
|
|
self.status_label.place(x=20, relx=0, rely=1.0, anchor='sw', relwidth=0.96, height=28)
|
|
|
|
def setup_plot(self):
|
|
"""Configure the matplotlib plot"""
|
|
self.fig = Figure(figsize=(8, 5), dpi=100, facecolor='#2E3440')
|
|
self.fig.subplots_adjust(left=0.1, right=0.9, top=0.9, bottom=0.15)
|
|
self.ax = self.fig.add_subplot(111)
|
|
self.ax.set_facecolor('#3B4252')
|
|
|
|
# Voltage plot
|
|
self.line_voltage, = self.ax.plot([], [], color='#00BFFF', label='Voltage (V)', linewidth=2)
|
|
self.ax.set_ylabel("Voltage (V)", color='#00BFFF')
|
|
self.ax.tick_params(axis='y', labelcolor='#00BFFF')
|
|
|
|
# Current plot (right axis)
|
|
self.ax2 = self.ax.twinx()
|
|
self.line_current, = self.ax2.plot([], [], 'r-', label='Current (A)', linewidth=2)
|
|
self.ax2.set_ylabel("Current (A)", color='r')
|
|
self.ax2.tick_params(axis='y', labelcolor='r')
|
|
|
|
self.ax.set_xlabel('Time (s)', color=self.fg_color)
|
|
self.ax.set_title('Battery Test (CC)', color=self.fg_color)
|
|
self.ax.tick_params(axis='x', colors=self.fg_color)
|
|
self.ax.grid(True, color='#4C566A')
|
|
|
|
self.ax.legend(loc='upper left')
|
|
self.ax2.legend(loc='upper right')
|
|
|
|
# Embed plot
|
|
self.canvas = FigureCanvasTkAgg(self.fig, master=self.plot_frame)
|
|
self.canvas.draw()
|
|
canvas_widget = self.canvas.get_tk_widget()
|
|
canvas_widget.configure(bg='#2E3440', bd=0, highlightthickness=0)
|
|
canvas_widget.pack(side=tk.TOP, fill=tk.BOTH, expand=True, pady=(10, 0))
|
|
|
|
def init_device(self):
|
|
"""Initialize the ADALM1000 device with continuous measurement"""
|
|
try:
|
|
# First try to clean up any existing session
|
|
if hasattr(self, 'session'):
|
|
try:
|
|
self.session.end()
|
|
del self.session
|
|
except:
|
|
pass
|
|
|
|
# Add small delay to allow device to reset
|
|
time.sleep(1)
|
|
|
|
self.session = pysmu.Session(ignore_dataflow=True, queue_size=10000)
|
|
if not self.session.devices:
|
|
raise Exception("No ADALM1000 detected - check connections")
|
|
|
|
self.dev = self.session.devices[0]
|
|
# Reset channels
|
|
self.dev.channels['A'].mode = pysmu.Mode.HI_Z
|
|
self.dev.channels['B'].mode = pysmu.Mode.HI_Z
|
|
self.dev.channels['A'].constant(0)
|
|
self.dev.channels['B'].constant(0)
|
|
|
|
self.session.start(0)
|
|
|
|
self.status_light.itemconfig(self.status_indicator, fill='green')
|
|
self.connection_label.config(text="Connected")
|
|
self.status_var.set("Device connected | Ready to measure")
|
|
self.session_active = True
|
|
self.start_button.config(state=tk.NORMAL)
|
|
|
|
# Start continuous measurement thread
|
|
self.measurement_event = threading.Event()
|
|
self.measurement_event.set()
|
|
self.measurement_thread = threading.Thread(
|
|
target=self.continuous_measurement,
|
|
daemon=True
|
|
)
|
|
self.measurement_thread.start()
|
|
|
|
except Exception as e:
|
|
self.handle_device_error(e)
|
|
|
|
def continuous_measurement(self):
|
|
"""Continuous measurement with moving average filtering and optimized I/O"""
|
|
filter_window_size = 10
|
|
voltage_window = []
|
|
current_window = []
|
|
last_plot_update = 0
|
|
log_buffer = []
|
|
|
|
# Initialize start_time for measurements
|
|
if not hasattr(self, 'start_time'):
|
|
self.start_time = time.time()
|
|
|
|
while self.measurement_event.is_set() and self.root.winfo_exists():
|
|
try:
|
|
# Read multiple samples for better accuracy
|
|
samples = self.dev.read(filter_window_size, 500, True)
|
|
if not samples:
|
|
raise DeviceDisconnectedError("No samples received")
|
|
|
|
# Get voltage from Channel B (HI_Z mode) and current from Channel A
|
|
raw_voltage = np.mean([s[1][0] for s in samples]) # Channel B voltage
|
|
raw_current = np.mean([s[0][1] for s in samples]) # Channel A current
|
|
current_time = time.time() - self.start_time
|
|
|
|
# Apply moving average filter
|
|
voltage_window.append(raw_voltage)
|
|
current_window.append(raw_current)
|
|
|
|
if len(voltage_window) > filter_window_size:
|
|
voltage_window.pop(0)
|
|
current_window.pop(0)
|
|
|
|
voltage = np.mean(voltage_window)
|
|
current = np.mean(current_window)
|
|
|
|
# Store filtered data
|
|
self.time_data.append(current_time)
|
|
self.voltage_data.append(voltage)
|
|
self.current_data.append(current)
|
|
|
|
# Update UI with filtered values (throttled)
|
|
if current_time - last_plot_update > 0.5: # Update at 2Hz max
|
|
self.root.after(0, lambda: self.update_measurement_display(voltage, current, current_time))
|
|
last_plot_update = current_time
|
|
|
|
# Buffered logging
|
|
if self.test_running and hasattr(self, 'filename'):
|
|
log_buffer.append([
|
|
f"{current_time:.3f}",
|
|
f"{voltage:.6f}",
|
|
f"{current:.6f}",
|
|
self.test_phase.get(),
|
|
f"{self.capacity_ah.get():.4f}",
|
|
f"{self.charge_capacity.get():.4f}",
|
|
f"{self.coulomb_efficiency.get():.1f}",
|
|
f"{self.cycle_count.get()}"
|
|
])
|
|
|
|
# Write in chunks of 10 samples
|
|
if len(log_buffer) >= 10:
|
|
with open(self.filename, 'a', newline='') as f:
|
|
writer = csv.writer(f)
|
|
writer.writerows(log_buffer)
|
|
log_buffer.clear()
|
|
|
|
time.sleep(max(0.05, self.interval))
|
|
|
|
except Exception as e:
|
|
error_msg = str(e)
|
|
if self.root.winfo_exists():
|
|
self.root.after(0, lambda msg=error_msg:
|
|
self.handle_device_error(f"Measurement error: {msg}") if self.root.winfo_exists() else None)
|
|
break
|
|
|
|
# Flush remaining buffer on exit
|
|
if log_buffer and hasattr(self, 'filename'):
|
|
with open(self.filename, 'a', newline='') as f:
|
|
writer = csv.writer(f)
|
|
writer.writerows(log_buffer)
|
|
|
|
def start_test(self):
|
|
"""Start the full battery test cycle"""
|
|
if not self.test_running:
|
|
try:
|
|
# Validate inputs
|
|
if self.capacity.get() <= 0:
|
|
raise ValueError("Battery capacity must be positive")
|
|
if self.charge_cutoff.get() <= self.discharge_cutoff.get():
|
|
raise ValueError("Charge cutoff must be higher than discharge cutoff")
|
|
if self.c_rate.get() <= 0:
|
|
raise ValueError("C-rate must be positive")
|
|
|
|
# Set continuous mode based on checkbox
|
|
self.continuous_mode = self.continuous_var.get()
|
|
|
|
# Reset timing for new test
|
|
self.measurement_start_time = time.time()
|
|
self.test_start_time = time.time()
|
|
|
|
# Calculate target current
|
|
test_current = self.c_rate.get() * self.capacity.get()
|
|
|
|
if test_current > 0.2:
|
|
raise ValueError("Current must be ≤200mA (0.2A) for ADALM1000")
|
|
|
|
# Clear previous data
|
|
self.time_data.clear()
|
|
self.voltage_data.clear()
|
|
self.current_data.clear()
|
|
self.phase_data.clear()
|
|
self.capacity_ah.set(0.0)
|
|
self.charge_capacity.set(0.0)
|
|
self.coulomb_efficiency.set(0.0)
|
|
self.cycle_count.set(0)
|
|
|
|
# Setup new log file with buffered writer
|
|
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
self.filename = os.path.join(self.log_dir, f"battery_test_{timestamp}.csv")
|
|
self.log_file = open(self.filename, 'w', newline='')
|
|
self.log_writer = csv.writer(self.log_file)
|
|
self.log_writer.writerow(["Time(s)", "Voltage(V)", "Current(A)", "Phase", "Discharge_Capacity(Ah)", "Charge_Capacity(Ah)", "Coulomb_Eff(%)", "Cycle"])
|
|
self.log_buffer = []
|
|
|
|
# Start test thread
|
|
self.test_running = True
|
|
self.start_time = time.time()
|
|
self.last_update_time = time.time()
|
|
self.test_phase.set("Initial Discharge")
|
|
|
|
self.start_button.config(state=tk.DISABLED)
|
|
self.stop_button.config(state=tk.NORMAL)
|
|
self.status_var.set(f"Test started | Discharging to {self.discharge_cutoff.get()}V @ {test_current:.3f}A")
|
|
|
|
# Start test sequence in a new thread
|
|
self.test_thread = threading.Thread(target=self.run_test_sequence, daemon=True)
|
|
self.test_thread.start()
|
|
|
|
except Exception as e:
|
|
messagebox.showerror("Error", str(e))
|
|
|
|
@staticmethod
|
|
def format_time(seconds):
|
|
"""Convert seconds to hh:mm:ss format"""
|
|
hours = int(seconds // 3600)
|
|
minutes = int((seconds % 3600) // 60)
|
|
seconds = int(seconds % 60)
|
|
return f"{hours:02d}:{minutes:02d}:{seconds:02d}"
|
|
|
|
def stop_test(self):
|
|
"""Request immediate stop of the test"""
|
|
if not self.test_running:
|
|
return
|
|
|
|
self.request_stop = True
|
|
self.test_running = False # This will break out of all test loops
|
|
self.measuring = False
|
|
|
|
# Immediately set device to safe state
|
|
if hasattr(self, 'dev'):
|
|
self.dev.channels['A'].mode = pysmu.Mode.HI_Z
|
|
self.dev.channels['A'].constant(0)
|
|
|
|
self.status_var.set("Test stopped immediately")
|
|
self.stop_button.config(state=tk.DISABLED)
|
|
self.start_button.config(state=tk.NORMAL)
|
|
|
|
# Finalize test data
|
|
self.root.after(100, self.finalize_test)
|
|
|
|
def center_window(self, window):
|
|
"""Center a window on screen"""
|
|
window.update_idletasks()
|
|
width = window.winfo_width()
|
|
height = window.winfo_height()
|
|
x = (window.winfo_screenwidth() // 2) - (width // 2)
|
|
y = (window.winfo_screenheight() // 2) - (height // 2)
|
|
window.geometry(f'{width}x{height}+{x}+{y}')
|
|
|
|
def run_test_sequence(self):
|
|
try:
|
|
test_current = self.c_rate.get() * self.capacity.get()
|
|
|
|
while self.test_running and (self.continuous_mode or self.cycle_count.get() == 0):
|
|
# Reset stop request at start of each cycle
|
|
self.request_stop = False
|
|
|
|
# 1. Charge phase (constant current)
|
|
self.test_phase.set("Charge")
|
|
self.status_var.set(f"Charging to {self.charge_cutoff.get()}V @ {test_current:.3f}A")
|
|
self.root.update() # Force UI update
|
|
|
|
self.measuring = True
|
|
self.dev.channels['B'].mode = pysmu.Mode.HI_Z
|
|
self.dev.channels['A'].mode = pysmu.Mode.SIMV
|
|
self.dev.channels['A'].constant(test_current)
|
|
self.charge_capacity.set(0.0)
|
|
target_voltage = self.charge_cutoff.get()
|
|
self.last_update_time = time.time()
|
|
|
|
while self.test_running and not self.request_stop:
|
|
if not self.voltage_data:
|
|
time.sleep(0.1)
|
|
continue
|
|
|
|
current_voltage = self.voltage_data[-1]
|
|
measured_current = abs(self.current_data[-1])
|
|
|
|
# Update charge capacity
|
|
now = time.time()
|
|
delta_t = now - self.last_update_time
|
|
self.last_update_time = now
|
|
self.charge_capacity.set(self.charge_capacity.get() + measured_current * delta_t / 3600)
|
|
|
|
self.status_var.set(
|
|
f"Charging: {current_voltage:.3f}V / {target_voltage}V | "
|
|
f"Current: {measured_current:.3f}A | "
|
|
f"Capacity: {self.charge_capacity.get():.4f}Ah"
|
|
)
|
|
self.root.update() # Force UI update
|
|
|
|
if current_voltage >= target_voltage or self.request_stop:
|
|
break
|
|
|
|
time.sleep(0.1) # More frequent checks
|
|
|
|
if self.request_stop or not self.test_running:
|
|
break
|
|
|
|
# 2. Rest period after charge
|
|
self.test_phase.set("Resting (Post-Charge)")
|
|
self.measuring = False
|
|
self.dev.channels['A'].mode = pysmu.Mode.HI_Z
|
|
self.dev.channels['A'].constant(0)
|
|
|
|
rest_end_time = time.time() + (self.rest_time.get() * 3600)
|
|
while time.time() < rest_end_time and self.test_running and not self.request_stop:
|
|
time_left = max(0, rest_end_time - time.time())
|
|
self.status_var.set(
|
|
f"Resting after charge | "
|
|
f"Time left: {time_left/60:.1f} min"
|
|
)
|
|
self.root.update()
|
|
time.sleep(1) # Check every second for stop request
|
|
|
|
if self.request_stop or not self.test_running:
|
|
break
|
|
|
|
# 3. Discharge phase (capacity measurement)
|
|
self.test_phase.set("Discharge")
|
|
self.status_var.set(f"Discharging to {self.discharge_cutoff.get()}V @ {test_current:.3f}A")
|
|
self.root.update()
|
|
|
|
self.measuring = True
|
|
self.dev.channels['A'].mode = pysmu.Mode.SIMV
|
|
self.dev.channels['A'].constant(-test_current)
|
|
self.capacity_ah.set(0.0)
|
|
self.last_update_time = time.time()
|
|
|
|
while self.test_running and not self.request_stop:
|
|
if not self.current_data:
|
|
time.sleep(0.1)
|
|
continue
|
|
|
|
current_voltage = self.voltage_data[-1]
|
|
current_current = abs(self.current_data[-1])
|
|
|
|
# Capacity calculation
|
|
now = time.time()
|
|
delta_t = now - self.last_update_time
|
|
self.last_update_time = now
|
|
self.capacity_ah.set(self.capacity_ah.get() + current_current * delta_t / 3600)
|
|
|
|
self.status_var.set(
|
|
f"Discharging: {current_voltage:.3f}V / {self.discharge_cutoff.get()}V | "
|
|
f"Current: {current_current:.3f}A | "
|
|
f"Capacity: {self.capacity_ah.get():.4f}Ah"
|
|
)
|
|
self.root.update()
|
|
|
|
if current_voltage <= self.discharge_cutoff.get() or self.request_stop:
|
|
break
|
|
|
|
time.sleep(0.1) # More frequent checks
|
|
|
|
# 4. Rest period after discharge (only if not stopping)
|
|
if self.test_running and not self.request_stop:
|
|
self.test_phase.set("Resting (Post-Discharge)")
|
|
self.measuring = False
|
|
self.dev.channels['A'].mode = pysmu.Mode.HI_Z
|
|
self.dev.channels['A'].constant(0)
|
|
|
|
rest_end_time = time.time() + (self.rest_time.get() * 3600)
|
|
while time.time() < rest_end_time and self.test_running and not self.request_stop:
|
|
time_left = max(0, rest_end_time - time.time())
|
|
self.status_var.set(
|
|
f"Resting after discharge | "
|
|
f"Time left: {time_left/60:.1f} min"
|
|
)
|
|
self.root.update()
|
|
time.sleep(1)
|
|
|
|
# Calculate Coulomb efficiency if not stopping
|
|
if not self.request_stop and self.charge_capacity.get() > 0:
|
|
efficiency = (self.capacity_ah.get() / self.charge_capacity.get()) * 100
|
|
self.coulomb_efficiency.set(efficiency)
|
|
self.cycle_count.set(self.cycle_count.get() + 1)
|
|
|
|
# Update cycle info
|
|
self.status_var.set(
|
|
f"Cycle {self.cycle_count.get()} complete | "
|
|
f"Discharge: {self.capacity_ah.get():.3f}Ah | "
|
|
f"Charge: {self.charge_capacity.get():.3f}Ah | "
|
|
f"Efficiency: {self.coulomb_efficiency.get():.1f}%"
|
|
)
|
|
self.root.update()
|
|
|
|
# Write cycle summary to log file
|
|
self.write_cycle_summary()
|
|
|
|
# Short rest between cycles (only in continuous mode)
|
|
if self.continuous_mode and self.test_running and not self.request_stop:
|
|
rest_end_time = time.time() + (self.rest_time.get() * 3600)
|
|
while time.time() < rest_end_time and self.test_running and not self.request_stop:
|
|
time_left = max(0, rest_end_time - time.time())
|
|
self.test_phase.set("Resting Between Cycles")
|
|
self.status_var.set(
|
|
f"Resting between cycles | "
|
|
f"Time left: {time_left/60:.1f} min"
|
|
)
|
|
self.root.update()
|
|
time.sleep(1)
|
|
|
|
# Finalize test if stopped or completed
|
|
self.root.after(0, self.finalize_test)
|
|
|
|
except Exception as e:
|
|
error_msg = str(e)
|
|
if self.root.winfo_exists():
|
|
self.root.after(0, lambda msg=error_msg: messagebox.showerror("Test Error", msg))
|
|
self.root.after(0, self.finalize_test)
|
|
|
|
def finalize_test(self):
|
|
"""Final cleanup after test completes or is stopped"""
|
|
self.measuring = False
|
|
if hasattr(self, 'dev'):
|
|
self.dev.channels['A'].constant(0)
|
|
|
|
# Flush and close log file
|
|
if hasattr(self, 'log_buffer') and self.log_buffer and hasattr(self, 'log_writer'):
|
|
self.log_writer.writerows(self.log_buffer)
|
|
self.log_buffer.clear()
|
|
if hasattr(self, 'log_file'):
|
|
self.log_file.close()
|
|
|
|
if hasattr(self, 'filename'):
|
|
self.write_cycle_summary()
|
|
|
|
self.start_button.config(state=tk.NORMAL)
|
|
self.stop_button.config(state=tk.DISABLED)
|
|
self.request_stop = False
|
|
|
|
# Erfolgsmeldung mit mehr Details
|
|
message = (
|
|
f"Test safely stopped after discharge phase | "
|
|
f"Cycle {self.cycle_count.get()} completed | "
|
|
f"Final capacity: {self.capacity_ah.get():.3f}Ah"
|
|
)
|
|
self.status_var.set(message)
|
|
|
|
# Optional: Messagebox anzeigen
|
|
self.root.after(0, lambda: messagebox.showinfo(
|
|
"Test Completed",
|
|
f"Test was safely stopped after discharge phase.\n\n"
|
|
f"Final discharge capacity: {self.capacity_ah.get():.3f}Ah\n"
|
|
f"Total cycles completed: {self.cycle_count.get()}"
|
|
))
|
|
|
|
def update_measurement_display(self, voltage, current, current_time):
|
|
"""Update display with current measurements (optimized to only update changed values)"""
|
|
try:
|
|
# Only update changed values
|
|
voltage_text = f"{voltage:.4f}"
|
|
if not hasattr(self, '_last_voltage_text') or self._last_voltage_text != voltage_text:
|
|
self.voltage_label.config(text=voltage_text)
|
|
self._last_voltage_text = voltage_text
|
|
|
|
capacity_text = f"{self.capacity_ah.get():.4f}"
|
|
if not hasattr(self, '_last_capacity_text') or self._last_capacity_text != capacity_text:
|
|
self.capacity_label.config(text=capacity_text)
|
|
self._last_capacity_text = capacity_text
|
|
|
|
charge_capacity_text = f"{self.charge_capacity.get():.4f}"
|
|
if not hasattr(self, '_last_charge_capacity_text') or self._last_charge_capacity_text != charge_capacity_text:
|
|
self.charge_capacity_label.config(text=charge_capacity_text)
|
|
self._last_charge_capacity_text = charge_capacity_text
|
|
|
|
efficiency_text = f"{self.coulomb_efficiency.get():.1f}"
|
|
if not hasattr(self, '_last_efficiency_text') or self._last_efficiency_text != efficiency_text:
|
|
self.efficiency_label.config(text=efficiency_text)
|
|
self._last_efficiency_text = efficiency_text
|
|
|
|
cycle_text = f"{self.cycle_count.get()}"
|
|
if not hasattr(self, '_last_cycle_text') or self._last_cycle_text != cycle_text:
|
|
self.cycle_label.config(text=cycle_text)
|
|
self._last_cycle_text = cycle_text
|
|
|
|
current_text = f"{current:.4f}"
|
|
if not hasattr(self, '_last_current_text') or self._last_current_text != current_text:
|
|
self.current_label.config(text=current_text)
|
|
self._last_current_text = current_text
|
|
|
|
phase_text = self.test_phase.get()
|
|
if not hasattr(self, '_last_phase_text') or self._last_phase_text != phase_text:
|
|
self.phase_label.config(text=phase_text)
|
|
self._last_phase_text = phase_text
|
|
|
|
time_text = self.format_time(current_time)
|
|
if not hasattr(self, '_last_time_text') or self._last_time_text != time_text:
|
|
self.time_label.config(text=time_text)
|
|
self._last_time_text = time_text
|
|
|
|
# Update plot with proper scaling (throttled)
|
|
if not hasattr(self, '_last_plot_update') or (time.time() - self._last_plot_update > 1.0):
|
|
self.update_plot()
|
|
self._last_plot_update = time.time()
|
|
|
|
except Exception as e:
|
|
print(f"GUI update error: {e}")
|
|
|
|
def write_cycle_summary(self):
|
|
"""Write cycle summary to the log file"""
|
|
if not hasattr(self, 'filename'):
|
|
return
|
|
|
|
summary_line = (
|
|
f"Cycle {self.cycle_count.get()} - "
|
|
f"Discharge={self.capacity_ah.get():.4f}Ah, "
|
|
f"Charge={self.charge_capacity.get():.4f}Ah, "
|
|
f"Efficiency={self.coulomb_efficiency.get():.1f}%"
|
|
)
|
|
|
|
with open(self.filename, 'a', newline='') as f:
|
|
f.write(summary_line + "\n")
|
|
|
|
def update_plot(self):
|
|
"""Update plot with proper scaling and limits (optimized)"""
|
|
if not self.time_data:
|
|
return
|
|
|
|
# Only update if there's significant new data
|
|
if hasattr(self, '_last_plot_len') and len(self.time_data) - self._last_plot_len < 10:
|
|
return
|
|
|
|
self._last_plot_len = len(self.time_data)
|
|
|
|
# Update plot data
|
|
self.line_voltage.set_data(self.time_data, self.voltage_data)
|
|
self.line_current.set_data(self.time_data, self.current_data)
|
|
|
|
# Set x-axis to always show from 0 to current max time
|
|
min_time = 0 # Always start from 0
|
|
max_time = self.time_data[-1] + 1 # Add 1 second padding
|
|
|
|
# Only adjust limits if needed
|
|
current_xlim = self.ax.get_xlim()
|
|
if abs(current_xlim[1] - max_time) > 5: # Only adjust if significant change
|
|
self.ax.set_xlim(min_time, max_time)
|
|
self.ax2.set_xlim(min_time, max_time)
|
|
|
|
# Auto-scale y-axes with some margin (only if significant change)
|
|
if self.voltage_data:
|
|
voltage_margin = 0.2
|
|
min_voltage = max(0, min(self.voltage_data) - voltage_margin)
|
|
max_voltage = max(self.voltage_data) + voltage_margin
|
|
current_ylim = self.ax.get_ylim()
|
|
if abs(current_ylim[0] - min_voltage) > 0.1 or abs(current_ylim[1] - max_voltage) > 0.1:
|
|
self.ax.set_ylim(min_voltage, max_voltage)
|
|
|
|
if self.current_data:
|
|
current_margin = 0.05
|
|
min_current = min(self.current_data) - current_margin
|
|
max_current = max(self.current_data) + current_margin
|
|
current_ylim2 = self.ax2.get_ylim()
|
|
if abs(current_ylim2[0] - min_current) > 0.02 or abs(current_ylim2[1] - max_current) > 0.02:
|
|
self.ax2.set_ylim(min_current, max_current)
|
|
|
|
# Only redraw if needed
|
|
self.canvas.draw_idle()
|
|
|
|
def handle_device_error(self, error):
|
|
"""Handle device connection errors"""
|
|
if not self.root.winfo_exists(): # Check if window still exists
|
|
return
|
|
|
|
error_msg = str(error)
|
|
print(f"Device error: {error_msg}")
|
|
|
|
self.root.after_idle(lambda: self.status_light.itemconfig(self.status_indicator, fill='red'))
|
|
self.connection_label.config(text="Disconnected")
|
|
self.status_var.set(f"Device error: {error_msg}")
|
|
|
|
self.session_active = False
|
|
self.test_running = False
|
|
self.continuous_mode = False
|
|
self.measuring = False
|
|
if hasattr(self, 'start_button'):
|
|
self.start_button.config(state=tk.DISABLED)
|
|
if hasattr(self, 'stop_button'):
|
|
self.stop_button.config(state=tk.DISABLED)
|
|
|
|
# Clear plot + buffers
|
|
self.time_data.clear()
|
|
self.voltage_data.clear()
|
|
self.current_data.clear()
|
|
if hasattr(self, 'line_voltage') and hasattr(self, 'line_current'):
|
|
self.line_voltage.set_data([], [])
|
|
self.line_current.set_data([], [])
|
|
self.ax.set_xlim(0, 1)
|
|
self.ax2.set_xlim(0, 1)
|
|
self.canvas.draw()
|
|
|
|
# Clean up session
|
|
if hasattr(self, 'session'):
|
|
try:
|
|
if self.session_active:
|
|
self.session.end()
|
|
del self.session
|
|
except:
|
|
pass
|
|
|
|
if self.root.winfo_exists(): # Double-check before showing message
|
|
try:
|
|
messagebox.showerror(
|
|
"Device Connection Error",
|
|
f"Could not connect to ADALM1000:\n\n{error_msg}\n\n"
|
|
"1. Check USB cable connection\n"
|
|
"2. Try the Reconnect button\n"
|
|
"3. Restart the application if problem persists"
|
|
)
|
|
except:
|
|
pass # Ignore errors if window is being destroyed
|
|
|
|
def reconnect_device(self):
|
|
"""Reconnect the device"""
|
|
self.status_var.set("Attempting to reconnect...")
|
|
self.test_running = False
|
|
self.continuous_mode = False
|
|
self.measuring = False
|
|
if hasattr(self, 'measurement_event'):
|
|
self.measurement_event.clear()
|
|
|
|
# Wait for threads to finish
|
|
if hasattr(self, 'measurement_thread'):
|
|
self.measurement_thread.join(timeout=1.0)
|
|
if hasattr(self, 'test_thread'):
|
|
self.test_thread.join(timeout=1.0)
|
|
|
|
# Reset before reinitializing
|
|
self.handle_device_error("Reconnecting...")
|
|
self.init_device()
|
|
|
|
def on_close(self):
|
|
"""Clean up on window close"""
|
|
if hasattr(self, 'measurement_event'):
|
|
self.measurement_event.clear()
|
|
|
|
if hasattr(self, 'measurement_thread'):
|
|
self.measurement_thread.join(timeout=1.0)
|
|
if hasattr(self, 'test_thread'):
|
|
self.test_thread.join(timeout=1.0)
|
|
|
|
if hasattr(self, 'session') and self.session:
|
|
try:
|
|
if self.session_active:
|
|
self.session.end()
|
|
except:
|
|
pass
|
|
|
|
self.root.destroy()
|
|
|
|
if __name__ == "__main__":
|
|
root = tk.Tk()
|
|
try:
|
|
app = BatteryTester(root)
|
|
root.mainloop()
|
|
except Exception as e:
|
|
if root.winfo_exists():
|
|
messagebox.showerror("Fatal Error", f"Application failed: {str(e)}")
|
|
else:
|
|
print(f"Fatal Error: {e}")
|
|
try:
|
|
root.destroy()
|
|
except:
|
|
pass |