ggw/chest_strap
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

improve analysis

Browse Source
This commit is contained in:
ggw
2026-05-30 19:13:22 -05:00
parent 612cbe59ef
commit 990eb95303
1 changed files with 45 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
code/l452_code/packet_parser_helpers.py
+45 -12
View File
@@ -4,8 +4,6 @@ import matplotlib.pyplot as plt
import scipy as sp
import time
# Try using filtfilt instead of filtfilt
matplotlib.rcParams['legend.framealpha'] = 1.0
packet_definitions = b"""packet_rtc 1 28 uint32_t t 0 4 RTC_TimeTypeDef sTime 4 20 RTC_DateTypeDef sDate 24 4
@@ -39,6 +37,25 @@ def get_type_list(lines):
i += 4
return types
class vbatt_vars():
charges = []
ts = []
def process_vbatt(d, t):
global vbatt_vars
for e in t['elements']:
block = d[e['offset']:e['offset'] + e['size']]
element_size = int(len(block) / e['n_elements'])
if e['name'] == b't':
vbatt_vars.ts.append((1 / 2000) * int.from_bytes(block[:4], byteorder = 'little', signed = True))
if e['name'] == b'vbatt_cnts':
vbatt_vars.charges.append(2 * 3.3 * int.from_bytes(block, byteorder = 'little', signed = False) / (1 << 12))
if len(vbatt_vars.charges) % 10 == 9:
plt.plot(vbatt_vars.ts, vbatt_vars.charges)
plt.savefig("vbatt.png")
plt.close()
class ppg_vars():
@@ -352,17 +369,20 @@ def adc_graphs():
axs[1].plot(np.array(r_peaks_[:-1]) / adc_vars.freq_Hz, 1000 * np.diff(r_peaks_) / adc_vars.freq_Hz, 'k.', alpha = 0.5)
hrv = np.abs(1000 * np.diff(np.diff(r_peaks_)) / adc_vars.freq_Hz)
hrv_ax.plot(np.array(r_peaks_[2:]) / adc_vars.freq_Hz, hrv, 'b.', alpha = 0.5)
if len(r_peaks_) > 10:
N = int(np.ceil(len(r_peaks_) / 30))
N = int(np.floor(len(r_peaks_) / 30))
for i in range(N):
block = 1000 * np.array(r_peaks_[30 * i : min(len(r_peaks_), 30 * i + 30)]) / adc_vars.freq_Hz
rmssd = np.power(np.sum(np.power(np.diff(np.diff(block)), 2.0)) / (len(block) - 2), 0.5)
sdnn = np.std(np.diff(block))
pNN50 = 100 * np.mean(np.abs(np.diff(np.diff(block))) > 50)
hrv_ax.plot([block[0] / 1000, block[-1] / 1000], [rmssd, rmssd], 'r', alpha = 0.5)
hrv_ax.plot([block[0] / 1000, block[-1] / 1000], [sdnn, sdnn], 'm', alpha = 0.5)
hrv_ax.plot([block[0] / 1000, block[-1] / 1000], [pNN50, pNN50], color = 'pink', alpha = 0.5)
if len(block) > 2:
bpm = 60 * len(block) / ((block[-1] - block[0]) / 1000)
rmssd = np.power(np.sum(np.power(np.diff(np.diff(block)), 2.0)) / (len(block) - 2), 0.5)
sdnn = np.std(np.diff(block))
pNN50 = 100 * np.mean(np.abs(np.diff(np.diff(block))) > 50)
hrv_ax.plot([block[0] / 1000, block[-1] / 1000], [rmssd, rmssd], 'r', alpha = 0.5)
hrv_ax.plot([block[0] / 1000, block[-1] / 1000], [sdnn, sdnn], 'm', alpha = 0.5)
hrv_ax.plot([block[0] / 1000, block[-1] / 1000], [pNN50, pNN50], color = 'pink', alpha = 0.5)
hrv_ax.minorticks_on()
hrv_ax.grid(alpha = 0.5)
@@ -376,8 +396,19 @@ def adc_graphs():
axs[1].plot([],'k.',label = 'R-R int', alpha = 0.5)
axs[1].plot([],'b.',label = '|delta(R-R int)|', alpha = 0.5)
axs[1].plot([],color = 'orange', label = 'Breath (tension)', alpha = 0.5)
axs[1].plot([],color = 'red', label = 'RMSSD', alpha = 0.5)
axs[1].plot([],color = 'magenta', label = 'SSDN', alpha = 0.5)
try:
axs[1].plot([],color = 'red', label = f'RMSSD {int(rmssd)}ms', alpha = 0.5)
except NameError:
axs[1].plot([],color = 'red', label = f'RMSSD', alpha = 0.5)
try:
axs[1].plot([],color = 'magenta', label = f'SSDN {int(sdnn)}ms', alpha = 0.5)
except NameError:
axs[1].plot([],color = 'magenta', label = 'SSDN', alpha = 0.5)
try:
axs[1].plot([],color = 'black', label = f'BPM {int(bpm)}', alpha = 0.5)
except:
pass
axs[1].plot([],color = 'pink', label = 'pNN50', alpha = 0.5)
axs[1].legend(loc = 'lower center', bbox_to_anchor = (0.5, 1.02), ncol = 4)
axs[1].set_xlabel("t (s)")
@@ -502,6 +533,8 @@ def read_and_process(types, cons, size):
process_adc(d, t)
if t['type_name'] == b'packet_spo2':
process_ppg(d, t)
if t['type_name'] == b'packet_vbatt':
process_vbatt(d, t)
index += 1 + t['size']