Spaces:
Runtime error
Runtime error
File size: 8,415 Bytes
d64fd55 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 | # src/tools/viz_executor.py
import matplotlib
matplotlib.use('Agg')
import matplotlib.pyplot as plt
import matplotlib.dates as mdates
from datetime import datetime
from typing import List, Dict, Any
from domain.training.charts import ChartSpec, ChartType
from .interfaces import IVisualizationExecutor
from src.tools.runner_ai import calculate_pace_s_per_km
class TemplateVisualizationExecutor(IVisualizationExecutor):
def render_chart(self, chart_spec: ChartSpec, features: List[Dict[str, Any]]) -> plt.Figure:
if chart_spec.chart_type == ChartType.PACE:
return self._render_pace_chart(features, chart_spec.title or "Pace over Time")
elif chart_spec.chart_type == ChartType.HEART_RATE:
return self._render_hr_chart(features, chart_spec.title or "Heart Rate over Time")
elif chart_spec.chart_type == ChartType.VOLUME:
return self._render_volume_chart(features, chart_spec.title or "Weekly Distance")
elif chart_spec.chart_type == ChartType.FREQUENCY:
return self._render_frequency_chart(features, chart_spec.title or "Run Frequency")
elif chart_spec.chart_type == ChartType.ZONES:
return self._render_zones_chart(features, chart_spec.title or "Effort Zones")
elif chart_spec.chart_type == ChartType.DYNAMIC:
return self._render_dynamic_chart(chart_spec.params, chart_spec.title or "Data Visualization")
else:
return self._empty_chart(f"Unsupported Chart Type: {chart_spec.chart_type}")
def _render_pace_chart(self, features: List[Dict[str, Any]], title: str) -> plt.Figure:
fig, ax = plt.subplots(figsize=(10, 5))
dates = []
paces = []
for run in features:
dt = run.get("start_time")
pace_s = calculate_pace_s_per_km(
run.get("total_distance_m"), run.get("total_duration_s")
)
if dt and pace_s:
dates.append(dt)
# Convert s/km to min/km (float)
paces.append(pace_s / 60.0)
if not dates:
return self._empty_chart(title)
# Convert strings to datetime objects if needed
processed_dates = []
for d in dates:
if isinstance(d, str):
try:
# Handle Z suffix for UTC
processed_dates.append(datetime.fromisoformat(d.replace("Z", "+00:00")))
except ValueError:
processed_dates.append(d)
else:
processed_dates.append(d)
ax.plot(processed_dates, paces, marker="o", linestyle="-", color="blue")
ax.set_title(title)
ax.set_ylabel("Pace (min/km)")
ax.set_xlabel("Date")
ax.grid(True)
# Invert y-axis for pace (smaller min/km is faster)
ax.invert_yaxis()
# Format dates as DD-MMM
ax.xaxis.set_major_formatter(mdates.DateFormatter('%d-%b'))
fig.autofmt_xdate()
plt.close(fig)
return fig
def _render_hr_chart(self, features: List[Dict[str, Any]], title: str) -> plt.Figure:
fig, ax = plt.subplots(figsize=(10, 5))
dates = []
hrs = []
for run in features:
dt = run.get("start_time")
hr = run.get("avg_hr_bpm")
if dt and hr:
dates.append(dt)
hrs.append(hr)
if not dates:
return self._empty_chart(title)
# Convert strings to datetime objects if needed
processed_dates = []
for d in dates:
if isinstance(d, str):
try:
processed_dates.append(datetime.fromisoformat(d.replace("Z", "+00:00")))
except ValueError:
processed_dates.append(d)
else:
processed_dates.append(d)
ax.plot(processed_dates, hrs, marker="s", linestyle="--", color="red")
ax.set_title(title)
ax.set_ylabel("Heart Rate (bpm)")
ax.set_xlabel("Date")
ax.grid(True)
# Format dates as DD-MMM
ax.xaxis.set_major_formatter(mdates.DateFormatter('%d-%b'))
fig.autofmt_xdate()
return fig
def _render_volume_chart(self, features: List[Dict[str, Any]], title: str) -> plt.Figure:
from collections import defaultdict
fig, ax = plt.subplots(figsize=(10, 5))
# Group by week
weekly_km = defaultdict(float)
for f in features:
dt = f.get("start_time")
if isinstance(dt, str):
dt = datetime.fromisoformat(dt.replace("Z", "+00:00"))
if dt:
# ISO Week
year, week, _ = dt.isocalendar()
key = f"{year}-W{week:02d}"
weekly_km[key] += f.get("total_distance_m", 0) / 1000.0
if not weekly_km:
return self._empty_chart(title)
sorted_weeks = sorted(weekly_km.keys())
km_values = [weekly_km[w] for w in sorted_weeks]
ax.bar(sorted_weeks, km_values, color="teal", alpha=0.7)
ax.set_title(title)
ax.set_ylabel("Distance (km)")
ax.set_xlabel("Week")
ax.grid(axis='y', linestyle='--', alpha=0.7)
return fig
def _render_frequency_chart(self, features: List[Dict[str, Any]], title: str) -> plt.Figure:
from collections import defaultdict
fig, ax = plt.subplots(figsize=(10, 5))
# Group by day of week
dow_counts = defaultdict(int)
days = ["Mon", "Tue", "Wed", "Thu", "Fri", "Sat", "Sun"]
for f in features:
dt = f.get("start_time")
if isinstance(dt, str):
dt = datetime.fromisoformat(dt.replace("Z", "+00:00"))
if dt:
dow = dt.weekday()
dow_counts[dow] += 1
if not dow_counts:
return self._empty_chart(title)
counts = [dow_counts[i] for i in range(7)]
ax.bar(days, counts, color="orange", alpha=0.7)
ax.set_title(title)
ax.set_ylabel("Number of Runs")
ax.grid(axis='y', linestyle='--', alpha=0.7)
return fig
def _render_zones_chart(self, features: List[Dict[str, Any]], title: str) -> plt.Figure:
# Simple placeholder for HR zones distribution if data available
# In this MVP, we might not have zones calculated per run, but we can do a mock distribution
fig, ax = plt.subplots(figsize=(8, 8))
labels = ['Zone 1', 'Zone 2', 'Zone 3', 'Zone 4', 'Zone 5']
# Mock logic based on average HR
avg_hrs = [f.get("avg_hr_bpm", 0) for f in features if f.get("avg_hr_bpm")]
if not avg_hrs:
return self._empty_chart(title)
# Simplified: just count runs whose avg HR falls into zones (assuming max HR 190)
distribution = [0] * 5
for hr in avg_hrs:
if hr < 130: distribution[0] += 1
elif hr < 150: distribution[1] += 1
elif hr < 165: distribution[2] += 1
elif hr < 180: distribution[3] += 1
else: distribution[4] += 1
ax.pie(distribution, labels=labels, autopct='%1.1f%%', colors=plt.cm.RdYlGn_r([0.2, 0.4, 0.6, 0.8, 1.0]))
ax.set_title(title)
return fig
def _render_dynamic_chart(self, params: Dict[str, Any], title: str) -> plt.Figure:
labels = params.get("labels", [])
values = params.get("values", [])
chart_type = params.get("type", "bar")
if not labels or not values:
return self._empty_chart(title)
fig, ax = plt.subplots(figsize=(10, 5))
if chart_type == "bar":
ax.bar(labels, values, color="skyblue")
elif chart_type == "pie":
ax.pie(values, labels=labels, autopct='%1.1f%%')
else:
ax.plot(labels, values, marker='o')
ax.set_title(title)
ax.grid(True, axis='y', alpha=0.3)
return fig
def _empty_chart(self, title: str) -> plt.Figure:
fig, ax = plt.subplots(figsize=(6, 3))
ax.text(0.5, 0.5, "No Data Available", ha="center", va="center")
ax.set_title(title)
ax.axis("off")
return fig
|