583 lines
22 KiB
Python
583 lines
22 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
|
|
from fastapi import APIRouter, Form, HTTPException, Request, status
|
|
from fastapi.responses import RedirectResponse
|
|
from fastapi.templating import Jinja2Templates
|
|
|
|
router = APIRouter()
|
|
templates = Jinja2Templates(directory='app/templates')
|
|
|
|
|
|
@router.get('/')
|
|
def home(request: Request):
|
|
return templates.TemplateResponse(
|
|
request=request,
|
|
name='dashboard.html',
|
|
context=_dashboard_context(request),
|
|
)
|
|
|
|
|
|
@router.post('/log/weight', status_code=303)
|
|
def log_weight(
|
|
request: Request,
|
|
user_id: str = Form(...),
|
|
recorded_at: str = Form(...),
|
|
weight_lbs: float = Form(...),
|
|
body_fat_percent: str = Form(''),
|
|
notes: str = Form(''),
|
|
):
|
|
payload = {
|
|
'recorded_at': recorded_at,
|
|
'weight_lbs': weight_lbs,
|
|
'body_fat_percent': _optional_float(body_fat_percent),
|
|
'notes': notes.strip() or None,
|
|
}
|
|
request.app.state.store.add_weight_entry(user_id, payload)
|
|
return RedirectResponse(url='/', status_code=status.HTTP_303_SEE_OTHER)
|
|
|
|
|
|
@router.post('/log/measurements', status_code=303)
|
|
def log_measurements(
|
|
request: Request,
|
|
user_id: str = Form(...),
|
|
recorded_at: str = Form(...),
|
|
waist_inches: str = Form(''),
|
|
hip_inches: str = Form(''),
|
|
chest_inches: str = Form(''),
|
|
neck_inches: str = Form(''),
|
|
notes: str = Form(''),
|
|
):
|
|
payload = {
|
|
'recorded_at': recorded_at,
|
|
'waist_inches': _optional_float(waist_inches),
|
|
'hip_inches': _optional_float(hip_inches),
|
|
'chest_inches': _optional_float(chest_inches),
|
|
'neck_inches': _optional_float(neck_inches),
|
|
'notes': notes.strip() or None,
|
|
}
|
|
if not any(
|
|
payload[key] is not None
|
|
for key in ('waist_inches', 'hip_inches', 'chest_inches', 'neck_inches')
|
|
):
|
|
raise HTTPException(status_code=400, detail='At least one body measurement is required.')
|
|
request.app.state.store.add_measurement_entry(user_id, payload)
|
|
return RedirectResponse(url='/', status_code=status.HTTP_303_SEE_OTHER)
|
|
|
|
|
|
@router.post('/templates', status_code=303)
|
|
def create_template(
|
|
request: Request,
|
|
name: str = Form(...),
|
|
primary_muscle: str = Form(''),
|
|
equipment: str = Form(''),
|
|
notes: str = Form(''),
|
|
):
|
|
request.app.state.store.add_exercise_template(
|
|
{
|
|
'name': name.strip(),
|
|
'primary_muscle': primary_muscle.strip() or None,
|
|
'equipment': equipment.strip() or None,
|
|
'notes': notes.strip() or None,
|
|
}
|
|
)
|
|
return RedirectResponse(url='/', status_code=status.HTTP_303_SEE_OTHER)
|
|
|
|
|
|
@router.post('/routines', status_code=303)
|
|
def create_routine(
|
|
request: Request,
|
|
user_id: str = Form(...),
|
|
name: str = Form(...),
|
|
exercise_lines: str = Form(...),
|
|
notes: str = Form(''),
|
|
):
|
|
payload = {
|
|
'user_id': user_id,
|
|
'name': name.strip(),
|
|
'notes': notes.strip() or None,
|
|
'exercises': _parse_routine_exercises(exercise_lines),
|
|
}
|
|
request.app.state.store.add_routine(payload)
|
|
return RedirectResponse(url='/', status_code=status.HTTP_303_SEE_OTHER)
|
|
|
|
|
|
@router.post('/workouts/quick-log', status_code=303)
|
|
def quick_log_workout(
|
|
request: Request,
|
|
user_id: str = Form(...),
|
|
performed_at: str = Form(...),
|
|
routine_name: str = Form(...),
|
|
exercise_lines: str = Form(...),
|
|
notes: str = Form(''),
|
|
):
|
|
payload = {
|
|
'user_id': user_id,
|
|
'performed_at': performed_at,
|
|
'routine_name': routine_name.strip(),
|
|
'notes': notes.strip() or None,
|
|
'exercises': _parse_workout_exercises(exercise_lines),
|
|
}
|
|
request.app.state.store.add_workout(payload)
|
|
return RedirectResponse(url='/', status_code=status.HTTP_303_SEE_OTHER)
|
|
|
|
|
|
@router.get('/routines/{routine_id}/log')
|
|
def routine_log_sheet(request: Request, routine_id: str):
|
|
routine = _get_routine_or_404(request, routine_id)
|
|
return templates.TemplateResponse(
|
|
request=request,
|
|
name='routine_log.html',
|
|
context={
|
|
'title': f"Log {routine['name']}",
|
|
'routine': routine,
|
|
'routine_sheet': _build_routine_log_sheet(request, routine),
|
|
'exercise_options': _exercise_options_for_routine(request, routine),
|
|
'routine_action_label': _routine_action_label(routine['name']),
|
|
'default_performed_at': _default_performed_at_value(),
|
|
},
|
|
)
|
|
|
|
|
|
@router.get('/routines/{routine_id}/live')
|
|
def live_workout_logger(request: Request, routine_id: str):
|
|
routine = _get_routine_or_404(request, routine_id)
|
|
return templates.TemplateResponse(
|
|
request=request,
|
|
name='live_workout.html',
|
|
context=_live_workout_context(request, routine),
|
|
)
|
|
|
|
|
|
@router.post('/routines/{routine_id}/live/start', status_code=303)
|
|
def start_live_workout_session(
|
|
request: Request,
|
|
routine_id: str,
|
|
user_id: str = Form(...),
|
|
performed_at: str = Form(...),
|
|
):
|
|
routine = _get_routine_or_404(request, routine_id)
|
|
if routine['user_id'] != user_id:
|
|
raise HTTPException(status_code=400, detail='Routine does not belong to that user.')
|
|
request.app.state.store.start_active_workout_session(
|
|
{
|
|
'routine_id': routine['id'],
|
|
'routine_name': routine['name'],
|
|
'user_id': user_id,
|
|
'performed_at': performed_at,
|
|
}
|
|
)
|
|
return RedirectResponse(url=f"/routines/{routine_id}/live", status_code=status.HTTP_303_SEE_OTHER)
|
|
|
|
|
|
@router.post('/routines/{routine_id}/live/add-exercise', status_code=303)
|
|
async def add_live_workout_exercise(
|
|
request: Request,
|
|
routine_id: str,
|
|
user_id: str = Form(...),
|
|
performed_at: str = Form(...),
|
|
exercise_name: str = Form(...),
|
|
set_count: int = Form(...),
|
|
):
|
|
routine = _get_routine_or_404(request, routine_id)
|
|
if routine['user_id'] != user_id:
|
|
raise HTTPException(status_code=400, detail='Routine does not belong to that user.')
|
|
active_session = request.app.state.store.get_active_workout_session(routine_id, user_id)
|
|
if active_session is None:
|
|
request.app.state.store.start_active_workout_session(
|
|
{
|
|
'routine_id': routine['id'],
|
|
'routine_name': routine['name'],
|
|
'user_id': user_id,
|
|
'performed_at': performed_at,
|
|
}
|
|
)
|
|
exercise = await _parse_live_workout_exercise_form(request, routine, exercise_name, set_count)
|
|
request.app.state.store.add_active_workout_exercise(routine_id, user_id, exercise)
|
|
return RedirectResponse(url=f"/routines/{routine_id}/live", status_code=status.HTTP_303_SEE_OTHER)
|
|
|
|
|
|
@router.post('/routines/{routine_id}/live/finish', status_code=303)
|
|
def finish_live_workout_session(
|
|
request: Request,
|
|
routine_id: str,
|
|
user_id: str = Form(...),
|
|
performed_at: str = Form(...),
|
|
notes: str = Form(''),
|
|
):
|
|
routine = _get_routine_or_404(request, routine_id)
|
|
if routine['user_id'] != user_id:
|
|
raise HTTPException(status_code=400, detail='Routine does not belong to that user.')
|
|
active_session = request.app.state.store.get_active_workout_session(routine_id, user_id)
|
|
if active_session is None:
|
|
raise HTTPException(status_code=400, detail='No live workout session in progress.')
|
|
if active_session['performed_at'] != performed_at:
|
|
raise HTTPException(status_code=400, detail='Performed-at timestamp does not match the active session.')
|
|
if not active_session.get('exercises'):
|
|
raise HTTPException(status_code=400, detail='Log at least one exercise before finishing the session.')
|
|
request.app.state.store.finish_active_workout_session(routine_id, user_id, notes.strip() or None)
|
|
return RedirectResponse(url='/', status_code=status.HTTP_303_SEE_OTHER)
|
|
|
|
|
|
@router.post('/workouts/routine-log', status_code=303)
|
|
async def log_seeded_routine_workout(
|
|
request: Request,
|
|
user_id: str = Form(...),
|
|
routine_id: str = Form(...),
|
|
performed_at: str = Form(...),
|
|
routine_name: str = Form(...),
|
|
exercise_count: int = Form(...),
|
|
notes: str = Form(''),
|
|
):
|
|
routine = _get_routine_or_404(request, routine_id)
|
|
if routine['user_id'] != user_id:
|
|
raise HTTPException(status_code=400, detail='Routine does not belong to that user.')
|
|
|
|
payload = {
|
|
'user_id': user_id,
|
|
'performed_at': performed_at,
|
|
'routine_name': routine_name.strip(),
|
|
'notes': notes.strip() or None,
|
|
'exercises': await _parse_structured_routine_form(request, routine, exercise_count),
|
|
}
|
|
request.app.state.store.add_workout(payload)
|
|
return RedirectResponse(url='/', status_code=status.HTTP_303_SEE_OTHER)
|
|
|
|
|
|
def _dashboard_context(request: Request) -> dict:
|
|
store = request.app.state.store
|
|
summary = store.dashboard_summary()
|
|
routines_with_live_state = []
|
|
for routine in summary['routines']:
|
|
active_session = store.get_active_workout_session(routine['id'], routine['user_id'])
|
|
routines_with_live_state.append({**routine, 'live_session': active_session})
|
|
summary['routines'] = routines_with_live_state
|
|
return {
|
|
'title': 'Doris Barbell',
|
|
'summary': summary,
|
|
'john_weight_history': store.weight_history('john')[:5],
|
|
'john_measurement_history': store.measurement_history('john')[:5],
|
|
'john_workout_history': store.workout_history('john')[:5],
|
|
'john_progression': store.exercise_progression('john')[:5],
|
|
}
|
|
|
|
|
|
def _live_workout_context(request: Request, routine: dict) -> dict:
|
|
active_session = request.app.state.store.get_active_workout_session(routine['id'], routine['user_id'])
|
|
routine_sheet = _build_routine_log_sheet(request, routine)
|
|
remaining_exercises = _remaining_live_exercises(routine_sheet, active_session)
|
|
logged_exercises = []
|
|
if active_session is not None:
|
|
for exercise in active_session.get('exercises', []):
|
|
logged_exercises.append(
|
|
{
|
|
'exercise_name': exercise['exercise_name'],
|
|
'set_count': len(exercise.get('sets', [])),
|
|
'notes': exercise.get('notes'),
|
|
}
|
|
)
|
|
return {
|
|
'title': f"Live log · {routine['name']}",
|
|
'routine': routine,
|
|
'routine_sheet': routine_sheet,
|
|
'active_session': active_session,
|
|
'logged_exercises': logged_exercises,
|
|
'remaining_exercises': remaining_exercises,
|
|
'default_performed_at': active_session['performed_at'] if active_session else _default_performed_at_value(),
|
|
'routine_action_label': _routine_action_label(routine['name']),
|
|
}
|
|
|
|
|
|
def _remaining_live_exercises(routine_sheet: list[dict], active_session: dict | None) -> list[dict]:
|
|
completed_counts: dict[str, int] = {}
|
|
if active_session is not None:
|
|
for exercise in active_session.get('exercises', []):
|
|
normalized_name = exercise['exercise_name'].strip().lower()
|
|
completed_counts[normalized_name] = completed_counts.get(normalized_name, 0) + 1
|
|
remaining = []
|
|
for exercise in routine_sheet:
|
|
normalized_name = exercise['exercise_name'].strip().lower()
|
|
completed_count = completed_counts.get(normalized_name, 0)
|
|
if completed_count > 0:
|
|
completed_counts[normalized_name] = completed_count - 1
|
|
continue
|
|
remaining.append(exercise)
|
|
return remaining
|
|
|
|
|
|
def _optional_float(value: str) -> float | None:
|
|
cleaned = value.strip()
|
|
if not cleaned:
|
|
return None
|
|
try:
|
|
return float(cleaned)
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=400, detail=f'Invalid numeric value: {value}') from exc
|
|
|
|
|
|
def _parse_routine_exercises(raw_lines: str) -> list[dict]:
|
|
exercises = []
|
|
for line in raw_lines.splitlines():
|
|
cleaned = line.strip()
|
|
if not cleaned:
|
|
continue
|
|
parts = [part.strip() for part in cleaned.split('|')]
|
|
if len(parts) < 3:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail='Routine lines must use exercise|sets|reps|optional notes format.',
|
|
)
|
|
exercise_name, target_sets_raw, target_reps = parts[:3]
|
|
try:
|
|
target_sets = int(target_sets_raw)
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=400, detail=f'Invalid target sets: {target_sets_raw}') from exc
|
|
exercises.append(
|
|
{
|
|
'exercise_name': exercise_name,
|
|
'target_sets': target_sets,
|
|
'target_reps': target_reps,
|
|
'notes': parts[3] if len(parts) > 3 and parts[3] else None,
|
|
}
|
|
)
|
|
if not exercises:
|
|
raise HTTPException(status_code=400, detail='At least one routine exercise is required.')
|
|
return exercises
|
|
|
|
|
|
def _parse_workout_exercises(raw_lines: str) -> list[dict]:
|
|
exercises = []
|
|
for line in raw_lines.splitlines():
|
|
cleaned = line.strip()
|
|
if not cleaned:
|
|
continue
|
|
parts = [part.strip() for part in cleaned.split('|')]
|
|
if len(parts) < 2:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail='Workout lines must use exercise|reps@weight, reps@weight|optional notes format.',
|
|
)
|
|
exercise_name = parts[0]
|
|
notes = parts[2] if len(parts) > 2 and parts[2] else None
|
|
sets = []
|
|
for set_chunk in parts[1].split(','):
|
|
token = set_chunk.strip()
|
|
if not token:
|
|
continue
|
|
if '@' not in token:
|
|
raise HTTPException(status_code=400, detail=f'Invalid set token: {token}')
|
|
reps_raw, weight_raw = [value.strip() for value in token.split('@', 1)]
|
|
try:
|
|
reps = int(reps_raw)
|
|
weight_lbs = float(weight_raw)
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=400, detail=f'Invalid set token: {token}') from exc
|
|
sets.append({'reps': reps, 'weight_lbs': weight_lbs})
|
|
if not sets:
|
|
raise HTTPException(status_code=400, detail='Each workout exercise needs at least one set.')
|
|
exercises.append({'exercise_name': exercise_name, 'notes': notes, 'sets': sets})
|
|
if not exercises:
|
|
raise HTTPException(status_code=400, detail='At least one workout exercise is required.')
|
|
return exercises
|
|
|
|
|
|
def _get_routine_or_404(request: Request, routine_id: str) -> dict:
|
|
routine = request.app.state.store.get_routine(routine_id)
|
|
if routine is None:
|
|
raise HTTPException(status_code=404, detail='Unknown routine')
|
|
return routine
|
|
|
|
|
|
def _build_routine_log_sheet(request: Request, routine: dict) -> list[dict]:
|
|
equipment_by_name = _exercise_equipment_lookup(request)
|
|
sheet = []
|
|
for exercise_index, exercise in enumerate(routine.get('exercises', [])):
|
|
exercise_name = exercise['exercise_name']
|
|
target_sets = max(int(exercise.get('target_sets', 1) or 1), 1)
|
|
default_reps = _default_reps_value(exercise.get('target_reps'))
|
|
is_band = equipment_by_name.get(exercise_name.strip().lower()) == 'band'
|
|
sheet.append(
|
|
{
|
|
'exercise_index': exercise_index,
|
|
'exercise_name': exercise_name,
|
|
'target_sets': target_sets,
|
|
'target_reps': exercise.get('target_reps'),
|
|
'default_reps': default_reps,
|
|
'notes': exercise.get('notes'),
|
|
'is_band': is_band,
|
|
'sets': [
|
|
{
|
|
'exercise_index': exercise_index,
|
|
'index': set_index,
|
|
'default_reps': default_reps,
|
|
'is_band': is_band,
|
|
}
|
|
for set_index in range(target_sets)
|
|
],
|
|
}
|
|
)
|
|
return sheet
|
|
|
|
|
|
def _exercise_options_for_routine(request: Request, routine: dict) -> list[str]:
|
|
routine_names = [exercise['exercise_name'] for exercise in routine.get('exercises', []) if exercise.get('exercise_name')]
|
|
template_names = [
|
|
template['name']
|
|
for template in request.app.state.store.list_exercise_templates()
|
|
if template.get('name')
|
|
]
|
|
ordered = []
|
|
seen = set()
|
|
for name in [*routine_names, *template_names]:
|
|
normalized = name.strip().lower()
|
|
if normalized in seen:
|
|
continue
|
|
ordered.append(name)
|
|
seen.add(normalized)
|
|
return ordered
|
|
|
|
|
|
def _exercise_equipment_lookup(request: Request) -> dict[str, str]:
|
|
lookup: dict[str, str] = {}
|
|
for template in request.app.state.store.list_exercise_templates():
|
|
name = str(template.get('name', '')).strip().lower()
|
|
equipment = str(template.get('equipment', '')).strip().lower()
|
|
if not name or not equipment:
|
|
continue
|
|
lookup[name] = equipment
|
|
return lookup
|
|
|
|
|
|
def _default_reps_value(target_reps: str | None) -> str:
|
|
if target_reps is None:
|
|
return ''
|
|
cleaned = str(target_reps).strip()
|
|
return cleaned if cleaned.isdigit() else ''
|
|
|
|
|
|
def _default_performed_at_value() -> str:
|
|
return datetime.now().astimezone().strftime('%Y-%m-%dT%H:%M')
|
|
|
|
|
|
def _routine_action_label(routine_name: str) -> str:
|
|
if ': ' in routine_name:
|
|
label = routine_name.split(': ', 1)[1].strip().lower()
|
|
else:
|
|
label = routine_name.strip().lower()
|
|
day_aliases = {
|
|
'shoulders': 'shoulder day',
|
|
'arms': 'arm day',
|
|
'legs': 'leg day',
|
|
}
|
|
return day_aliases.get(label, label)
|
|
|
|
|
|
async def _parse_live_workout_exercise_form(
|
|
request: Request,
|
|
routine: dict,
|
|
exercise_name: str,
|
|
set_count: int,
|
|
) -> dict:
|
|
form = await request.form()
|
|
sheet = _build_routine_log_sheet(request, routine)
|
|
matching_exercise = next(
|
|
(exercise for exercise in sheet if exercise['exercise_name'].strip().lower() == exercise_name.strip().lower()),
|
|
None,
|
|
)
|
|
if matching_exercise is None:
|
|
raise HTTPException(status_code=400, detail=f'Unknown routine exercise: {exercise_name}')
|
|
default_reps_raw = matching_exercise.get('default_reps', '')
|
|
is_band_exercise = bool(matching_exercise.get('is_band'))
|
|
sets = []
|
|
for set_index in range(set_count):
|
|
reps_raw = str(form.get(f'set_{set_index}_reps', '')).strip()
|
|
weight_raw = str(form.get(f'set_{set_index}_weight_lbs', '')).strip()
|
|
if not reps_raw and not weight_raw:
|
|
if is_band_exercise and default_reps_raw.isdigit():
|
|
reps_raw = default_reps_raw
|
|
else:
|
|
continue
|
|
if not reps_raw and default_reps_raw.isdigit():
|
|
reps_raw = default_reps_raw
|
|
if not reps_raw:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f'Exercise {exercise_name} set {set_index + 1} needs reps.',
|
|
)
|
|
if not weight_raw and not is_band_exercise:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f'Exercise {exercise_name} set {set_index + 1} needs both reps and weight.',
|
|
)
|
|
try:
|
|
reps = int(reps_raw)
|
|
weight_lbs = None if not weight_raw else float(weight_raw)
|
|
except ValueError as exc:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f'Exercise {exercise_name} set {set_index + 1} has invalid reps or weight.',
|
|
) from exc
|
|
sets.append({'reps': reps, 'weight_lbs': weight_lbs})
|
|
if not sets:
|
|
raise HTTPException(status_code=400, detail=f'Log at least one set for {exercise_name}.')
|
|
exercise_notes = str(form.get('exercise_notes', '')).strip() or None
|
|
return {'exercise_name': matching_exercise['exercise_name'], 'notes': exercise_notes, 'sets': sets}
|
|
|
|
|
|
async def _parse_structured_routine_form(request: Request, routine: dict, exercise_count: int) -> list[dict]:
|
|
form = await request.form()
|
|
routine_exercises = routine.get('exercises', [])
|
|
equipment_by_name = _exercise_equipment_lookup(request)
|
|
exercises = []
|
|
for exercise_index in range(exercise_count):
|
|
exercise_name = str(form.get(f'exercise_{exercise_index}_name', '')).strip()
|
|
if not exercise_name:
|
|
continue
|
|
is_band_exercise = equipment_by_name.get(exercise_name.lower()) == 'band'
|
|
try:
|
|
set_count = int(str(form.get(f'exercise_{exercise_index}_set_count', '0')).strip())
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=400, detail='Invalid set count in routine log form.') from exc
|
|
target_reps = None
|
|
if exercise_index < len(routine_exercises):
|
|
target_reps = routine_exercises[exercise_index].get('target_reps')
|
|
default_reps_raw = _default_reps_value(target_reps)
|
|
sets = []
|
|
for set_index in range(set_count):
|
|
reps_raw = str(form.get(f'exercise_{exercise_index}_set_{set_index}_reps', '')).strip()
|
|
weight_raw = str(form.get(f'exercise_{exercise_index}_set_{set_index}_weight_lbs', '')).strip()
|
|
if not reps_raw and not weight_raw:
|
|
if is_band_exercise and default_reps_raw.isdigit():
|
|
reps_raw = default_reps_raw
|
|
else:
|
|
continue
|
|
if not reps_raw and default_reps_raw.isdigit():
|
|
reps_raw = default_reps_raw
|
|
if not reps_raw:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f'Exercise {exercise_name} set {set_index + 1} needs reps.',
|
|
)
|
|
if not weight_raw and not is_band_exercise:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f'Exercise {exercise_name} set {set_index + 1} needs both reps and weight.',
|
|
)
|
|
try:
|
|
reps = int(reps_raw)
|
|
weight_lbs = None if not weight_raw else float(weight_raw)
|
|
except ValueError as exc:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f'Exercise {exercise_name} set {set_index + 1} has invalid reps or weight.',
|
|
) from exc
|
|
sets.append({'reps': reps, 'weight_lbs': weight_lbs})
|
|
if not sets:
|
|
continue
|
|
exercise_notes = str(form.get(f'exercise_{exercise_index}_notes', '')).strip() or None
|
|
exercises.append({'exercise_name': exercise_name, 'notes': exercise_notes, 'sets': sets})
|
|
if not exercises:
|
|
raise HTTPException(status_code=400, detail='Log at least one completed exercise set.')
|
|
return exercises
|