Exergie/DataFlux
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Codex crash hardening

Browse Source
This commit is contained in:
Hector van der Aa
2026-06-07 13:30:45 +02:00
parent 00b1e67f89
commit 0330ca94ae
10 changed files with 372 additions and 162 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
src/dataflux/app.py
View File

@@ -6,6 +6,7 @@ from datetime import datetime
from pathlib import Path
import sys
from threading import Thread
import traceback
import dearpygui.dearpygui as dpg
from dataflux.state import AppState
@@ -48,13 +49,12 @@ def run() -> None:
user_agent="DataFlux/0.1 contact:h3cx@h3cx.dev", cache_dir="./.cache"
)
# Create application context and viewport
dpg.create_context()
try:
dpg.configure_app(manual_callback_management=True)
dpg.create_viewport(title="DataFlux", width=600, height=600)
# Add Inter font to registry and bind as main app font
with dpg.font_registry():
app_font = dpg.add_font(_asset_path("assets/fonts/Inter-Regular.ttf"), 18 * 2)
mono_font = dpg.add_font(
@@ -77,7 +77,9 @@ def run() -> None:
state.telemetry_thread_running = True
state.telemetry_thread = Thread(
target=dataflux.services.telemetry.telemetry_worker, args=(state,), daemon=True
target=dataflux.services.telemetry.telemetry_worker,
args=(state,),
daemon=True,
)
state.telemetry_thread.start()
@@ -90,14 +92,21 @@ def run() -> None:
ui_updater = dataflux.ui.worker.UiFrameUpdater()
while dpg.is_dearpygui_running():
jobs = dpg.get_callback_queue()
try:
dpg.run_callbacks(jobs)
ui_updater.update(state)
except Exception:
traceback.print_exc()
dpg.render_dearpygui_frame()
finally:
state.running = False
state.telemetry_thread_running = False
state.ports_thread_running = False
state.lora_thread_running = False
state.serial_thread_running = False
try:
dataflux.services.lora.disconnect_lora(state)
dataflux.services.serial_console.disconnect_serial(state)
finally:
dpg.destroy_context()

40
src/dataflux/callbacks/menu.py
View File

@@ -23,6 +23,17 @@ from dataflux.tags import (
)
def _file_dialog_path(app_data) -> str | None:
if not isinstance(app_data, dict):
return None
path = app_data.get("file_path_name")
if not path:
return None
return str(path)
def open_lora_connection_window(sender, app_data, user_data: AppState) -> None:
dataflux.ui.routines.windows.update_window_lora_connection_menu_combo(user_data)
dpg.show_item(WINDOW_LORA_CONNECTION_MENU)
@@ -56,9 +67,18 @@ def menu_file_quit(sender, app_data, user_data) -> None:
def window_file_dialog_dump_buffers_ok(sender, app_data, user_data: AppState) -> None:
path = _file_dialog_path(app_data)
if path is None:
print("No dump path selected")
return
if user_data.buffer_dump_thread is not None and user_data.buffer_dump_thread.is_alive():
print("Buffer dump is already running")
return
user_data.buffer_dump_thread = Thread(
target=dataflux.services.telemetry.buffer_dump,
args=(user_data, app_data["file_path_name"]),
args=(user_data, path),
daemon=True,
)
user_data.buffer_dump_thread.start()
@@ -75,19 +95,33 @@ def menu_file_autosave_buffers(sender, app_state, user_data: AppState) -> None:
def window_file_dialog_autosave_buffers_ok(
sender, app_data, user_data: AppState
) -> None:
path = _file_dialog_path(app_data)
if path is None:
print("No autosave folder selected")
return
user_data.autosave_enabled = True
user_data.autosave_buffer_thread = Thread(
target=dataflux.services.telemetry.autosave_worker,
args=(user_data, app_data["file_path_name"]),
args=(user_data, path),
daemon=True,
)
user_data.autosave_buffer_thread.start()
def window_file_dialog_load_lap_ok(sender, app_data, user_data: AppState) -> None:
path = _file_dialog_path(app_data)
if path is None:
print("No lap file selected")
return
if user_data.lap_loader_thread is not None and user_data.lap_loader_thread.is_alive():
print("Lap file loader is already running")
return
user_data.lap_loader_thread = Thread(
target=dataflux.services.telemetry.lap_load_worker,
args=(user_data, app_data["file_path_name"]),
args=(user_data, path),
daemon=True,
)
user_data.lap_loader_thread.start()

11
src/dataflux/callbacks/serial.py
View File

@@ -36,5 +36,16 @@ def connection_window_connect_serial(sender, app_data, user_data: AppState) -> N
def serial_console_button_send(sender, app_data, user_data: AppState) -> None:
text = dpg.get_value(INPUT_SERIAL_CONSOLE)
dpg.set_value(INPUT_SERIAL_CONSOLE, "")
if text is None:
return
text = str(text)
if not text:
return
if user_data.serial_port is None or not user_data.serial_thread_running:
print("Serial console is not connected")
return
user_data.serial_send_queue.put(text)
print("Put into send queue: " + text)

55
src/dataflux/services/lora.py
View File

@@ -3,6 +3,7 @@
# SPDX-License-Identifier: GPL-3.0-or-later
from threading import Thread
import struct
from serial import Serial
import serial
@@ -11,19 +12,29 @@ import dataflux.telemetry_common.telemetry_common
from dataflux.state import AppState
def connect_lora(state: AppState, device: str) -> bool:
def _close_lora_port(state: AppState) -> None:
if state.lora_port is None:
return
try:
state.lora_port.close()
except (OSError, serial.SerialException) as exc:
print(f"Could not close LoRa port: {exc}")
finally:
state.lora_port = None
def connect_lora(state: AppState, device: str | None) -> bool:
if not device:
print("No LoRa port selected")
state.connection_status_dirty = True
return False
if state.lora_port is not None:
state.lora_port.close()
state.lora_port = None
_close_lora_port(state)
try:
state.lora_port = Serial(port=device, baudrate=115200)
except serial.SerialException as exc:
state.lora_port = Serial(port=device, baudrate=115200, timeout=0.1)
except (OSError, serial.SerialException, ValueError) as exc:
print(f"Could not open LoRa port {device!r}: {exc}")
state.lora_port = None
state.lora_thread_running = False
@@ -43,11 +54,7 @@ def disconnect_lora(state: AppState) -> None:
return
state.lora_thread_running = False
try:
state.lora_port.close()
except OSError:
pass
state.lora_port = None
_close_lora_port(state)
state.connection_status_dirty = True
@@ -70,13 +77,20 @@ def lora_reader_worker(state: AppState) -> None:
state.packet_queue.put(parsed)
state.lora_status_queue.put(0.1)
except Exception:
except (OSError, serial.SerialException) as exc:
print(f"LoRa read failed: {exc}")
break
except Exception as exc:
print(f"LoRa worker ignored malformed packet: {exc}")
break
disconnect_lora(state)
def read_one_uart_packet(port: Serial) -> bytes | None:
if port is None or port.closed:
return None
first = port.read(1)
if not first:
return None
@@ -109,9 +123,14 @@ def parse_uart_packet(body: bytes) -> dict | None:
if len(body) < telemetry_common.LORA_HEADER_SIZE:
return None
try:
lora = telemetry_common.unpack_lora_header(
body[: telemetry_common.LORA_HEADER_SIZE]
)
except (ValueError, TypeError, IndexError, struct.error) as exc:
print(f"Invalid LoRa header: {exc}")
return None
payload = body[telemetry_common.LORA_HEADER_SIZE :]
if lora.size != len(payload):
@@ -133,7 +152,11 @@ def parse_uart_packet(body: bytes) -> dict | None:
}
if lora.version == 1:
try:
pkt = telemetry_common.unpack_packet1(payload)
except (ValueError, TypeError, IndexError, struct.error) as exc:
print(f"Invalid packet1 payload: {exc}")
return None
return {
**base,
"type": "packet1",
@@ -141,7 +164,11 @@ def parse_uart_packet(body: bytes) -> dict | None:
}
if lora.version == 2:
try:
pkt = telemetry_common.unpack_packet2(payload)
except (ValueError, TypeError, IndexError, struct.error) as exc:
print(f"Invalid packet2 payload: {exc}")
return None
return {
**base,
"type": "packet2",
@@ -154,7 +181,11 @@ def parse_uart_packet(body: bytes) -> dict | None:
}
if lora.version == 3:
try:
pkt = telemetry_common.unpack_packet3(payload)
except (ValueError, TypeError, IndexError, struct.error) as exc:
print(f"Invalid packet3 payload: {exc}")
return None
return {
**base,
"type": "packet3",

6
src/dataflux/services/ports.py
View File

@@ -13,6 +13,8 @@ def list_serial_ports(state: AppState) -> list[str]:
valid_ports: list[str] = []
for port in state.ports:
if port is None:
continue
if port.vid is not None and port.pid is not None:
valid_ports.append(port.device)
@@ -21,5 +23,9 @@ def list_serial_ports(state: AppState) -> list[str]:
def ports_worker(state: AppState) -> None:
while state.ports_thread_running:
try:
state.ports = serial.tools.list_ports.comports()
except (OSError, serial.SerialException) as exc:
print(f"Could not list serial ports: {exc}")
state.ports = []
time.sleep(5)

62
src/dataflux/services/serial_console.py
View File

@@ -12,21 +12,31 @@ import serial
from dataflux.state import AppState
def connect_serial(state: AppState, device: str) -> bool:
def _close_serial_port(state: AppState) -> None:
if state.serial_port is None:
return
try:
state.serial_port.close()
except (OSError, serial.SerialException) as exc:
print(f"Could not close serial console port: {exc}")
finally:
state.serial_port = None
def connect_serial(state: AppState, device: str | None) -> bool:
if not device:
print("No serial console port selected")
state.connection_status_dirty = True
return False
if state.serial_port is not None:
state.serial_port.close()
state.serial_port = None
_close_serial_port(state)
try:
state.serial_port = Serial(
port=device, baudrate=115200, timeout=0.05, write_timeout=0.1
)
except serial.SerialException as exc:
except (OSError, serial.SerialException, ValueError) as exc:
print(f"Could not open serial console port {device!r}: {exc}")
state.serial_port = None
state.serial_thread_running = False
@@ -46,11 +56,7 @@ def disconnect_serial(state: AppState) -> None:
return
state.serial_thread_running = False
try:
state.serial_port.close()
except OSError:
pass
state.serial_port = None
_close_serial_port(state)
state.connection_status_dirty = True
@@ -67,11 +73,8 @@ def serial_worker(state: AppState) -> None:
try:
line = port.readline()
except TypeError:
break
except serial.SerialException:
break
except OSError:
except (TypeError, OSError, serial.SerialException) as exc:
print(f"Serial console read failed: {exc}")
break
if line:
@@ -79,14 +82,31 @@ def serial_worker(state: AppState) -> None:
state.serial_data_queue.put(text)
state.serial_status_queue.put(0.05)
if port.writable():
try:
data: str = state.serial_send_queue.get_nowait()
writable = port.writable()
except (OSError, serial.SerialException) as exc:
print(f"Serial console write check failed: {exc}")
break
if not writable:
continue
try:
data = state.serial_send_queue.get_nowait()
except Empty:
pass
else:
state.serial_data_queue.put(data + "\n")
continue
if data is None:
continue
text_to_send = str(data)
try:
port.write(text_to_send.encode("utf-8"))
except (OSError, serial.SerialException, TypeError) as exc:
print(f"Serial console write failed: {exc}")
break
state.serial_data_queue.put(text_to_send + "\n")
state.serial_status_queue.put(0.05)
port.write(data.encode("utf-8"))
disconnect_serial(state)

104
src/dataflux/services/telemetry/__init__.py
View File

@@ -5,8 +5,6 @@
from bisect import bisect_left
from datetime import datetime, timezone
from queue import Empty
import stat
from tracemalloc import start
from dataflux.state import AppState, Buffers, LapInfo
import time
from pathlib import Path
@@ -52,22 +50,47 @@ def telemetry_worker(state: AppState):
except Empty:
continue
if not isinstance(dataframe, dict):
print(f"Ignoring telemetry packet with unexpected type: {type(dataframe)}")
continue
packet_type = dataframe.get("type")
now = datetime.now(timezone.utc)
time_stamp = datetime_to_cs(now)
if (
dataframe["type"] == "packet2"
and abs(dataframe["time_stamp"] - time_stamp) <= 60 * 100
):
state.latest_telemetry = dataframe
if packet_type == "packet2":
try:
packet_time = int(dataframe["time_stamp"])
speed = float(dataframe["speed"])
vbat = float(dataframe["vbat"])
teng = float(dataframe["teng"])
lat = float(dataframe["lat"])
lng = float(dataframe["lng"])
except (KeyError, TypeError, ValueError) as exc:
print(f"Ignoring malformed telemetry packet2: {exc}")
continue
if abs(packet_time - time_stamp) > 60 * 100:
continue
state.latest_telemetry = {
**dataframe,
"time_stamp": packet_time,
"speed": speed,
"vbat": vbat,
"teng": teng,
"lat": lat,
"lng": lng,
}
state.telemetry_valid = True
with state.lock:
state.raw_buffers.timestamp.append(dataframe["time_stamp"])
state.raw_buffers.speed.append(dataframe["speed"])
state.raw_buffers.vbat.append(dataframe["vbat"])
state.raw_buffers.teng.append(dataframe["teng"])
state.raw_buffers.lat.append(dataframe["lat"])
state.raw_buffers.lng.append(dataframe["lng"])
state.raw_buffers.timestamp.append(packet_time)
state.raw_buffers.speed.append(speed)
state.raw_buffers.vbat.append(vbat)
state.raw_buffers.teng.append(teng)
state.raw_buffers.lat.append(lat)
state.raw_buffers.lng.append(lng)
state.live_buffers_updated = True
state.live_buffers.timestamp.clear()
@@ -103,21 +126,37 @@ def telemetry_worker(state: AppState):
state.live_buffers.teng.reverse()
state.live_buffers.lat.reverse()
state.live_buffers.lng.reverse()
elif dataframe["type"] == "packet3":
start_time: int = dataframe["start_time"]
end_time: int = dataframe["duration"] + start_time
lap_count = dataframe["count"]
elif packet_type == "packet3":
try:
start_time = int(dataframe["start_time"])
end_time = int(dataframe["duration"]) + start_time
lap_count = int(dataframe["count"])
except (KeyError, TypeError, ValueError) as exc:
print(f"Ignoring malformed telemetry packet3: {exc}")
continue
lap: LapInfo = LapInfo(start_time, end_time, lap_count)
state.laps.append(lap)
state.new_laps.put(lap)
def save_lap(state: AppState, start_time: int, end_time: int, count: int) -> None:
if state.autosave_path is None:
print("Cannot save lap without an autosave path")
return
try:
time_str = cs_to_datetime(datetime.now(timezone.utc), start_time).strftime(
"%m_%d_%Y_%H_%M"
)
except ValueError:
time_str = datetime.now(timezone.utc).strftime("%m_%d_%Y_%H_%M")
save_path = Path(state.autosave_path) / f"{time_str}_lap_{count}.csv"
data: Buffers = isolate_lap(state, start_time, end_time)
try:
save_path.parent.mkdir(parents=True, exist_ok=True)
with save_path.open("w", newline="", encoding="utf-8") as f:
writer = csv.writer(f)
@@ -127,6 +166,8 @@ def save_lap(state: AppState, start_time: int, end_time: int, count: int) -> Non
data.timestamp, data.speed, data.vbat, data.teng, data.lat, data.lng
):
writer.writerow(row)
except OSError as exc:
print(f"Could not save lap to {save_path}: {exc}")
def buffer_dump(state: AppState, path: str) -> None:
@@ -144,6 +185,8 @@ def buffer_dump(state: AppState, path: str) -> None:
lng=list(state.raw_buffers.lng),
)
try:
save_path.parent.mkdir(parents=True, exist_ok=True)
with save_path.open("w", newline="", encoding="utf-8") as f:
writer = csv.writer(f)
@@ -158,6 +201,8 @@ def buffer_dump(state: AppState, path: str) -> None:
local_raw_buffers.lng,
):
writer.writerow(row)
except OSError as exc:
print(f"Could not dump buffers to {save_path}: {exc}")
state.buffer_dump_thread = None
@@ -165,6 +210,14 @@ def buffer_dump(state: AppState, path: str) -> None:
def autosave_worker(state: AppState, path: str) -> None:
output_dir = Path(path)
state.autosave_path = output_dir
try:
output_dir.mkdir(parents=True, exist_ok=True)
except OSError as exc:
print(f"Could not create autosave directory {output_dir}: {exc}")
state.autosave_enabled = False
state.autosave_buffer_thread = None
return
ctr: int = 0
while state.autosave_enabled:
date_str = state.start_time.strftime("%m_%d_%Y_%H_%M")
@@ -183,6 +236,8 @@ def autosave_worker(state: AppState, path: str) -> None:
time.sleep(30)
state.autosave_buffer_thread = None
def lap_load_worker(state: AppState, path: str) -> None:
state.lap_recap_buffers.timestamp.clear()
@@ -207,11 +262,11 @@ def lap_load_worker(state: AppState, path: str) -> None:
state.lap_recap_buffers.lng.append(float(row["lng"]))
except FileNotFoundError:
pass
except KeyError:
pass
except ValueError:
pass
print(f"Lap file not found: {load_path}")
except (KeyError, ValueError) as exc:
print(f"Invalid lap file {load_path}: {exc}")
except OSError as exc:
print(f"Could not load lap file {load_path}: {exc}")
else:
state.lap_recap_updated = True
@@ -241,8 +296,13 @@ def closest_idx(values: list[int], target: int) -> int:
def isolate_lap(state: AppState, start_time: int, end_time: int) -> Buffers:
output: Buffers = Buffers()
if not state.raw_buffers.timestamp:
return output
start_idx = closest_idx(state.raw_buffers.timestamp, start_time)
end_idx = closest_idx(state.raw_buffers.timestamp, end_time)
if start_idx > end_idx:
start_idx, end_idx = end_idx, start_idx
output.timestamp = state.raw_buffers.timestamp[start_idx : end_idx + 1]
output.speed = state.raw_buffers.speed[start_idx : end_idx + 1]

