214 lines
8.1 KiB
Python
214 lines
8.1 KiB
Python
from collections import defaultdict
|
|
from flask import Blueprint, render_template, redirect, request, url_for
|
|
import humanize
|
|
from pytz import timezone, utc
|
|
from sqlalchemy import func
|
|
from app.models import Reading, db
|
|
from app.forms import DeleteForm
|
|
from flask_login import login_required, current_user
|
|
from datetime import date, datetime, timedelta
|
|
|
|
main = Blueprint('main', __name__)
|
|
|
|
@main.route('/', methods=['GET'])
|
|
def landing():
|
|
return redirect(url_for('main.dashboard')) if current_user.is_authenticated else render_template('landing.html')
|
|
|
|
@main.route('/health')
|
|
def health():
|
|
return "OK", 200
|
|
|
|
@main.route('/dashboard', methods=['GET', 'POST'])
|
|
@login_required
|
|
def dashboard():
|
|
"""Render the dashboard with readings, stats, and calendar views."""
|
|
user_tz = timezone(current_user.profile.timezone or 'UTC')
|
|
|
|
# Get date range for readings
|
|
first_reading, last_reading = get_reading_date_range(current_user.id, user_tz)
|
|
start_date = request.form.get('start_date') or (first_reading and first_reading.strftime('%Y-%m-%d'))
|
|
end_date = request.form.get('end_date') or (last_reading and last_reading.strftime('%Y-%m-%d'))
|
|
|
|
# Fetch filtered readings
|
|
readings = fetch_readings(current_user.id, start_date, end_date, user_tz)
|
|
|
|
# Annotate readings with relative and localized timestamps
|
|
annotate_readings(readings, user_tz)
|
|
|
|
# Generate calendar view
|
|
week_view = generate_weekly_calendar(readings, datetime.now(user_tz), user_tz)
|
|
month_view = generate_monthly_calendar(readings, datetime.now(user_tz), user_tz)
|
|
|
|
# Calculate stats and badges
|
|
systolic_avg, diastolic_avg, heart_rate_avg = calculate_weekly_summary(readings)
|
|
badges = calculate_progress_badges(readings)
|
|
|
|
# Prepare graph data
|
|
graph_data = prepare_graph_data(readings)
|
|
|
|
return render_template(
|
|
'dashboard.html',
|
|
readings=readings,
|
|
profile=current_user.profile,
|
|
badges=badges,
|
|
systolic_avg=systolic_avg,
|
|
diastolic_avg=diastolic_avg,
|
|
heart_rate_avg=heart_rate_avg,
|
|
delete_form=DeleteForm(),
|
|
start_date=start_date,
|
|
end_date=end_date,
|
|
month=month_view,
|
|
week = week_view,
|
|
date=date,
|
|
**graph_data
|
|
)
|
|
|
|
def get_reading_date_range(user_id, user_tz):
|
|
"""Fetch the earliest and latest reading timestamps for a user."""
|
|
result = db.session.query(
|
|
func.min(Reading.timestamp).label('first'),
|
|
func.max(Reading.timestamp).label('last')
|
|
).filter(Reading.user_id == user_id).first()
|
|
|
|
first = utc.localize(result.first).astimezone(user_tz) if result.first else None
|
|
last = utc.localize(result.last).astimezone(user_tz) if result.last else None
|
|
return first, last
|
|
|
|
|
|
def fetch_readings(user_id, start_date, end_date, user_tz):
|
|
"""Retrieve readings filtered by date range."""
|
|
query = Reading.query.filter_by(user_id=user_id)
|
|
|
|
if start_date and end_date:
|
|
# Convert dates to the user's timezone
|
|
start_dt = user_tz.localize(datetime.strptime(start_date, '%Y-%m-%d')).astimezone(utc)
|
|
end_dt = user_tz.localize(datetime.strptime(end_date, '%Y-%m-%d')).astimezone(utc) + timedelta(days=1) - timedelta(seconds=1)
|
|
|
|
query = query.filter(
|
|
Reading.timestamp >= start_dt,
|
|
Reading.timestamp <= end_dt
|
|
)
|
|
|
|
return query.order_by(Reading.timestamp.desc()).all()
|
|
|
|
def annotate_readings(readings, user_tz):
|
|
"""Add relative and localized timestamps to readings."""
|
|
now = datetime.utcnow()
|
|
for reading in readings:
|
|
reading.relative_timestamp = humanize.naturaltime(now - reading.timestamp)
|
|
reading.local_timestamp = utc.localize(reading.timestamp).astimezone(user_tz)
|
|
|
|
def calculate_weekly_summary(readings):
|
|
"""Calculate averages for the past week."""
|
|
one_week_ago = datetime.utcnow() - timedelta(days=7)
|
|
weekly_readings = [r for r in readings if r.timestamp >= one_week_ago]
|
|
if weekly_readings:
|
|
return (
|
|
round(sum(r.systolic for r in weekly_readings) / len(weekly_readings), 1),
|
|
round(sum(r.diastolic for r in weekly_readings) / len(weekly_readings), 1),
|
|
round(sum(r.heart_rate for r in weekly_readings) / len(weekly_readings), 1),
|
|
)
|
|
return 0, 0, 0
|
|
|
|
def generate_monthly_calendar(readings, selected_date, local_tz):
|
|
"""Generate a monthly calendar view."""
|
|
today = datetime.now(local_tz).date()
|
|
first_day = selected_date.replace(day=1)
|
|
start_date = first_day - timedelta(days=(first_day.weekday() + 1) % 7)
|
|
end_date = start_date + timedelta(days=41)
|
|
|
|
readings_by_day = defaultdict(list)
|
|
for reading in readings:
|
|
local_date = reading.timestamp.astimezone(local_tz).date()
|
|
readings_by_day[local_date].append(reading)
|
|
|
|
return [
|
|
{
|
|
'day': current_date.day,
|
|
'is_today': current_date == today,
|
|
'is_in_current_month': current_date.month == selected_date.month,
|
|
'readings': readings_by_day.get(current_date.date(), []),
|
|
}
|
|
for current_date in (start_date + timedelta(days=i) for i in range((end_date - start_date).days + 1))
|
|
]
|
|
|
|
def generate_weekly_calendar(readings, selected_date, local_tz):
|
|
"""Generate a weekly calendar view."""
|
|
# Get the start of the week (Monday) and the end of the week (Sunday)
|
|
today = datetime.now(local_tz).date()
|
|
start_of_week = selected_date - timedelta(days=selected_date.weekday())
|
|
end_of_week = start_of_week + timedelta(days=6)
|
|
|
|
# Group readings by day
|
|
readings_by_day = defaultdict(list)
|
|
for reading in readings:
|
|
local_date = reading.timestamp.astimezone(local_tz).date()
|
|
readings_by_day[local_date].append(reading)
|
|
|
|
# Build the weekly calendar view
|
|
return [
|
|
{
|
|
'date': current_date.strftime('%a, %b %d') ,
|
|
'is_today': current_date == today,
|
|
'readings': readings_by_day.get(current_date.date(), []),
|
|
}
|
|
for current_date in (start_of_week + timedelta(days=i) for i in range(7))
|
|
]
|
|
|
|
def prepare_graph_data(readings):
|
|
"""Prepare data for graph rendering."""
|
|
return {
|
|
'timestamps': [r.timestamp.strftime('%b %d') for r in readings],
|
|
'systolic': [r.systolic for r in readings],
|
|
'diastolic': [r.diastolic for r in readings],
|
|
'heart_rate': [r.heart_rate for r in readings],
|
|
}
|
|
|
|
def calculate_progress_badges(readings):
|
|
"""Generate badges based on user activity and milestones."""
|
|
now = datetime.utcnow().date()
|
|
streak_count, badges = 1, []
|
|
|
|
if not readings:
|
|
return badges
|
|
|
|
sorted_readings = sorted(readings, key=lambda r: r.timestamp.date())
|
|
streak_count = 1
|
|
daily_streak = True
|
|
|
|
# Start with the first reading
|
|
previous_date = sorted_readings[0].timestamp.date()
|
|
|
|
for reading in sorted_readings[1:]:
|
|
current_date = reading.timestamp.date()
|
|
|
|
# Check for consecutive daily streaks
|
|
if (current_date - previous_date).days == 1:
|
|
streak_count += 1
|
|
elif (current_date - previous_date).days > 1:
|
|
daily_streak = False
|
|
|
|
previous_date = current_date
|
|
|
|
# Add streak badges
|
|
if daily_streak and streak_count >= 1:
|
|
badges.append(f"Current Streak: {streak_count} Days")
|
|
if daily_streak and streak_count >= 7:
|
|
badges.append("Logged Every Day for a Week")
|
|
|
|
if daily_streak and streak_count >= 30 and previous_date == now:
|
|
badges.append("Monthly Streak")
|
|
|
|
if all(5 <= r.timestamp.hour < 12 for r in sorted_readings[-7:]):
|
|
badges.append("Morning Riser: Logged Readings Every Morning for a Week")
|
|
|
|
if all(18 <= r.timestamp.hour <= 23 for r in sorted_readings[-7:]):
|
|
badges.append("Night Owl: Logged Readings Every Night for a Week")
|
|
|
|
milestones = [10, 50, 100, 500, 1000, 5000, 10000]
|
|
highest_milestone = max((m for m in milestones if len(readings) >= m), default=None)
|
|
if highest_milestone:
|
|
badges.append(f"{highest_milestone} Readings Logged")
|
|
|
|
return badges
|