Jan/Adalm1000_Battery_SMU
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

MainCode/adalm1000_logger.py aktualisiert

Browse Source 
Safer:

    Application shutdown

    Thread cleanup

    Error recovery

    Reconnection scenarios
(Deepseek)
This commit is contained in:
Jan 2025-05-25 17:08:26 +02:00
parent 24cc224138
commit 25322bc59d
1 changed files with 65 additions and 30 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

93
MainCode/adalm1000_logger.py
View File

@ -19,6 +19,8 @@ class DeviceDisconnectedError(Exception):
class BatteryTester:
def __init__(self, root):
self._after_ids = set() # Track scheduled callbacks
self._after_lock = threading.Lock()
# Color scheme
self.bg_color = "#2E3440"
self.fg_color = "#D8DEE9"
@ -342,14 +344,14 @@ class BatteryTester:
# Throttle UI updates to prevent lag
now = time.time()
if now - last_ui_update > update_interval:
self.root.after(0, lambda: self.update_measurement_display(
self.safe_after(0, lambda: self.update_measurement_display(
voltage, current, current_time
))
last_ui_update = now
# Throttle plot updates even more (1Hz max)
if now - last_plot_update > 1.0:
self.root.after(0, self.update_plot)
self.safe_after(0, self.update_plot)
last_plot_update = now
# Buffered logging
@ -377,7 +379,7 @@ class BatteryTester:
except Exception as e:
error_msg = str(e)
if self.root.winfo_exists():
self.root.after(0, lambda msg=error_msg:
self.safe_after(0, lambda msg=error_msg:
self.handle_device_error(f"Measurement error: {msg}") if self.root.winfo_exists() else None)
break
@ -505,7 +507,7 @@ class BatteryTester:
self.start_button.config(state=tk.NORMAL)
# Finalize test data
self.root.after(100, self.finalize_test)
self.safe_after(100, self.finalize_test)
def center_window(self, window):
"""Center a window on screen"""
@ -674,13 +676,13 @@ class BatteryTester:
time.sleep(1)
# Finalize test if stopped or completed
self.root.after(0, self.finalize_test)
self.safe_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)
self.safe_after(0, lambda msg=error_msg: messagebox.showerror("Test Error", msg))
self.safe_after(0, self.finalize_test)
def finalize_test(self):
"""Final cleanup after test completes or is stopped"""
@ -716,7 +718,7 @@ class BatteryTester:
)
self.status_var.set(message)
self.root.after(0, lambda: messagebox.showinfo(
self.safe_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"
@ -763,7 +765,7 @@ class BatteryTester:
def update_plot(self):
"""Optimized plot update with change detection"""
if not self.time_data or len(self.time_data) < 10: # Wait for at least 10 samples
if not self.time_data or len(self.time_data) < 5: # Wait for at least 5 samples
return
# Only update if there's significant new data
@ -860,23 +862,43 @@ class BatteryTester:
# Show error message and attempt reconnect automatically
if self.root.winfo_exists():
self.root.after(100, self.attempt_reconnect)
self.safe_after(100, self.attempt_reconnect)
def safe_after(self, delay_ms, callback, *args):
"""Safely schedule a callback with window existence check"""
if not self.root.winfo_exists():
return None
def wrapped_callback():
if self.root.winfo_exists():
try:
callback(*args)
except Exception as e:
print(f"Callback error: {e}")
after_id = self.root.after(delay_ms, wrapped_callback)
self._after_ids.add(after_id)
return after_id
def attempt_reconnect(self):
"""Attempt to reconnect automatically"""
if not self.root.winfo_exists():
return
# Show error message first
messagebox.showerror(
"Device Connection Error",
"Could not connect to ADALM1000\n\n"
"1. Check USB cable connection\n"
"2. The device will attempt to reconnect automatically"
)
try:
# Show error message first
messagebox.showerror(
"Device Connection Error",
"Could not connect to ADALM1000\n\n"
"1. Check USB cable connection\n"
"2. The device will attempt to reconnect automatically"
)
except Exception as e:
print(f"Error showing message: {e}")
return
# Then attempt reconnect after a short delay
self.root.after(1000, self.reconnect_device)
# Schedule reconnect attempt
self.safe_after(1000, self.reconnect_device)
def reconnect_device(self):
"""Reconnect the device with proper cleanup"""
@ -923,33 +945,46 @@ class BatteryTester:
# If we get here, reconnection failed
self.status_var.set("Reconnect failed - will retry...")
self.root.after(2000, self.reconnect_device) # Retry after 2 seconds
self.safe_after(2000, self.reconnect_device) # Retry after 2 seconds
def on_close(self):
"""Clean up on window close"""
# Set flags to stop all threads
self.test_running = False
self.measuring = False
self.session_active = False
if hasattr(self, 'measurement_event'):
self.measurement_event.clear()
# Cancel all pending callbacks
with self._after_lock:
for after_id in self._after_ids:
try:
self.root.after_cancel(after_id)
except:
pass
self._after_ids.clear()
# Give threads time to clean up
timeout = 2.0 # seconds
timeout = 2.0
if hasattr(self, 'measurement_thread'):
self.measurement_thread.join(timeout=timeout)
if self.measurement_thread.is_alive():
print("Warning: Measurement thread did not terminate cleanly")
if hasattr(self, 'test_thread'):
self.test_thread.join(timeout=timeout)
if self.test_thread.is_alive():
print("Warning: Test thread did not terminate cleanly")
# Clean up device session
if hasattr(self, 'session') and self.session:
try:
if self.session_active:
self.session.end()
self.session.end()
except Exception as e:
print(f"Error ending session: {e}")
self.root.destroy()
# Finally destroy window
try:
self.root.destroy()
except:
pass
if __name__ == "__main__":
root = tk.Tk()