feat: add Doris Barbell v1 scaffold
This commit is contained in:
Fizzlepoof
302
home/doris-barbell/app/services/store.py
Normal file
302
home/doris-barbell/app/services/store.py
Normal file
@@ -0,0 +1,302 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from copy import deepcopy
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
from uuid import uuid4
|
||||
|
||||
DEFAULT_STATE = {
|
||||
'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.',
|
||||
},
|
||||
},
|
||||
'exercise_templates': [],
|
||||
'routines': [],
|
||||
'weight_entries': [],
|
||||
'measurement_entries': [],
|
||||
'workout_sessions': [],
|
||||
}
|
||||
|
||||
|
||||
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))
|
||||
|
||||
def _read(self) -> dict[str, Any]:
|
||||
return json.loads(self.state_path.read_text())
|
||||
|
||||
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 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
|
||||
total_volume_lbs += item['reps'] * item['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
|
||||
stats['max_weight_lbs'] = max(stats['max_weight_lbs'], item['weight_lbs'])
|
||||
stats['best_set_estimated_1rm'] = max(
|
||||
stats['best_set_estimated_1rm'],
|
||||
_estimated_1rm(item['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]
|
||||
return {
|
||||
'users': users,
|
||||
'recent_workouts': recent_workouts,
|
||||
'exercise_templates': self.list_exercise_templates()[:5],
|
||||
'routines': self.list_routines()[:5],
|
||||
'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 _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)
|
||||
Reference in New Issue
Block a user