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

Added buffer autosave and fixed slow comports() call by delocalizing to thread

Browse Source
This commit is contained in:
Hector van der Aa
2026-05-18 21:47:02 +02:00
parent e5fd5cda89
commit a69b45ed27
11 changed files with 165 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -2,6 +2,7 @@
# Copyright (C) 2026 Association Exergie <association.exergie@gmail.com>
# SPDX-License-Identifier: GPL-3.0-or-later
from datetime import datetime
from queue import Empty
from dataflux.state import AppState, Buffers
import time
@@ -19,54 +20,59 @@ def telemetry_worker(state: AppState):
except Empty:
continue
state.latest_telemetry = dataframe
state.telemetry_valid = True
if dataframe["type"] == "packet2":
state.latest_telemetry = dataframe
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"])
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.live_buffers_updated = True
state.live_buffers.timestamp.clear()
state.live_buffers.speed.clear()
state.live_buffers.vbat.clear()
state.live_buffers.teng.clear()
state.live_buffers.lat.clear()
state.live_buffers.lng.clear()
state.live_buffers_updated = True
state.live_buffers.timestamp.clear()
state.live_buffers.speed.clear()
state.live_buffers.vbat.clear()
state.live_buffers.teng.clear()
state.live_buffers.lat.clear()
state.live_buffers.lng.clear()
if not state.raw_buffers.timestamp:
return
if not state.raw_buffers.timestamp:
return
last_timestamp = state.raw_buffers.timestamp[-1]
cutoff = last_timestamp - (state.live_buffer_len * 100)
last_timestamp = state.raw_buffers.timestamp[-1]
cutoff = last_timestamp - (state.live_buffer_len * 100)
i = len(state.raw_buffers.timestamp) - 1
i = len(state.raw_buffers.timestamp) - 1
while i >= 0 and state.raw_buffers.timestamp[i] >= cutoff:
elapsed_seconds = (
state.raw_buffers.timestamp[i] - last_timestamp
) / 100.0
state.live_buffers.timestamp.append(elapsed_seconds)
state.live_buffers.speed.append(state.raw_buffers.speed[i])
state.live_buffers.vbat.append(state.raw_buffers.vbat[i])
state.live_buffers.teng.append(state.raw_buffers.teng[i])
state.live_buffers.lat.append(state.raw_buffers.lat[i])
state.live_buffers.lng.append(state.raw_buffers.lng[i])
i -= 1
while i >= 0 and state.raw_buffers.timestamp[i] >= cutoff:
elapsed_seconds = (
state.raw_buffers.timestamp[i] - last_timestamp
) / 100.0
state.live_buffers.timestamp.append(elapsed_seconds)
state.live_buffers.speed.append(state.raw_buffers.speed[i])
state.live_buffers.vbat.append(state.raw_buffers.vbat[i])
state.live_buffers.teng.append(state.raw_buffers.teng[i])
state.live_buffers.lat.append(state.raw_buffers.lat[i])
state.live_buffers.lng.append(state.raw_buffers.lng[i])
i -= 1
state.live_buffers.timestamp.reverse()
state.live_buffers.speed.reverse()
state.live_buffers.vbat.reverse()
state.live_buffers.teng.reverse()
state.live_buffers.lat.reverse()
state.live_buffers.lng.reverse()
state.live_buffers.timestamp.reverse()
state.live_buffers.speed.reverse()
state.live_buffers.vbat.reverse()
state.live_buffers.teng.reverse()
state.live_buffers.lat.reverse()
state.live_buffers.lng.reverse()
elif dataframe["type"] == "packet3":
print(dataframe["type"])
print(dataframe["start_time"])
print(dataframe["duration"])
print(dataframe["count"])
def buffer_dump(state: AppState, path: str):
def buffer_dump(state: AppState, path: str) -> None:
save_path = Path(path)
if save_path.is_dir():
save_path = save_path / "output.csv"
@@ -97,3 +103,16 @@ def buffer_dump(state: AppState, path: str):
writer.writerow(row)
state.buffer_dump_thread = None
def autosave_worker(state: AppState, path: str) -> None:
output_dir = Path(path)
ctr: int = 0
while state.autosave_enabled:
date_str = state.start_time.strftime("%m_%d_%Y_%H_%M")
filename = date_str + ".csv"
save_path = output_dir / filename
buffer_dump(state, save_path)
print(f"Autosave {ctr} complete")
ctr += 1
time.sleep(30)