bolade/bio-performx
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

added extra page

Browse Source
This commit is contained in:
bolade
2025-11-24 19:37:28 +01:00
parent 580ad5d248
commit 8e8280bcb0
11 changed files with 1073 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/services/graph_generator.py
+234
View File
@@ -6,6 +6,7 @@ Based on the analysis notebooks in services_dfdf/.
"""
import base64
import io
from pathlib import Path
import matplotlib
@@ -1680,3 +1681,236 @@ class GraphGenerator:
)
plt.close(fig)
return str(output_path)
def generate_muscle_oxygenation_chart(
self, oxygenation_df: pd.DataFrame, save_as_base64: bool = True
) -> tuple:
"""
Generate comprehensive muscle oxygenation (SmO2) chart with both legs and heart rate.
Args:
oxygenation_df: DataFrame with muscle oxygenation data (Train.Red CSV format)
save_as_base64: If True, return base64 string, else return file path
Returns:
Tuple of (chart_string, metrics_dict) where metrics_dict contains key values
"""
# Data preparation
df_oxy = oxygenation_df.copy()
# Convert columns to numeric
df_oxy["Timestamp (seconds passed)"] = pd.to_numeric(
df_oxy["Timestamp (seconds passed)"], errors="coerce"
)
df_oxy["Left_SmO2"] = pd.to_numeric(df_oxy["SmO2"], errors="coerce")
df_oxy["Right_SmO2"] = pd.to_numeric(df_oxy["SmO2.1"], errors="coerce")
df_oxy["Heart_Rate"] = pd.to_numeric(
df_oxy["Heart Rate (BPM)"], errors="coerce"
)
df_oxy["Lap"] = pd.to_numeric(df_oxy["Lap/Event"], errors="coerce")
# Drop rows with missing timestamps
df_oxy = df_oxy.dropna(subset=["Timestamp (seconds passed)"])
df_oxy = df_oxy.sort_values("Timestamp (seconds passed)").reset_index(drop=True)
# Apply 10-second rolling mean smoothing
time_diffs = df_oxy["Timestamp (seconds passed)"].diff().dropna()
avg_sampling_interval = time_diffs.median()
sampling_freq = 1 / avg_sampling_interval if avg_sampling_interval > 0 else 10
window_samples = int(10 * sampling_freq)
df_oxy["Left_SmO2_smooth"] = (
df_oxy["Left_SmO2"]
.rolling(window=window_samples, center=True, min_periods=1)
.mean()
)
df_oxy["Right_SmO2_smooth"] = (
df_oxy["Right_SmO2"]
.rolling(window=window_samples, center=True, min_periods=1)
.mean()
)
df_oxy["Heart_Rate_smooth"] = (
df_oxy["Heart_Rate"]
.rolling(window=window_samples, center=True, min_periods=1)
.mean()
)
# Identify stage boundaries
lap_changes = df_oxy[df_oxy["Lap"].diff() != 0].copy()
lap_starts = {}
for idx, row in lap_changes.iterrows():
lap_num = int(row["Lap"])
lap_starts[lap_num] = row["Timestamp (seconds passed)"]
warm_up_end = lap_starts.get(1, df_oxy["Timestamp (seconds passed)"].max())
recovery_start = lap_starts.get(7, df_oxy["Timestamp (seconds passed)"].max())
# Calculate recovery percentages
warm_up_last_30_start = warm_up_end - 30
warm_up_mask = (
df_oxy["Timestamp (seconds passed)"] >= warm_up_last_30_start
) & (df_oxy["Timestamp (seconds passed)"] <= warm_up_end)
recovery_end = df_oxy["Timestamp (seconds passed)"].max()
recovery_last_30_start = recovery_end - 30
recovery_mask = (
df_oxy["Timestamp (seconds passed)"] >= recovery_last_30_start
) & (df_oxy["Timestamp (seconds passed)"] <= recovery_end)
left_warmup_avg = df_oxy.loc[warm_up_mask, "Left_SmO2_smooth"].mean()
left_recovery_avg = df_oxy.loc[recovery_mask, "Left_SmO2_smooth"].mean()
left_recovery_pct = round((left_recovery_avg / left_warmup_avg) * 100)
right_warmup_avg = df_oxy.loc[warm_up_mask, "Right_SmO2_smooth"].mean()
right_recovery_avg = df_oxy.loc[recovery_mask, "Right_SmO2_smooth"].mean()
right_recovery_pct = round((right_recovery_avg / right_warmup_avg) * 100)
# Calculate key metrics
active_mask = (df_oxy["Timestamp (seconds passed)"] >= warm_up_end) & (
df_oxy["Timestamp (seconds passed)"] <= recovery_start
)
active_data = df_oxy[active_mask]
left_min = active_data["Left_SmO2_smooth"].min()
left_min_lap = int(
active_data.loc[active_data["Left_SmO2_smooth"].idxmin(), "Lap"]
)
right_min = active_data["Right_SmO2_smooth"].min()
right_min_lap = int(
active_data.loc[active_data["Right_SmO2_smooth"].idxmin(), "Lap"]
)
left_drop = left_warmup_avg - left_min
right_drop = right_warmup_avg - right_min
hr_warmup = df_oxy[df_oxy["Timestamp (seconds passed)"] <= warm_up_end][
"Heart_Rate_smooth"
].mean()
hr_max = active_data["Heart_Rate_smooth"].max()
# Create the plot
fig, ax1 = plt.subplots(figsize=(18, 8))
time = df_oxy["Timestamp (seconds passed)"]
ax1.plot(
time,
df_oxy["Left_SmO2_smooth"],
label=f"Left SmO₂ (Rec {left_recovery_pct}% of warm-up)",
color="#2E86AB",
linewidth=2,
)
ax1.plot(
time,
df_oxy["Right_SmO2_smooth"],
label=f"Right SmO₂ (Rec {right_recovery_pct}% of warm-up)",
color="#A23B72",
linewidth=2,
)
ax1.set_xlabel("Time (seconds)", fontsize=12, fontweight="bold")
ax1.set_ylabel("SmO₂ (%)", fontsize=12, fontweight="bold")
ax1.tick_params(axis="y", labelcolor="black")
ax1.grid(True, alpha=0.3, linestyle="--")
# Add secondary axis for heart rate
ax2 = ax1.twinx()
ax2.plot(
time,
df_oxy["Heart_Rate_smooth"],
label="Heart Rate",
color="red",
linewidth=1.5,
linestyle="--",
alpha=0.7,
)
ax2.set_ylabel("Heart Rate (BPM)", fontsize=12, fontweight="bold", color="red")
ax2.tick_params(axis="y", labelcolor="red")
# Add shaded regions
ax1.axvspan(0, warm_up_end, alpha=0.15, color="blue", label="Warm-up")
active_laps = [1, 2, 3, 4, 5, 6]
colors_active = ["yellow", "orange"] * 3
for i, lap in enumerate(active_laps):
start = lap_starts.get(lap, 0)
end = lap_starts.get(lap + 1, recovery_start) if lap < 6 else recovery_start
ax1.axvspan(start, end, alpha=0.1, color=colors_active[i])
ax1.axvspan(
recovery_start, recovery_end, alpha=0.2, color="gray", label="Recovery"
)
ax1.axvline(
x=recovery_start, color="black", linestyle="-", linewidth=2, alpha=0.7
)
# Add lap labels
for lap in range(1, 7):
start = lap_starts.get(lap, 0)
end = lap_starts.get(lap + 1, recovery_start) if lap < 6 else recovery_start
mid = (start + end) / 2
ax1.text(
mid,
ax1.get_ylim()[1] * 0.97,
f"Lap {lap}",
ha="center",
va="top",
fontsize=10,
fontweight="bold",
bbox=dict(boxstyle="round,pad=0.3", facecolor="white", alpha=0.7),
)
plt.title(
"Train.Red SmO₂ Ramp - Muscle Oxygenation Analysis",
fontsize=16,
fontweight="bold",
pad=20,
)
# Combine legends
lines1, labels1 = ax1.get_legend_handles_labels()
lines2, labels2 = ax2.get_legend_handles_labels()
ax1.legend(
lines1 + lines2,
labels1 + labels2,
loc="upper left",
fontsize=10,
framealpha=0.9,
)
plt.tight_layout()
# Prepare metrics dictionary
metrics = {
"left_baseline_smo2": f"{left_warmup_avg:.1f}%",
"right_baseline_smo2": f"{right_warmup_avg:.1f}%",
"left_minimum_smo2": f"{left_min:.1f}%",
"right_minimum_smo2": f"{right_min:.1f}%",
"left_minimum_lap": f"Lap {left_min_lap}",
"right_minimum_lap": f"Lap {right_min_lap}",
"left_oxygen_drop": f"{left_drop:.1f}%",
"right_oxygen_drop": f"{right_drop:.1f}%",
"left_drop_percentage": f"{(left_drop / left_warmup_avg * 100):.0f}% decrease",
"right_drop_percentage": f"{(right_drop / right_warmup_avg * 100):.0f}% decrease",
"left_recovery_percentage": f"{left_recovery_pct}%",
"right_recovery_percentage": f"{right_recovery_pct}%",
"hr_warmup": f"{hr_warmup:.0f}",
"hr_max": f"{hr_max:.0f}",
"test_duration": f"~{(recovery_start - warm_up_end) / 60:.0f} minutes active test",
"recovery_assessment": "Excellent recovery capacity"
if (left_recovery_pct + right_recovery_pct) / 2 >= 100
else "Good recovery capacity",
}
# Save or return
if save_as_base64:
buf = io.BytesIO()
plt.savefig(buf, format="png", dpi=300, bbox_inches="tight")
plt.close(fig)
buf.seek(0)
chart_str = base64.b64encode(buf.read()).decode("utf-8")
return chart_str, metrics
else:
output_path = self.charts_dir / "muscle_oxygenation_chart.png"
plt.savefig(output_path, dpi=300, bbox_inches="tight")
plt.close(fig)
return str(output_path), metrics