Files
truenas-stacks/home/doris-barbell/app/services/store.py
Fizzlepoof 1227814277
Some checks failed
secret-guardrails / artifact-secret-scan (push) Has been cancelled
secret-guardrails / gitleaks (push) Has been cancelled
Add Doris Barbell live workout logger
2026-05-28 21:32:09 +00:00

614 lines
26 KiB
Python

from __future__ import annotations
import json
from copy import deepcopy
from pathlib import Path
from typing import Any
from uuid import uuid4
DEFAULT_USERS = {
'john': {
'id': 'john',
'display_name': 'John',
'is_stub': False,
'measurement_system': 'imperial',
'height_inches': 68.5,
'goal_weight_lbs': None,
'notes': 'Seeded with John default height assumption from current intake.',
},
'manndra': {
'id': 'manndra',
'display_name': 'Manndra',
'is_stub': True,
'measurement_system': 'imperial',
'height_inches': None,
'goal_weight_lbs': None,
'notes': 'Stubbed for future shared use.',
},
}
SEEDED_EXERCISE_TEMPLATES = [
{'name': 'Bench Press', 'primary_muscle': 'chest', 'equipment': 'barbell'},
{'name': 'Incline Bench Press', 'primary_muscle': 'upper chest', 'equipment': 'barbell'},
{'name': 'Decline Bench Press', 'primary_muscle': 'lower chest', 'equipment': 'barbell'},
{'name': 'Butterflies', 'primary_muscle': 'chest', 'equipment': 'machine'},
{'name': 'Lat Pulldowns', 'primary_muscle': 'back', 'equipment': 'cable'},
{'name': 'Bent-Over Supported Rows', 'primary_muscle': 'back', 'equipment': 'machine'},
{'name': 'Landmine Rows', 'primary_muscle': 'back', 'equipment': 'barbell'},
{'name': 'Seated Machine Rows', 'primary_muscle': 'back', 'equipment': 'machine'},
{'name': 'Resistance Band Ab Holds', 'primary_muscle': 'abs', 'equipment': 'band'},
{'name': 'Lat Pulldown Drop Set', 'primary_muscle': 'back', 'equipment': 'cable'},
{'name': 'Bicep Curls', 'primary_muscle': 'biceps', 'equipment': 'dumbbells'},
{'name': 'Front Tricep Extensions', 'primary_muscle': 'triceps', 'equipment': 'cable'},
{'name': 'Overhead Tricep Extensions', 'primary_muscle': 'triceps', 'equipment': 'cable'},
{'name': 'Incline Barbell Curls', 'primary_muscle': 'biceps', 'equipment': 'barbell'},
{'name': 'Isolation Bicep Curls', 'primary_muscle': 'biceps', 'equipment': 'machine'},
{'name': '7-7-7 Barbell Curl', 'primary_muscle': 'biceps', 'equipment': 'barbell'},
{'name': 'Suitcase Carry', 'primary_muscle': 'abs', 'equipment': 'dumbbell'},
{'name': 'Shoulder Press', 'primary_muscle': 'shoulders', 'equipment': 'machine'},
{'name': 'Arnold Press', 'primary_muscle': 'shoulders', 'equipment': 'dumbbells'},
{'name': 'Side Lateral Raise', 'primary_muscle': 'shoulders', 'equipment': 'dumbbells'},
{'name': 'Front Lateral Raise', 'primary_muscle': 'shoulders', 'equipment': 'dumbbells'},
{'name': 'Upright Row', 'primary_muscle': 'shoulders', 'equipment': 'barbell'},
{'name': 'Shoulder Shrugs', 'primary_muscle': 'traps', 'equipment': 'dumbbells'},
{'name': '7-7-7 Shoulder Press', 'primary_muscle': 'shoulders', 'equipment': 'machine'},
{'name': 'Resistance Band Punch-Out Training', 'primary_muscle': 'shoulders', 'equipment': 'band'},
{'name': 'Squats', 'primary_muscle': 'legs', 'equipment': 'barbell'},
{'name': 'Leg Extensions', 'primary_muscle': 'quads', 'equipment': 'machine'},
{'name': 'Step Ups', 'primary_muscle': 'legs', 'equipment': 'bodyweight'},
{'name': 'Deadlifts', 'primary_muscle': 'posterior chain', 'equipment': 'barbell'},
{'name': 'Calf Raises', 'primary_muscle': 'calves', 'equipment': 'machine'},
{'name': 'Resistance Band Crunches', 'primary_muscle': 'abs', 'equipment': 'band'},
]
SEEDED_ROUTINES = [
{
'user_id': 'john',
'name': 'Monday: Chest',
'notes': "Imported from John's current routine. Default prescription is 3 sets of 10 reps. Weights are still TBD.",
'exercises': [
{'exercise_name': 'Bench Press', 'target_sets': 3, 'target_reps': '10'},
{'exercise_name': 'Incline Bench Press', 'target_sets': 3, 'target_reps': '10'},
{'exercise_name': 'Decline Bench Press', 'target_sets': 3, 'target_reps': '10'},
{'exercise_name': 'Butterflies', 'target_sets': 3, 'target_reps': '10'},
],
},
{
'user_id': 'john',
'name': 'Tuesday: Back',
'notes': "Imported from John's current routine. Default prescription is 3 sets of 10 reps. Weights are still TBD.",
'exercises': [
{'exercise_name': 'Lat Pulldowns', 'target_sets': 3, 'target_reps': '10'},
{'exercise_name': 'Bent-Over Supported Rows', 'target_sets': 3, 'target_reps': '10'},
{'exercise_name': 'Landmine Rows', 'target_sets': 3, 'target_reps': '10'},
{'exercise_name': 'Seated Machine Rows', 'target_sets': 3, 'target_reps': '10'},
{'exercise_name': 'Resistance Band Ab Holds', 'target_sets': 3, 'target_reps': '10'},
{'exercise_name': 'Lat Pulldown Drop Set', 'target_sets': 3, 'target_reps': '10'},
],
},
{
'user_id': 'john',
'name': 'Wednesday: Arms',
'notes': "Imported from John's current routine. Default prescription is 3 sets of 10 reps. Weights are still TBD. Includes Bis and Tris focus plus suitcase carries for abs.",
'exercises': [
{'exercise_name': 'Bicep Curls', 'target_sets': 3, 'target_reps': '10'},
{'exercise_name': 'Front Tricep Extensions', 'target_sets': 3, 'target_reps': '10'},
{'exercise_name': 'Overhead Tricep Extensions', 'target_sets': 3, 'target_reps': '10'},
{'exercise_name': 'Incline Barbell Curls', 'target_sets': 3, 'target_reps': '10'},
{'exercise_name': 'Isolation Bicep Curls', 'target_sets': 3, 'target_reps': '10'},
{'exercise_name': '7-7-7 Barbell Curl', 'target_sets': 1, 'target_reps': '7-7-7'},
{'exercise_name': 'Suitcase Carry', 'target_sets': 3, 'target_reps': '10'},
],
},
{
'user_id': 'john',
'name': 'Thursday: Shoulders',
'notes': "Imported from John's current routine. Default prescription is 3 sets of 10 reps. Weights are still TBD. Includes resistance band punch-out training.",
'exercises': [
{'exercise_name': 'Shoulder Press', 'target_sets': 3, 'target_reps': '10'},
{'exercise_name': 'Arnold Press', 'target_sets': 3, 'target_reps': '10'},
{'exercise_name': 'Side Lateral Raise', 'target_sets': 3, 'target_reps': '10'},
{'exercise_name': 'Front Lateral Raise', 'target_sets': 3, 'target_reps': '10'},
{'exercise_name': 'Upright Row', 'target_sets': 3, 'target_reps': '10'},
{'exercise_name': 'Shoulder Shrugs', 'target_sets': 3, 'target_reps': '10'},
{'exercise_name': '7-7-7 Shoulder Press', 'target_sets': 1, 'target_reps': '7-7-7'},
{'exercise_name': 'Resistance Band Punch-Out Training', 'target_sets': 3, 'target_reps': '10'},
],
},
{
'user_id': 'john',
'name': 'Friday: Legs',
'notes': "Imported from John's current routine. Default prescription is 3 sets of 10 reps. Weights are still TBD. Includes resistance band crunches for abs.",
'exercises': [
{'exercise_name': 'Squats', 'target_sets': 3, 'target_reps': '10'},
{'exercise_name': 'Leg Extensions', 'target_sets': 3, 'target_reps': '10'},
{'exercise_name': 'Step Ups', 'target_sets': 3, 'target_reps': '10'},
{'exercise_name': 'Deadlifts', 'target_sets': 3, 'target_reps': '10'},
{'exercise_name': 'Calf Raises', 'target_sets': 3, 'target_reps': '10'},
{'exercise_name': 'Resistance Band Crunches', 'target_sets': 3, 'target_reps': '10'},
],
},
]
DEFAULT_STATE = {
'users': DEFAULT_USERS,
'exercise_templates': [
{
'id': f'seed-template-{index + 1:02d}',
'name': template['name'],
'primary_muscle': template.get('primary_muscle'),
'equipment': template.get('equipment'),
'notes': template.get('notes'),
}
for index, template in enumerate(SEEDED_EXERCISE_TEMPLATES)
],
'routines': [
{
'id': f'seed-routine-{index + 1:02d}',
'user_id': routine['user_id'],
'name': routine['name'],
'notes': routine.get('notes'),
'exercises': [dict(exercise) for exercise in routine['exercises']],
}
for index, routine in enumerate(SEEDED_ROUTINES)
],
'weight_entries': [],
'measurement_entries': [],
'workout_sessions': [],
'active_workout_sessions': [],
}
LEGACY_TEMPLATE_ALIASES = {
'resistance band punch-out holds': 'Resistance Band Punch-Out Training',
}
class JSONStore:
def __init__(self, state_dir: Path) -> None:
self.state_dir = Path(state_dir)
self.state_path = self.state_dir / 'state.json'
self.state_dir.mkdir(parents=True, exist_ok=True)
if not self.state_path.exists():
self._write(deepcopy(DEFAULT_STATE))
return
state = self._read()
seeded_state = _ensure_seeded_library(state)
if seeded_state != state:
self._write(seeded_state)
def _read(self) -> dict[str, Any]:
payload = json.loads(self.state_path.read_text())
seeded_payload = _ensure_seeded_library(payload)
if seeded_payload != payload:
self._write(seeded_payload)
return seeded_payload
def _write(self, payload: dict[str, Any]) -> None:
self.state_path.write_text(json.dumps(payload, indent=2, sort_keys=True))
def update_profile(self, user_id: str, changes: dict[str, Any]) -> dict[str, Any]:
state = self._read()
user = state['users'].get(user_id)
if user is None:
raise KeyError(user_id)
for key, value in changes.items():
if value is not None:
user[key] = value
state['users'][user_id] = user
self._write(state)
return user
def add_weight_entry(self, user_id: str, entry: dict[str, Any]) -> dict[str, Any]:
state = self._read()
user = self._user_or_raise(state, user_id)
saved = {
'id': uuid4().hex,
'user_id': user_id,
'recorded_at': entry['recorded_at'],
'weight_lbs': entry['weight_lbs'],
'body_fat_percent': entry.get('body_fat_percent'),
'notes': entry.get('notes'),
'bmi': _calculate_bmi(entry['weight_lbs'], user.get('height_inches')),
}
state['weight_entries'].append(saved)
self._write(state)
return saved
def weight_history(self, user_id: str) -> list[dict[str, Any]]:
state = self._read()
self._user_or_raise(state, user_id)
return _history_for_user(state['weight_entries'], user_id, 'recorded_at')
def add_measurement_entry(self, user_id: str, entry: dict[str, Any]) -> dict[str, Any]:
state = self._read()
self._user_or_raise(state, user_id)
saved = {
'id': uuid4().hex,
'user_id': user_id,
'recorded_at': entry['recorded_at'],
'waist_inches': entry.get('waist_inches'),
'hip_inches': entry.get('hip_inches'),
'chest_inches': entry.get('chest_inches'),
'neck_inches': entry.get('neck_inches'),
'notes': entry.get('notes'),
}
state['measurement_entries'].append(saved)
self._write(state)
return saved
def measurement_history(self, user_id: str) -> list[dict[str, Any]]:
state = self._read()
self._user_or_raise(state, user_id)
return _history_for_user(state['measurement_entries'], user_id, 'recorded_at')
def add_exercise_template(self, payload: dict[str, Any]) -> dict[str, Any]:
state = self._read()
saved = {
'id': uuid4().hex,
'name': payload['name'],
'primary_muscle': payload.get('primary_muscle'),
'equipment': payload.get('equipment'),
'notes': payload.get('notes'),
}
state['exercise_templates'].append(saved)
self._write(state)
return saved
def list_exercise_templates(self) -> list[dict[str, Any]]:
state = self._read()
return sorted(state['exercise_templates'], key=lambda item: item['name'].lower())
def add_routine(self, payload: dict[str, Any]) -> dict[str, Any]:
state = self._read()
self._user_or_raise(state, payload['user_id'])
saved = {
'id': uuid4().hex,
'user_id': payload['user_id'],
'name': payload['name'],
'notes': payload.get('notes'),
'exercises': [dict(exercise) for exercise in payload['exercises']],
}
state['routines'].append(saved)
self._write(state)
return saved
def list_routines(self) -> list[dict[str, Any]]:
state = self._read()
return sorted(state['routines'], key=lambda item: item['name'].lower())
def get_routine(self, routine_id: str) -> dict[str, Any] | None:
state = self._read()
for routine in state['routines']:
if routine.get('id') == routine_id:
return routine
return None
def get_active_workout_session(self, routine_id: str, user_id: str) -> dict[str, Any] | None:
state = self._read()
self._user_or_raise(state, user_id)
for session in state['active_workout_sessions']:
if session['routine_id'] == routine_id and session['user_id'] == user_id:
return session
return None
def start_active_workout_session(self, payload: dict[str, Any]) -> dict[str, Any]:
state = self._read()
self._user_or_raise(state, payload['user_id'])
state['active_workout_sessions'] = [
session
for session in state['active_workout_sessions']
if not (
session['routine_id'] == payload['routine_id'] and session['user_id'] == payload['user_id']
)
]
saved = {
'id': uuid4().hex,
'routine_id': payload['routine_id'],
'routine_name': payload['routine_name'],
'user_id': payload['user_id'],
'performed_at': payload['performed_at'],
'notes': payload.get('notes'),
'exercises': [],
}
state['active_workout_sessions'].append(saved)
self._write(state)
return saved
def add_active_workout_exercise(self, routine_id: str, user_id: str, exercise: dict[str, Any]) -> dict[str, Any]:
state = self._read()
self._user_or_raise(state, user_id)
for session in state['active_workout_sessions']:
if session['routine_id'] == routine_id and session['user_id'] == user_id:
session['exercises'].append(
{
'exercise_name': exercise['exercise_name'],
'notes': exercise.get('notes'),
'sets': [dict(item) for item in exercise['sets']],
}
)
self._write(state)
return session
raise KeyError((routine_id, user_id))
def finish_active_workout_session(self, routine_id: str, user_id: str, notes: str | None = None) -> dict[str, Any]:
state = self._read()
self._user_or_raise(state, user_id)
for session in state['active_workout_sessions']:
if session['routine_id'] != routine_id or session['user_id'] != user_id:
continue
workout_payload = {
'user_id': user_id,
'performed_at': session['performed_at'],
'routine_name': session['routine_name'],
'notes': notes,
'exercises': [
{
'exercise_name': exercise['exercise_name'],
'notes': exercise.get('notes'),
'sets': [dict(item) for item in exercise['sets']],
}
for exercise in session['exercises']
],
}
saved_workout = self.add_workout(workout_payload)
refreshed_state = self._read()
refreshed_state['active_workout_sessions'] = [
active_session
for active_session in refreshed_state['active_workout_sessions']
if active_session.get('id') != session['id']
]
self._write(refreshed_state)
return saved_workout
raise KeyError((routine_id, user_id))
def add_workout(self, payload: dict[str, Any]) -> dict[str, Any]:
state = self._read()
self._user_or_raise(state, payload['user_id'])
set_count = 0
total_volume_lbs = 0.0
exercises: list[dict[str, Any]] = []
for exercise in payload['exercises']:
normalized_sets = []
for item in exercise['sets']:
set_count += 1
weight_lbs = item.get('weight_lbs')
if weight_lbs is not None:
total_volume_lbs += item['reps'] * weight_lbs
normalized_sets.append(dict(item))
exercises.append({
'exercise_name': exercise['exercise_name'],
'notes': exercise.get('notes'),
'sets': normalized_sets,
})
saved = {
'id': uuid4().hex,
'user_id': payload['user_id'],
'performed_at': payload['performed_at'],
'routine_name': payload['routine_name'],
'notes': payload.get('notes'),
'exercise_count': len(exercises),
'set_count': set_count,
'total_volume_lbs': int(total_volume_lbs) if total_volume_lbs.is_integer() else total_volume_lbs,
'exercises': exercises,
}
state['workout_sessions'].append(saved)
self._write(state)
return saved
def workout_history(self, user_id: str) -> list[dict[str, Any]]:
state = self._read()
self._user_or_raise(state, user_id)
return _history_for_user(state['workout_sessions'], user_id, 'performed_at')
def exercise_progression(self, user_id: str) -> list[dict[str, Any]]:
state = self._read()
self._user_or_raise(state, user_id)
progression: dict[str, dict[str, Any]] = {}
for workout in _history_for_user(state['workout_sessions'], user_id, 'performed_at'):
for exercise in workout['exercises']:
name = exercise['exercise_name']
stats = progression.setdefault(
name,
{
'exercise_name': name,
'sessions': 0,
'set_count': 0,
'max_weight_lbs': 0.0,
'best_set_estimated_1rm': 0.0,
'last_performed_at': workout['performed_at'],
},
)
stats['sessions'] += 1
if workout['performed_at'] > stats['last_performed_at']:
stats['last_performed_at'] = workout['performed_at']
for item in exercise['sets']:
stats['set_count'] += 1
weight_lbs = item.get('weight_lbs')
if weight_lbs is None:
continue
stats['max_weight_lbs'] = max(stats['max_weight_lbs'], weight_lbs)
stats['best_set_estimated_1rm'] = max(
stats['best_set_estimated_1rm'],
_estimated_1rm(weight_lbs, item['reps']),
)
results = []
for item in progression.values():
results.append(
{
**item,
'max_weight_lbs': _normalize_number(item['max_weight_lbs']),
'best_set_estimated_1rm': _normalize_number(item['best_set_estimated_1rm']),
}
)
return sorted(results, key=lambda item: item['exercise_name'].lower())
def dashboard_summary(self) -> dict[str, Any]:
state = self._read()
users = []
weight_entries = state['weight_entries']
measurement_entries = state['measurement_entries']
for user in state['users'].values():
latest_weight = _latest_entry_for_user(weight_entries, user['id'])
latest_measurements = _latest_entry_for_user(measurement_entries, user['id'])
users.append({
'id': user['id'],
'display_name': user['display_name'],
'is_stub': user['is_stub'],
'measurement_system': user.get('measurement_system', 'imperial'),
'height_inches': user['height_inches'],
'goal_weight_lbs': user['goal_weight_lbs'],
'current_weight_lbs': latest_weight['weight_lbs'] if latest_weight else None,
'current_bmi': latest_weight['bmi'] if latest_weight else None,
'last_weight_recorded_at': latest_weight['recorded_at'] if latest_weight else None,
'latest_measurements': _trim_measurement_entry(latest_measurements),
})
recent_workouts = sorted(
state['workout_sessions'],
key=lambda item: item['performed_at'],
reverse=True,
)[:5]
recent_templates = list(reversed(state['exercise_templates'][-5:]))
routine_board = _routine_board(state['routines'])
return {
'users': users,
'recent_workouts': recent_workouts,
'exercise_templates': recent_templates,
'routines': routine_board,
'counts': {
'exercise_templates': len(state['exercise_templates']),
'measurement_entries': len(state['measurement_entries']),
'routines': len(state['routines']),
'weight_entries': len(state['weight_entries']),
'workout_sessions': len(state['workout_sessions']),
},
}
def _user_or_raise(self, state: dict[str, Any], user_id: str) -> dict[str, Any]:
user = state['users'].get(user_id)
if user is None:
raise KeyError(user_id)
return user
def _latest_entry_for_user(entries: list[dict[str, Any]], user_id: str) -> dict[str, Any] | None:
matches = [entry for entry in entries if entry['user_id'] == user_id]
if not matches:
return None
return max(matches, key=lambda item: item['recorded_at'])
def _history_for_user(entries: list[dict[str, Any]], user_id: str, sort_key: str) -> list[dict[str, Any]]:
matches = [entry for entry in entries if entry['user_id'] == user_id]
return sorted(matches, key=lambda item: item[sort_key], reverse=True)
def _ensure_seeded_library(state: dict[str, Any]) -> dict[str, Any]:
payload = deepcopy(state)
payload.setdefault('users', deepcopy(DEFAULT_USERS))
payload.setdefault('exercise_templates', [])
payload.setdefault('routines', [])
payload.setdefault('weight_entries', [])
payload.setdefault('measurement_entries', [])
payload.setdefault('workout_sessions', [])
payload.setdefault('active_workout_sessions', [])
for template in payload['exercise_templates']:
legacy_name = template.get('name', '').strip().lower()
replacement_name = LEGACY_TEMPLATE_ALIASES.get(legacy_name)
if replacement_name is None:
continue
seeded_template = next(item for item in SEEDED_EXERCISE_TEMPLATES if item['name'] == replacement_name)
template['name'] = seeded_template['name']
template['primary_muscle'] = seeded_template.get('primary_muscle')
template['equipment'] = seeded_template.get('equipment')
template['notes'] = seeded_template.get('notes')
existing_template_names = {
item.get('name', '').strip().lower()
for item in payload['exercise_templates']
if item.get('name')
}
for index, template in enumerate(SEEDED_EXERCISE_TEMPLATES, start=1):
normalized_name = template['name'].strip().lower()
if normalized_name in existing_template_names:
continue
payload['exercise_templates'].append(
{
'id': f'seed-template-{index:02d}',
'name': template['name'],
'primary_muscle': template.get('primary_muscle'),
'equipment': template.get('equipment'),
'notes': template.get('notes'),
}
)
existing_template_names.add(normalized_name)
existing_routines = {
item.get('name', '').strip().lower(): item
for item in payload['routines']
if item.get('name')
}
for index, routine in enumerate(SEEDED_ROUTINES, start=1):
normalized_name = routine['name'].strip().lower()
existing_routine = existing_routines.get(normalized_name)
seeded_exercises = [dict(exercise) for exercise in routine['exercises']]
if existing_routine is not None:
existing_routine['user_id'] = routine['user_id']
existing_routine['notes'] = routine.get('notes')
existing_routine['exercises'] = seeded_exercises
continue
new_routine = {
'id': f'seed-routine-{index:02d}',
'user_id': routine['user_id'],
'name': routine['name'],
'notes': routine.get('notes'),
'exercises': seeded_exercises,
}
payload['routines'].append(new_routine)
existing_routines[normalized_name] = new_routine
return payload
def _routine_board(routines: list[dict[str, Any]]) -> list[dict[str, Any]]:
by_name = {
routine.get('name', '').strip().lower(): routine
for routine in routines
if routine.get('name')
}
ordered = []
for seeded_routine in SEEDED_ROUTINES:
matched = by_name.get(seeded_routine['name'].strip().lower())
if matched is not None:
ordered.append(matched)
return ordered
def _trim_measurement_entry(entry: dict[str, Any] | None) -> dict[str, Any] | None:
if entry is None:
return None
return {
'recorded_at': entry['recorded_at'],
'waist_inches': entry['waist_inches'],
'hip_inches': entry['hip_inches'],
'chest_inches': entry['chest_inches'],
'neck_inches': entry['neck_inches'],
}
def _normalize_number(value: float) -> int | float:
return int(value) if float(value).is_integer() else round(value, 1)
def _estimated_1rm(weight_lbs: float, reps: int) -> float:
return weight_lbs * (1 + reps / 30)
def _calculate_bmi(weight_lbs: float, height_inches: float | None) -> float | None:
if not height_inches:
return None
return 703 * weight_lbs / (height_inches * height_inches)