2
src/dataflux/ui/routines/serial.py
View File

@@ -9,6 +9,8 @@ from dataflux.tags import CHILD_WINDOW_SERIAL_CONSOLE, TEXT_SERIAL_CONSOLE
def append_text_to_console(text: str) -> None:
old = dpg.get_value(TEXT_SERIAL_CONSOLE)
if old is None:
old = ""
dpg.set_value(TEXT_SERIAL_CONSOLE, old + text)
def scroll_to_bottom() -> None:

1
src/dataflux/ui/windows.py
View File

@@ -2,7 +2,6 @@
# Copyright (C) 2026 Association Exergie <association.exergie@gmail.com>
# SPDX-License-Identifier: GPL-3.0-or-later
from ast import arg
import dearpygui.dearpygui as dpg
import dpg_map as dpgm
from dataflux.callbacks.map import (

76
src/dataflux/ui/worker.py
View File

@@ -50,14 +50,20 @@ class UiFrameUpdater:
self.last_map_update = 0.0
def update(self, state: AppState) -> None:
self._update_connection_status(state)
self._update_status_flashes(state)
self._update_autosave_menu(state)
self._update_utc_time()
self._drain_serial_console(state)
self._update_lap_recap(state)
self._update_live_telemetry(state)
self._update_live_map(state)
for updater in (
self._update_connection_status,
self._update_status_flashes,
self._update_autosave_menu,
self._update_utc_time,
self._drain_serial_console,
self._update_lap_recap,
self._update_live_telemetry,
self._update_live_map,
):
try:
updater(state)
except Exception as exc:
print(f"UI update failed in {updater.__name__}: {exc}")
def _update_connection_status(self, state: AppState) -> None:
if not state.connection_status_dirty:
@@ -98,19 +104,25 @@ class UiFrameUpdater:
def _drain_flash_queue(
self, queue, tag: str, flash_until: float, now: float
) -> float:
if queue is None:
return flash_until
while True:
try:
duration = queue.get_nowait()
except Empty:
return flash_until
flash_until = max(flash_until, now + duration)
try:
flash_until = max(flash_until, now + float(duration))
except (TypeError, ValueError):
continue
dpg.bind_item_theme(tag, THEME_STATUS_CONNECTED_BRIGHT)
def _update_autosave_menu(self, state: AppState) -> None:
dpg.set_value(MENU_FILE_AUTOSAVE_BUFFERS, state.autosave_enabled)
def _update_utc_time(self) -> None:
def _update_utc_time(self, state: AppState | None = None) -> None:
formatted = datetime.now(timezone.utc).strftime("%H:%M:%S")
if formatted != self.last_datetime:
@@ -118,13 +130,17 @@ class UiFrameUpdater:
self.last_datetime = formatted
def _drain_serial_console(self, state: AppState) -> None:
if state.serial_data_queue is None:
return
while True:
try:
text = state.serial_data_queue.get_nowait()
except Empty:
break
append_text_to_console(text)
if text is not None:
append_text_to_console(str(text))
def _update_lap_recap(self, state: AppState) -> None:
if not state.lap_recap_updated:
@@ -164,10 +180,16 @@ class UiFrameUpdater:
with state.lock:
self.no_data_written = False
veh_time = state.latest_telemetry["time_stamp"]
veh_speed = state.latest_telemetry["speed"]
vbat = state.latest_telemetry["vbat"]
teng = state.latest_telemetry["teng"]
try:
veh_time = int(state.latest_telemetry["time_stamp"])
veh_speed = float(state.latest_telemetry["speed"])
vbat = float(state.latest_telemetry["vbat"])
teng = float(state.latest_telemetry["teng"])
except (KeyError, TypeError, ValueError):
state.telemetry_valid = False
self._write_no_data()
return
if state.live_buffers_updated:
x_common = list(state.live_buffers.timestamp)
speed_y = list(state.live_buffers.speed)
@@ -222,24 +244,33 @@ class UiFrameUpdater:
with state.lock:
try:
lat = float(state.latest_telemetry.get("lat", DEFAULT_LAT))
except TypeError, ValueError:
except (TypeError, ValueError):
lat = DEFAULT_LAT
try:
lon = float(state.latest_telemetry.get("lng", DEFAULT_LNG))
except TypeError, ValueError:
except (TypeError, ValueError):
lon = DEFAULT_LNG
try:
dpgm.set_center(lat=lat, lon=lon, map_tag="live_map")
except Exception as exc:
print(f"Could not recenter map: {exc}")
state.live_map_pending_recenter = False
if state.live_map_pending_zoom_in:
try:
zl: int = dpgm.get_zoom(map_tag="live_map")
dpgm.set_zoom(zoom=zl + 1, map_tag="live_map")
except Exception as exc:
print(f"Could not zoom map in: {exc}")
state.live_map_pending_zoom_in = False
if state.live_map_pending_zoom_out:
try:
zl: int = dpgm.get_zoom(map_tag="live_map")
dpgm.set_zoom(zoom=zl - 1, map_tag="live_map")
except Exception as exc:
print(f"Could not zoom map out: {exc}")
state.live_map_pending_zoom_out = False
if now - self.last_map_update < 0.1:
@@ -249,8 +280,15 @@ class UiFrameUpdater:
return
with state.lock:
lat: float = state.latest_telemetry["lat"]
lon: float = state.latest_telemetry["lng"]
try:
lat = float(state.latest_telemetry["lat"])
lon = float(state.latest_telemetry["lng"])
except (KeyError, TypeError, ValueError):
return
try:
dpgm.update_marker("vehicle", lat=lat, lon=lon, map_tag="live_map")
except Exception as exc:
print(f"Could not update map marker: {exc}")
return
self.last_map_update = now