initial commit

This commit is contained in:
parovin committed 2026-10-11 14:12:10 +03:00
commit 39fdd43d8e
127 files changed
+13919

No files matched your search

+589
View File
@@ -0,0 +1,589 @@
from __future__ import annotations
from datetime import date, datetime, timedelta
from typing import Optional
from fastapi import HTTPException
from sqlalchemy.orm import Session, joinedload
from app.config import settings
from app.models import AccessoryState, Profile, Workout, WorkoutExercise, WorkoutSet
from app.schemas import (
ExerciseResponse,
ProfileResponse,
SetResponse,
WorkoutDetail,
WorkoutSummary,
)
from app.services.schedule import apply_accessory_weight_to_future, generate_schedule
from app.services.weights import round_exercise_weight, round_to_plate, weight_step_for
def profile_response(profile: Profile, now: Optional[datetime] = None) -> ProfileResponse:
now = now or datetime.utcnow()
rest_countdown = None
can_probe = False
if profile.phase == "rest" and profile.probe_available_at:
delta = (profile.probe_available_at - now).total_seconds()
if delta > 0:
rest_countdown = int(delta)
else:
can_probe = True
elif profile.phase == "probe":
can_probe = True
return ProfileResponse(
one_rm_kg=profile.one_rm_kg,
training_weekdays=profile.training_weekdays,
start_date=profile.start_date,
onboarded=profile.onboarded,
phase=profile.phase if not (profile.phase == "rest" and can_probe) else "probe",
rest_until=profile.rest_until,
probe_available_at=profile.probe_available_at,
cycle_number=profile.cycle_number,
rest_countdown_seconds=rest_countdown,
can_probe=can_probe,
)
def set_response(s: WorkoutSet) -> SetResponse:
return SetResponse(
id=s.id,
set_index=s.set_index,
target_percent=s.target_percent,
target_reps=s.target_reps,
target_weight_kg=s.target_weight_kg,
is_failure=s.is_failure,
completed=s.completed,
actual_reps=s.actual_reps,
actual_weight_kg=s.actual_weight_kg,
)
def exercise_response(ex: WorkoutExercise) -> ExerciseResponse:
return ExerciseResponse(
id=ex.id,
order_index=ex.order_index,
name=ex.name,
kind=ex.kind,
prescribed_weight_kg=ex.prescribed_weight_kg,
weight_locked=ex.weight_locked,
needs_weight_input=ex.needs_weight_input,
pending_prompt=ex.pending_prompt,
weight_step_kg=weight_step_for(ex.name, kind=ex.kind),
sets=[set_response(s) for s in sorted(ex.sets, key=lambda x: x.set_index)],
)
def workout_summary(w: Workout) -> WorkoutSummary:
return WorkoutSummary(
id=w.id,
cycle_number=w.cycle_number,
week_number=w.week_number,
day_index=w.day_index,
scheduled_date=w.scheduled_date,
status=w.status,
exercise_count=len(w.exercises) if w.exercises is not None else 0,
)
def maybe_expire_pause(db: Session, workout: Workout, now: Optional[datetime] = None) -> Workout:
now = now or datetime.utcnow()
if workout.status == "paused" and workout.paused_at:
limit = workout.paused_at + timedelta(minutes=settings.pause_limit_minutes)
if now > limit:
workout.status = "abandoned"
workout.paused_at = None
workout.rest_started_at = None
db.commit()
db.refresh(workout)
return workout
def workout_detail(db: Session, workout: Workout, today: Optional[date] = None) -> WorkoutDetail:
today = today or date.today()
now = datetime.utcnow()
workout = maybe_expire_pause(db, workout, now)
active = (
db.query(Workout)
.filter(
Workout.cycle_number == workout.cycle_number,
Workout.id != workout.id,
Workout.status.in_(["in_progress", "paused"]),
)
.first()
)
can_start = workout.status == "planned" and workout.scheduled_date == today and active is None
can_resume = False
pause_expires_at = None
if workout.status == "paused" and workout.paused_at:
pause_expires_at = workout.paused_at + timedelta(minutes=settings.pause_limit_minutes)
can_resume = now <= pause_expires_at and workout.scheduled_date == today
rest_elapsed = 0
if workout.status == "in_progress" and workout.rest_started_at:
rest_elapsed = max(0, int((now - workout.rest_started_at).total_seconds()))
return WorkoutDetail(
id=workout.id,
cycle_number=workout.cycle_number,
week_number=workout.week_number,
day_index=workout.day_index,
scheduled_date=workout.scheduled_date,
status=workout.status,
started_at=workout.started_at,
paused_at=workout.paused_at,
completed_at=workout.completed_at,
current_exercise_index=workout.current_exercise_index,
current_set_index=workout.current_set_index,
rest_started_at=workout.rest_started_at,
rest_elapsed_seconds=rest_elapsed,
can_start=can_start,
can_resume=can_resume,
pause_expires_at=pause_expires_at,
exercises=[exercise_response(e) for e in sorted(workout.exercises, key=lambda x: x.order_index)],
)
def load_workout(db: Session, workout_id: int) -> Workout:
def _fetch() -> Workout | None:
return (
db.query(Workout)
.options(joinedload(Workout.exercises).joinedload(WorkoutExercise.sets))
.filter(Workout.id == workout_id)
.first()
)
workout = _fetch()
if not workout:
raise HTTPException(404, "Тренировка не найдена")
if sync_accessory_weights(db, workout):
db.commit()
workout = _fetch()
assert workout is not None
return workout
def sync_accessory_weights(db: Session, workout: Workout) -> bool:
"""Подтянуть сохранённый вес каждого вспомогательного из AccessoryState."""
changed = False
for ex in workout.exercises:
if ex.kind != "accessory":
continue
if any(s.completed for s in ex.sets):
continue
state = (
db.query(AccessoryState)
.filter(
AccessoryState.cycle_number == workout.cycle_number,
AccessoryState.exercise_name == ex.name,
)
.first()
)
if not state:
continue
weight = state.pending_weight_kg if state.pending_weight_kg is not None else state.weight_kg
if weight is None:
continue
if (
ex.prescribed_weight_kg != weight
or ex.needs_weight_input
or ex.weight_locked != bool(state.locked)
):
ex.prescribed_weight_kg = weight
ex.needs_weight_input = False
ex.weight_locked = bool(state.locked)
for s in ex.sets:
if not s.completed:
s.target_weight_kg = weight
changed = True
return changed
def start_workout(db: Session, workout: Workout, today: Optional[date] = None) -> Workout:
today = today or date.today()
if workout.scheduled_date != today:
raise HTTPException(400, "Начать можно только сегодняшнюю тренировку")
if workout.status not in ("planned",):
raise HTTPException(400, "Тренировку нельзя начать")
active = (
db.query(Workout)
.filter(
Workout.cycle_number == workout.cycle_number,
Workout.status.in_(["in_progress", "paused"]),
)
.first()
)
if active:
raise HTTPException(400, "Уже есть активная тренировка")
workout.status = "in_progress"
workout.started_at = datetime.utcnow()
workout.current_exercise_index = 0
workout.current_set_index = 0
db.commit()
db.refresh(workout)
return load_workout(db, workout.id)
def pause_workout(db: Session, workout: Workout) -> Workout:
if workout.status != "in_progress":
raise HTTPException(400, "Пауза доступна только во время тренировки")
workout.status = "paused"
workout.paused_at = datetime.utcnow()
workout.rest_started_at = None
db.commit()
return load_workout(db, workout.id)
def resume_workout(db: Session, workout: Workout, today: Optional[date] = None) -> Workout:
today = today or date.today()
workout = maybe_expire_pause(db, workout)
if workout.status == "abandoned":
raise HTTPException(400, "Тренировка остановлена: пауза дольше 15 минут")
if workout.status != "paused":
raise HTTPException(400, "Тренировка не на паузе")
if workout.scheduled_date != today:
raise HTTPException(400, "Продолжить можно только в тот же день")
assert workout.paused_at
if datetime.utcnow() > workout.paused_at + timedelta(minutes=settings.pause_limit_minutes):
workout.status = "abandoned"
db.commit()
raise HTTPException(400, "Тренировка остановлена: пауза дольше 15 минут")
workout.status = "in_progress"
workout.paused_at = None
db.commit()
return load_workout(db, workout.id)
def set_exercise_weight(db: Session, workout: Workout, exercise: WorkoutExercise, weight_kg: float) -> Workout:
if exercise.kind != "accessory":
raise HTTPException(400, "Вес задаётся только для вспомогательных")
if workout.status == "completed":
raise HTTPException(400, "Выполненную тренировку менять нельзя")
if exercise.sets and all(s.completed for s in exercise.sets):
raise HTTPException(400, "Это упражнение уже выполнено")
weight = round_exercise_weight(weight_kg, exercise.name, kind=exercise.kind)
exercise.prescribed_weight_kg = weight
exercise.needs_weight_input = False
for s in exercise.sets:
if not s.completed:
s.target_weight_kg = weight
apply_accessory_weight_to_future(
db,
cycle_number=workout.cycle_number,
exercise_name=exercise.name,
weight_kg=weight,
from_date=workout.scheduled_date,
lock=False,
)
db.commit()
return load_workout(db, workout.id)
def adjust_completed_set(
db: Session,
workout: Workout,
exercise: WorkoutExercise,
workout_set: WorkoutSet,
*,
actual_weight_kg: float,
actual_reps: int,
) -> Workout:
if exercise.kind != "accessory":
raise HTTPException(400, "Править можно только вспомогательные")
if not workout_set.completed:
raise HTTPException(400, "Можно править только выполненный подход")
workout_set.actual_weight_kg = round_exercise_weight(actual_weight_kg, exercise.name, kind=exercise.kind)
workout_set.actual_reps = actual_reps
db.commit()
return load_workout(db, workout.id)
def _advance_pointer(workout: Workout, exercises: list[WorkoutExercise]) -> bool:
"""Advance to next incomplete set. Returns True if workout completed."""
for ei, ex in enumerate(exercises):
for s in sorted(ex.sets, key=lambda x: x.set_index):
if not s.completed:
workout.current_exercise_index = ei
workout.current_set_index = s.set_index
return False
return True
def _analyze_accessory(db: Session, workout: Workout, exercise: WorkoutExercise) -> None:
if exercise.kind != "accessory":
return
if exercise.weight_locked:
return
if any(not s.completed for s in exercise.sets):
return
all_hit = all(
(s.actual_reps or 0) >= (s.target_reps or 0) for s in exercise.sets if s.target_reps is not None
)
if not all_hit:
exercise.pending_prompt = "choose_weight_missed"
else:
exercise.pending_prompt = "confirm_easy"
def complete_set(
db: Session,
workout: Workout,
*,
actual_reps: Optional[int] = None,
actual_weight_kg: Optional[float] = None,
) -> Workout:
if workout.status != "in_progress":
raise HTTPException(400, "Тренировка не активна")
exercises = sorted(workout.exercises, key=lambda x: x.order_index)
if workout.current_exercise_index >= len(exercises):
raise HTTPException(400, "Нет текущего упражнения")
exercise = exercises[workout.current_exercise_index]
if exercise.pending_prompt:
raise HTTPException(400, "Сначала ответьте на вопрос по упражнению")
if exercise.kind == "accessory" and exercise.needs_weight_input:
raise HTTPException(400, "Сначала введите рабочий вес")
sets = sorted(exercise.sets, key=lambda x: x.set_index)
current = next((s for s in sets if s.set_index == workout.current_set_index), None)
if not current or current.completed:
# find first incomplete
current = next((s for s in sets if not s.completed), None)
if not current:
raise HTTPException(400, "Нет активного подхода")
if exercise.kind == "bench" or exercise.kind == "failure":
current.completed = True
current.completed_at = datetime.utcnow()
current.actual_weight_kg = current.target_weight_kg
current.actual_reps = current.target_reps
else:
if actual_reps is None:
raise HTTPException(400, "Укажите выполненные повторения")
weight = actual_weight_kg if actual_weight_kg is not None else exercise.prescribed_weight_kg
if weight is None:
raise HTTPException(400, "Не задан вес")
current.completed = True
current.completed_at = datetime.utcnow()
current.actual_reps = actual_reps
current.actual_weight_kg = round_exercise_weight(weight, exercise.name, kind=exercise.kind)
# Check if exercise finished → maybe prompt
if all(s.completed for s in sets):
_analyze_accessory(db, workout, exercise)
if exercise.pending_prompt:
workout.rest_started_at = None
db.commit()
return load_workout(db, workout.id)
done = _advance_pointer(workout, exercises)
if done:
workout.status = "completed"
workout.completed_at = datetime.utcnow()
workout.rest_started_at = None
_maybe_finish_cycle(db, workout)
else:
workout.rest_started_at = datetime.utcnow()
db.commit()
return load_workout(db, workout.id)
def resolve_prompt(
db: Session,
workout: Workout,
exercise: WorkoutExercise,
*,
choice: str,
weight_kg: Optional[float] = None,
increase_kg: Optional[float] = None,
decrease_kg: Optional[float] = None,
) -> Workout:
if workout.status != "in_progress":
raise HTTPException(400, "Тренировка не активна")
if not exercise.pending_prompt:
raise HTTPException(400, "Нет активного вопроса")
prompt = exercise.pending_prompt
today = workout.scheduled_date
current_weight = exercise.prescribed_weight_kg
step = weight_step_for(exercise.name, kind=exercise.kind)
if prompt == "confirm_easy":
if choice == "easy_no":
if current_weight is None:
raise HTTPException(400, "Вес не задан")
apply_accessory_weight_to_future(
db,
cycle_number=workout.cycle_number,
exercise_name=exercise.name,
weight_kg=current_weight,
from_date=today,
lock=True,
)
exercise.weight_locked = True
exercise.pending_prompt = None
elif choice == "easy_yes":
exercise.pending_prompt = "choose_weight_easy"
db.commit()
return load_workout(db, workout.id)
else:
raise HTTPException(400, "Неверный выбор")
elif prompt == "choose_weight_easy":
if current_weight is None:
raise HTTPException(400, "Текущий вес не задан")
if increase_kg is None or increase_kg <= 0:
raise HTTPException(400, "Укажите, на сколько поднять вес")
bump = round_to_plate(increase_kg, step)
if bump < step:
bump = step
weight = round_to_plate(current_weight + bump, step)
if weight <= current_weight:
weight = round_to_plate(current_weight + step, step)
apply_accessory_weight_to_future(
db,
cycle_number=workout.cycle_number,
exercise_name=exercise.name,
weight_kg=weight,
from_date=today + timedelta(days=1),
lock=False,
)
exercise.pending_prompt = None
elif prompt == "choose_weight_missed":
if current_weight is None:
raise HTTPException(400, "Текущий вес не задан")
if decrease_kg is None or decrease_kg <= 0:
raise HTTPException(400, "Укажите, на сколько снизить вес")
drop = round_to_plate(decrease_kg, step)
if drop < step:
drop = step
weight = round_to_plate(current_weight - drop, step)
if weight >= current_weight:
weight = round_to_plate(current_weight - step, step)
if weight < step:
weight = step
apply_accessory_weight_to_future(
db,
cycle_number=workout.cycle_number,
exercise_name=exercise.name,
weight_kg=weight,
from_date=today + timedelta(days=1),
lock=False,
)
exercise.pending_prompt = None
else:
raise HTTPException(400, "Неизвестный вопрос")
exercises = sorted(workout.exercises, key=lambda x: x.order_index)
done = _advance_pointer(workout, exercises)
if done:
workout.status = "completed"
workout.completed_at = datetime.utcnow()
workout.rest_started_at = None
_maybe_finish_cycle(db, workout)
else:
workout.rest_started_at = datetime.utcnow()
db.commit()
return load_workout(db, workout.id)
def _maybe_finish_cycle(db: Session, workout: Workout) -> None:
remaining = (
db.query(Workout)
.filter(
Workout.cycle_number == workout.cycle_number,
Workout.status.in_(["planned", "in_progress", "paused"]),
Workout.id != workout.id,
)
.count()
)
# also count if this one not yet marked - it's being completed
unfinished_others = (
db.query(Workout)
.filter(
Workout.cycle_number == workout.cycle_number,
Workout.id != workout.id,
Workout.status.notin_(["completed", "abandoned"]),
)
.count()
)
if unfinished_others == 0:
profile = db.query(Profile).first()
if profile:
profile.phase = "rest"
# wait for user to choose rest days via API; temporarily set probe far
profile.probe_available_at = None
profile.rest_until = None
def start_rest_phase(db: Session, profile: Profile, rest_days: int) -> Profile:
if profile.phase not in ("rest", "probe"):
# allow if cycle just finished
unfinished = (
db.query(Workout)
.filter(
Workout.cycle_number == profile.cycle_number,
Workout.status.notin_(["completed", "abandoned"]),
)
.count()
)
if unfinished > 0 and profile.phase == "active":
raise HTTPException(400, "Цикл ещё не завершён")
if rest_days < settings.rest_min_days or rest_days > settings.rest_max_days:
raise HTTPException(400, f"Отдых от {settings.rest_min_days} до {settings.rest_max_days} дней")
now = datetime.utcnow()
profile.phase = "rest"
profile.probe_available_at = now + timedelta(days=rest_days)
profile.rest_until = profile.probe_available_at
db.commit()
db.refresh(profile)
return profile
def submit_probe(
db: Session,
profile: Profile,
*,
one_rm_kg: float,
start_date: date,
training_weekdays: Optional[list[int]] = None,
) -> Profile:
now = datetime.utcnow()
if profile.phase == "rest" and profile.probe_available_at and now < profile.probe_available_at:
raise HTTPException(400, "Ещё рано для проходки")
profile.phase = "probe" if profile.phase == "rest" else profile.phase
weekdays = training_weekdays or profile.training_weekdays
if not weekdays:
raise HTTPException(400, "Не заданы дни тренировок")
one_rm = round_to_plate(one_rm_kg)
new_cycle = profile.cycle_number + 1
profile.one_rm_kg = one_rm
profile.training_weekdays = weekdays
profile.start_date = start_date
profile.cycle_number = new_cycle
profile.phase = "active"
profile.rest_until = None
profile.probe_available_at = None
profile.onboarded = True
generate_schedule(
db,
one_rm_kg=one_rm,
training_weekdays=weekdays,
start_date=start_date,
cycle_number=new_cycle,
)
db.commit()
db.refresh(profile)
return profile