shudder5305/pupilometer
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Handled Ctrl C gracefully... I hope.

Browse Source
This commit is contained in:
Tempest 2025-08-08 15:52:38 +07:00
parent ef12927410
commit f3c2529394
1 changed files with 31 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

33
src/controllerSoftware/app.py
View File

@ -4,6 +4,9 @@ from bleak import BleakScanner, BleakClient
import threading
import time
import json
import sys
import signal
import os
# =================================================================================================
# APP CONFIGURATION
@ -125,7 +128,8 @@ async def connect_to_ble_device():
await ble_client.connect()
if ble_client.is_connected:
print(f"Connected to {target_device.name}")
services = ble_client.services
services = [service for service in ble_client.services if service.handle != 1]
# The previous logic for filtering services seems incorrect; let's grab all characteristics
characteristics = [
char for service in services for char in service.characteristics
@ -247,7 +251,32 @@ def set_matrix():
# APP STARTUP
# =================================================================================================
def signal_handler(signum, frame):
print("Received shutdown signal, gracefully shutting down...")
if not DEBUG_MODE and ble_client and ble_client.is_connected:
print("Disconnecting BLE client...")
disconnect_future = asyncio.run_coroutine_threadsafe(ble_client.disconnect(), ble_event_loop)
try:
# Wait for the disconnect to complete with a timeout
disconnect_future.result(timeout=5)
print("BLE client disconnected successfully.")
except Exception as e:
print(f"Error during BLE disconnect: {e}")
if not DEBUG_MODE and ble_event_loop and ble_event_loop.is_running():
print("Stopping BLE event loop...")
# Schedule a stop and wait for the thread to finish
ble_event_loop.call_soon_threadsafe(ble_event_loop.stop)
ble_thread.join(timeout=1)
print("BLE event loop stopped.")
os._exit(0)
if __name__ == '__main__':
# Register the signal handler before running the app
signal.signal(signal.SIGINT, signal_handler)
signal.signal(signal.SIGTERM, signal_handler)
if not DEBUG_MODE:
print("Starting BLE event loop in background thread...")
ble_event_loop = asyncio.new_event_loop()
@ -258,4 +287,4 @@ if __name__ == '__main__':
future = asyncio.run_coroutine_threadsafe(connect_to_ble_device(), ble_event_loop)
future.result(timeout=10) # Wait up to 10 seconds for connection
app.run(debug=True, use_reloader=False)
app.run(debug=True, use_reloader=False, host="0.0.0.0")