peterstockings/bloodpressure
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Redesign BP tracker with lightweight views and unobtrusive save confirmations

Browse Source
This commit is contained in:
Peter Stockings
2026-09-08 15:07:57 +10:00
parent 25d1774e53
commit 87b60aac9b
40 changed files with 2693 additions and 2553 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/routes/data.py
+97 -77
View File
@@ -1,77 +1,97 @@
import csv
from io import StringIO
from flask import Blueprint, render_template, redirect, request, send_file, url_for, flash
from app.models import Reading, db
from flask_login import login_required, current_user
from datetime import datetime
data = Blueprint('data', __name__)
@data.route('/', methods=['GET', 'POST'])
@login_required
def manage_data():
if request.method == 'POST':
# Handle CSV file upload
file = request.files.get('file')
if file and file.filename.endswith('.csv'):
try:
csv_data = csv.reader(StringIO(file.read().decode('utf-8')))
next(csv_data) # Skip the header row
readings_to_add = []
for row in csv_data:
timestamp, systolic, diastolic, heart_rate = row
readings_to_add.append(Reading(
user_id=current_user.id,
timestamp=datetime.strptime(timestamp, '%Y-%m-%d %H:%M:%S'),
systolic=int(systolic),
diastolic=int(diastolic),
heart_rate=int(heart_rate),
))
db.session.bulk_save_objects(readings_to_add)
db.session.commit()
flash('Data imported successfully!', 'success')
except Exception as e:
flash(f'Error importing data: {str(e)}', 'danger')
else:
flash('Please upload a valid CSV file.', 'danger')
return redirect(url_for('data.manage_data'))
return render_template('data.html')
@data.route('/export', methods=['GET'])
@login_required
def export_data():
import io
from flask import Response
def generate_csv():
"""Stream CSV rows to avoid loading all readings into memory."""
# Write header
output = io.StringIO()
writer = csv.writer(output)
writer.writerow(['Timestamp', 'Systolic', 'Diastolic', 'Heart Rate'])
yield output.getvalue()
output.seek(0)
output.truncate(0)
# Stream readings in chunks using yield_per
readings = Reading.query.filter_by(user_id=current_user.id).order_by(
Reading.timestamp
).yield_per(500)
for reading in readings:
writer.writerow([
reading.timestamp.strftime('%Y-%m-%d %H:%M:%S'),
reading.systolic,
reading.diastolic,
reading.heart_rate,
])
yield output.getvalue()
output.seek(0)
output.truncate(0)
return Response(
generate_csv(),
mimetype='text/csv',
headers={'Content-Disposition': 'attachment; filename=readings.csv'}
)
import csv
from io import StringIO
from flask import Blueprint, render_template, redirect, request, url_for, flash, stream_with_context
from app.models import Reading, db
from flask_login import login_required, current_user
from datetime import datetime
from app.forms import DeleteForm
from app.insights import parse_range, readings_query, user_timezone
data = Blueprint('data', __name__)
@data.route('/', methods=['GET', 'POST'])
@login_required
def manage_data():
form = DeleteForm()
if request.method == 'POST':
if not form.validate_on_submit():
flash('Your form expired. Please try again.', 'danger')
return redirect(url_for('data.manage_data'))
# Handle CSV file upload
file = request.files.get('file')
if file and file.filename.endswith('.csv'):
try:
payload = file.read(2_000_001)
if len(payload) > 2_000_000:
raise ValueError('CSV must be 2 MB or smaller.')
csv_data = csv.reader(StringIO(payload.decode('utf-8-sig')))
if next(csv_data, None) != ['Timestamp', 'Systolic', 'Diastolic', 'Heart Rate']:
raise ValueError('Use the column headings from a BP Tracker export.')
readings_to_add = []
for line, row in enumerate(csv_data, 2):
timestamp, systolic, diastolic, heart_rate = row
if not (50 <= int(systolic) <= 250 and 30 <= int(diastolic) <= 150 and 30 <= int(heart_rate) <= 200):
raise ValueError(f'Check the measurement values on row {line}.')
readings_to_add.append(Reading(
user_id=current_user.id,
timestamp=datetime.strptime(timestamp, '%Y-%m-%d %H:%M:%S'),
systolic=int(systolic),
diastolic=int(diastolic),
heart_rate=int(heart_rate),
))
db.session.bulk_save_objects(readings_to_add)
db.session.commit()
flash('Data imported successfully!', 'success')
except Exception as e:
db.session.rollback()
flash(f'Error importing data: {str(e)}', 'danger')
else:
flash('Please upload a valid CSV file.', 'danger')
return redirect(url_for('data.manage_data'))
return render_template('data.html', form=form)
@data.route('/export', methods=['GET'])
@login_required
def export_data():
import io
from flask import Response
user_id = current_user.id
selected = None
if any(request.args.get(key) for key in ('start_date', 'end_date', 'days')):
try:
selected = parse_range(request.args, user_timezone(), 30)
except ValueError as error:
return render_template('error.html', message=str(error)), 400
def generate_csv():
"""Stream CSV rows to avoid loading all readings into memory."""
# Write header
output = io.StringIO()
writer = csv.writer(output)
writer.writerow(['Timestamp', 'Systolic', 'Diastolic', 'Heart Rate'])
yield output.getvalue()
output.seek(0)
output.truncate(0)
# Stream readings in chunks using yield_per
readings = readings_query(user_id, selected).order_by(
Reading.timestamp, Reading.id
).yield_per(500)
for reading in readings:
writer.writerow([
reading.timestamp.strftime('%Y-%m-%d %H:%M:%S'),
reading.systolic,
reading.diastolic,
reading.heart_rate,
])
yield output.getvalue()
output.seek(0)
output.truncate(0)
return Response(
stream_with_context(generate_csv()),
mimetype='text/csv',
headers={'Content-Disposition': 'attachment; filename=readings.csv'}
)
app/routes/main.py
+176 -347
View File
@@ -1,347 +1,176 @@
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__)
# Number of readings to show per page in list view
PAGE_SIZE = 25
@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 shell and default list view."""
user_tz = timezone(current_user.profile.timezone or 'UTC')
# Calculate weekly averages via SQL
systolic_avg, diastolic_avg, heart_rate_avg = calculate_weekly_summary_sql(current_user.id)
badges = calculate_progress_badges(current_user.id, user_tz)
return render_template(
'dashboard.html',
profile=current_user.profile,
badges=badges,
systolic_avg=systolic_avg,
diastolic_avg=diastolic_avg,
heart_rate_avg=heart_rate_avg,
delete_form=DeleteForm(),
active_view='list',
)
@main.route('/dashboard/list', methods=['GET'])
@login_required
def dashboard_list():
user_tz = timezone(current_user.profile.timezone or 'UTC')
page = request.args.get('page', 1, type=int)
# List view is no longer constrained by date filter
paginated = fetch_readings_paginated(current_user.id, None, None, user_tz, page, PAGE_SIZE)
annotate_readings(paginated.items, user_tz)
return render_template('partials/dashboard_list.html', readings=paginated.items, pagination=paginated)
@main.route('/dashboard/table', methods=['GET'])
@login_required
def dashboard_table():
user_tz = timezone(current_user.profile.timezone or 'UTC')
first_reading, last_reading = get_reading_date_range(current_user.id, user_tz)
start_date = request.args.get('start_date') or (first_reading and first_reading.strftime('%Y-%m-%d'))
end_date = request.args.get('end_date') or (last_reading and last_reading.strftime('%Y-%m-%d'))
page = request.args.get('page', 1, type=int)
paginated = fetch_readings_paginated(current_user.id, start_date, end_date, user_tz, page, PAGE_SIZE)
annotate_readings(paginated.items, user_tz)
return render_template('partials/dashboard_table.html', readings=paginated.items, pagination=paginated, start_date=start_date, end_date=end_date)
@main.route('/dashboard/weekly', methods=['GET'])
@login_required
def dashboard_weekly():
user_tz = timezone(current_user.profile.timezone or 'UTC')
week_offset = request.args.get('week_offset', 0, type=int)
now = datetime.now(user_tz)
target_week_date = now + timedelta(weeks=week_offset)
target_week_start = target_week_date - timedelta(days=target_week_date.weekday())
target_week_start_utc = target_week_start.replace(hour=0, minute=0, second=0, microsecond=0).astimezone(utc)
target_week_end_utc = (target_week_start + timedelta(days=7)).replace(hour=0, minute=0, second=0, microsecond=0).astimezone(utc)
calendar_readings = fetch_readings_for_range(current_user.id, target_week_start_utc, target_week_end_utc)
annotate_readings(calendar_readings, user_tz)
readings_by_day = build_readings_by_day(calendar_readings, user_tz)
week_view = generate_weekly_calendar(readings_by_day, target_week_date, user_tz)
return render_template('partials/dashboard_weekly.html', week=week_view, week_offset=week_offset)
@main.route('/dashboard/monthly', methods=['GET'])
@login_required
def dashboard_monthly():
user_tz = timezone(current_user.profile.timezone or 'UTC')
month_offset = request.args.get('month_offset', 0, type=int)
now = datetime.now(user_tz)
target_month_year = now.year + (now.month + month_offset - 1) // 12
target_month_month = (now.month + month_offset - 1) % 12 + 1
target_month_date = now.replace(year=target_month_year, month=target_month_month, day=1, hour=0, minute=0, second=0, microsecond=0)
first_day = target_month_date
start_date = first_day - timedelta(days=(first_day.weekday() + 1) % 7)
end_date = start_date + timedelta(days=42)
start_utc = start_date.replace(hour=0, minute=0, second=0, microsecond=0).astimezone(utc)
end_utc = end_date.replace(hour=0, minute=0, second=0, microsecond=0).astimezone(utc)
calendar_readings = fetch_readings_for_range(current_user.id, start_utc, end_utc)
annotate_readings(calendar_readings, user_tz)
readings_by_day = build_readings_by_day(calendar_readings, user_tz)
month_view = generate_monthly_calendar(readings_by_day, target_month_date, user_tz)
return render_template('partials/dashboard_monthly.html', month=month_view, target_month_date=target_month_date, month_offset=month_offset)
@main.route('/dashboard/graph', methods=['GET'])
@login_required
def dashboard_graph():
user_tz = timezone(current_user.profile.timezone or 'UTC')
first_reading, last_reading = get_reading_date_range(current_user.id, user_tz)
start_date = request.args.get('start_date') or (first_reading and first_reading.strftime('%Y-%m-%d'))
end_date = request.args.get('end_date') or (last_reading and last_reading.strftime('%Y-%m-%d'))
if start_date and end_date:
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)
calendar_readings = fetch_readings_for_range(current_user.id, start_dt, end_dt)
else:
now = datetime.now(user_tz)
month_start = now.replace(day=1, hour=0, minute=0, second=0, microsecond=0)
month_start_utc = month_start.astimezone(utc)
calendar_readings = fetch_readings_for_range(current_user.id, month_start_utc)
annotate_readings(calendar_readings, user_tz)
graph_data = prepare_graph_data(calendar_readings)
graph_data['start_date'] = start_date
graph_data['end_date'] = end_date
return render_template('partials/dashboard_graph.html', **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_paginated(user_id, start_date, end_date, user_tz, page, per_page):
"""Retrieve paginated readings filtered by date range."""
query = Reading.query.filter_by(user_id=user_id)
if start_date and end_date:
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()).paginate(page=page, per_page=per_page, error_out=False)
def fetch_readings_for_range(user_id, start_utc, end_utc=None):
"""Fetch readings from a UTC start time onwards (for calendar/graph views)."""
query = Reading.query.filter(
Reading.user_id == user_id,
Reading.timestamp >= start_utc
)
if end_utc:
query = query.filter(Reading.timestamp <= end_utc)
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 build_readings_by_day(readings, user_tz):
"""Build a dict mapping dates to readings (single pass, shared by calendar views)."""
readings_by_day = defaultdict(list)
for reading in readings:
local_date = reading.local_timestamp.date() if hasattr(reading, 'local_timestamp') else utc.localize(reading.timestamp).astimezone(user_tz).date()
readings_by_day[local_date].append(reading)
return readings_by_day
def calculate_weekly_summary_sql(user_id):
"""Calculate weekly averages using SQL aggregation (single DB query)."""
one_week_ago = datetime.utcnow() - timedelta(days=7)
result = db.session.query(
func.round(func.avg(Reading.systolic), 1).label('sys_avg'),
func.round(func.avg(Reading.diastolic), 1).label('dia_avg'),
func.round(func.avg(Reading.heart_rate), 1).label('hr_avg'),
).filter(
Reading.user_id == user_id,
Reading.timestamp >= one_week_ago
).first()
if result and result.sys_avg is not None:
return float(result.sys_avg), float(result.dia_avg), float(result.hr_avg)
return 0, 0, 0
def generate_monthly_calendar(readings_by_day, selected_date, local_tz):
"""Generate a monthly calendar view from pre-built readings_by_day."""
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)
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_by_day, selected_date, local_tz):
"""Generate a weekly calendar view from pre-built readings_by_day."""
today = datetime.now(local_tz).date()
start_of_week = selected_date - timedelta(days=selected_date.weekday())
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, reversing so chronological order is left-to-right."""
chronological_readings = list(reversed(readings))
n = len(chronological_readings)
time_percentages = []
systolic_vals = [r.systolic for r in chronological_readings]
diastolic_vals = [r.diastolic for r in chronological_readings]
hr_vals = [r.heart_rate for r in chronological_readings]
sys_avg = round(sum(systolic_vals) / n, 1) if n > 0 else 0
dia_avg = round(sum(diastolic_vals) / n, 1) if n > 0 else 0
hr_avg = round(sum(hr_vals) / n, 1) if n > 0 else 0
if n == 0:
pass
elif n == 1:
time_percentages = [0.5]
else:
first_time = chronological_readings[0].timestamp.timestamp()
last_time = chronological_readings[-1].timestamp.timestamp()
time_span = last_time - first_time
for r in chronological_readings:
if time_span == 0:
time_percentages.append(0.5)
else:
p = (r.timestamp.timestamp() - first_time) / time_span
time_percentages.append(p)
return {
'timestamps': [r.timestamp.strftime('%b %d\n%H:%M') for r in chronological_readings],
'systolic': systolic_vals,
'diastolic': diastolic_vals,
'heart_rate': hr_vals,
'time_percentages': time_percentages,
'sys_avg': sys_avg,
'dia_avg': dia_avg,
'hr_avg': hr_avg,
}
def calculate_progress_badges(user_id, user_tz):
"""Generate badges based on user activity and milestones using optimized queries."""
# Fetch only timestamps (index-only scan on the composite index)
timestamps = db.session.query(Reading.timestamp).filter(Reading.user_id == user_id).order_by(Reading.timestamp.desc()).all()
total_readings = len(timestamps)
return _compute_badges(total_readings, timestamps, user_tz)
def _compute_badges(total_readings, timestamps, user_tz, now_local=None):
if now_local is None:
now_local = datetime.now(user_tz).date()
badges = []
if total_readings == 0:
return badges
streak_count = 0
if timestamps:
distinct_dates = []
last_date = None
for (ts,) in timestamps:
local_date = utc.localize(ts).astimezone(user_tz).date()
if local_date != last_date:
distinct_dates.append(local_date)
last_date = local_date
if distinct_dates:
most_recent_date = distinct_dates[0]
if (now_local - most_recent_date).days <= 1:
streak_count = 1
current_check_date = most_recent_date
for d in distinct_dates[1:]:
if (current_check_date - d).days == 1:
streak_count += 1
current_check_date = d
else:
break
if streak_count >= 1:
badges.append(f"Current Streak: {streak_count} Days")
if streak_count >= 7:
badges.append("Logged Every Day for a Week")
if streak_count >= 30:
badges.append("Monthly Streak")
last_7_readings = timestamps[:7]
if len(last_7_readings) == 7:
if all(5 <= utc.localize(ts).astimezone(user_tz).hour < 12 for (ts,) in last_7_readings):
badges.append("Morning Riser: Logged Readings Every Morning for a Week")
if all(18 <= utc.localize(ts).astimezone(user_tz).hour <= 23 for (ts,) in last_7_readings):
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 total_readings >= m), default=None)
if highest_milestone:
badges.append(f"{highest_milestone} Readings Logged")
return badges
from datetime import date, datetime, timedelta
from flask import Blueprint, render_template, redirect, request, url_for
from flask_login import login_required, current_user
from pytz import utc
from app.models import Reading
from app.insights import DateRange, user_timezone, parse_range, readings_query, summary, annotate, chart_data
main = Blueprint('main', __name__)
PAGE_SIZE = 25
@main.route('/')
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'
@main.route('/dashboard')
@login_required
def dashboard():
view = request.args.get('view', 'overview')
if view not in ('overview', 'history', 'trends'):
view = 'overview'
tz = user_timezone()
mode = request.args.get('mode', 'readings') if view == 'history' else 'readings'
calendar = None
try:
selected = parse_range(request.args, tz, 7 if view == 'overview' else 30)
if mode == 'calendar':
unit = request.args.get('unit', 'month')
if unit not in ('week', 'month'):
raise ValueError('Choose a week or month calendar.')
anchor = date.fromisoformat(request.args.get('anchor', datetime.now(tz).date().isoformat()))
if not 1900 <= anchor.year <= 2099:
raise ValueError('Choose a calendar date between 1900 and 2099.')
if unit == 'week':
start = anchor - timedelta(days=anchor.weekday())
end = start + timedelta(days=6)
prev, following = start - timedelta(days=7), start + timedelta(days=7)
else:
start = anchor.replace(day=1)
following = (start + timedelta(days=32)).replace(day=1)
end = following - timedelta(days=1)
prev = (start - timedelta(days=1)).replace(day=1)
selected = DateRange(start, end, tz)
calendar = dict(unit=unit, anchor=anchor.isoformat(), prev=prev.isoformat(), next=following.isoformat())
except ValueError as error:
return render_template('error.html', message=str(error)), 400
query = readings_query(current_user.id, selected)
stats = summary(query)
pagination = None
readings = []
latest = None
chart = None
if view == 'overview':
latest = readings_query(current_user.id).order_by(Reading.timestamp.desc(), Reading.id.desc()).first()
if latest:
annotate([latest], tz)
readings = annotate(query.order_by(Reading.timestamp.desc(), Reading.id.desc()).limit(5).all(), tz)
elif mode != 'calendar':
if view == 'history':
page = max(1, request.args.get('page', 1, type=int))
pagination = query.order_by(Reading.timestamp.desc(), Reading.id.desc()).paginate(page=page, per_page=PAGE_SIZE, error_out=False)
readings = annotate(pagination.items, tz)
if view in ('overview', 'trends') or calendar:
chart = chart_data(query, selected)
if calendar:
by_day = {p['date']: p for p in chart['points']}
calendar['days'] = [dict(date=(selected.start + timedelta(days=i)), data=by_day.get((selected.start + timedelta(days=i)).isoformat()))
for i in range((selected.end - selected.start).days + 1)]
profile = current_user.profile
active_days = (selected.end - selected.start).days + 1 if selected.end == datetime.now(tz).date() else None
return render_template('dashboard.html', view=view, mode=mode, selected=selected, stats=stats,
readings=readings, latest=latest, chart=chart, pagination=pagination,
calendar=calendar, timezone_name=str(tz), active_days=active_days,
sys_threshold=(profile.systolic_threshold or 140) if profile else 140,
dia_threshold=(profile.diastolic_threshold or 90) if profile else 90)
@main.route('/dashboard/report')
@login_required
def report():
tz = user_timezone()
try:
selected = parse_range(request.args, tz, 30)
except ValueError as error:
return render_template('error.html', message=str(error)), 400
query = readings_query(current_user.id, selected)
readings = annotate(query.order_by(Reading.timestamp, Reading.id).all(), tz)
return render_template('report.html', selected=selected, stats=summary(query), readings=readings, timezone_name=str(tz))
# Preserve bookmarks to the former separate views.
@main.route('/dashboard/list')
@main.route('/dashboard/table')
@login_required
def dashboard_table():
params = request.args.to_dict()
params['view'] = 'history'
return redirect(url_for('main.dashboard', **params))
@main.route('/dashboard/graph')
@login_required
def dashboard_graph():
params = request.args.to_dict()
params['view'] = 'trends'
return redirect(url_for('main.dashboard', **params))
@main.route('/dashboard/weekly')
@main.route('/dashboard/monthly')
@login_required
def dashboard_calendar():
unit = 'week' if request.path.endswith('weekly') else 'month'
anchor = datetime.now(user_timezone()).date()
offset = request.args.get('week_offset' if unit == 'week' else 'month_offset', 0, type=int)
try:
if unit == 'week':
anchor += timedelta(weeks=offset)
else:
month_index = anchor.year * 12 + anchor.month - 1 + offset
anchor = date(month_index // 12, month_index % 12 + 1, 1)
except (OverflowError, ValueError):
return render_template('error.html', message='Choose a valid calendar date.'), 400
return redirect(url_for('main.dashboard', view='history', mode='calendar', unit=unit, anchor=anchor.isoformat()))
def _compute_badges(total_readings, timestamps, user_tz, now_local=None):
if now_local is None:
now_local = datetime.now(user_tz).date()
badges = []
if total_readings == 0:
return badges
streak_count = 0
if timestamps:
distinct_dates = []
last_date = None
for (ts,) in timestamps:
local_date = utc.localize(ts).astimezone(user_tz).date()
if local_date != last_date:
distinct_dates.append(local_date)
last_date = local_date
if distinct_dates:
most_recent_date = distinct_dates[0]
if (now_local - most_recent_date).days <= 1:
streak_count = 1
current_check_date = most_recent_date
for d in distinct_dates[1:]:
if (current_check_date - d).days == 1:
streak_count += 1
current_check_date = d
else:
break
if streak_count >= 1:
badges.append(f"Current Streak: {streak_count} Days")
if streak_count >= 7:
badges.append("Logged Every Day for a Week")
if streak_count >= 30:
badges.append("Monthly Streak")
last_7_readings = timestamps[:7]
if len(last_7_readings) == 7:
if all(5 <= utc.localize(ts).astimezone(user_tz).hour < 12 for (ts,) in last_7_readings):
badges.append("Morning Riser: Logged Readings Every Morning for a Week")
if all(18 <= utc.localize(ts).astimezone(user_tz).hour <= 23 for (ts,) in last_7_readings):
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 total_readings >= m), default=None)
if highest_milestone:
badges.append(f"{highest_milestone} Readings Logged")
return badges
app/routes/reading.py
+86 -79
View File
@@ -1,79 +1,86 @@
from flask import Blueprint, render_template, redirect, request, url_for, flash
from pytz import timezone, utc
from app.models import Reading, db
from app.forms import ReadingForm
from flask_login import login_required, current_user
from datetime import datetime
reading = Blueprint('reading', __name__)
def get_user_timezone():
"""Fetch the user's timezone, defaulting to UTC."""
return timezone(current_user.profile.timezone if current_user.profile and current_user.profile.timezone else 'UTC')
def localize_timestamp(timestamp, user_tz):
"""Convert a UTC timestamp to the user's local timezone."""
return utc.localize(timestamp).astimezone(user_tz)
def save_reading_from_form(reading, form, user_tz):
"""Update a reading with form data and convert the timestamp to UTC."""
local_timestamp = form.timestamp.data
reading.timestamp = user_tz.localize(local_timestamp.replace(tzinfo=None)).astimezone(utc)
reading.systolic = form.systolic.data
reading.diastolic = form.diastolic.data
reading.heart_rate = form.heart_rate.data
db.session.commit()
@reading.route('/add', methods=['GET', 'POST'])
@login_required
def add_reading():
form = ReadingForm()
user_tz = get_user_timezone()
if form.validate_on_submit():
new_reading = Reading(
user_id=current_user.id,
timestamp=user_tz.localize(form.timestamp.data.replace(tzinfo=None)).astimezone(utc),
systolic=form.systolic.data,
diastolic=form.diastolic.data,
heart_rate=form.heart_rate.data,
)
db.session.add(new_reading)
db.session.commit()
flash("Reading added successfully.", "success")
return redirect(url_for('main.dashboard'))
form.timestamp.data = localize_timestamp(datetime.utcnow(), user_tz)
return render_template('reading/add_reading.html', form=form)
@reading.route('/<int:reading_id>/edit', methods=['GET', 'POST'])
@login_required
def edit_reading(reading_id):
reading = Reading.query.filter_by(id=reading_id, user_id=current_user.id).first_or_404()
user_tz = get_user_timezone()
form = ReadingForm(obj=reading)
form.timestamp.data = localize_timestamp(reading.timestamp, user_tz)
if form.validate_on_submit():
save_reading_from_form(reading, form, user_tz)
flash('Reading updated successfully!', 'success')
return redirect(url_for('main.dashboard'))
return render_template('reading/edit_reading.html', form=form, reading=reading)
@reading.route('/<int:reading_id>/confirm_delete', methods=['GET'])
@login_required
def confirm_delete(reading_id):
reading = Reading.query.filter_by(id=reading_id, user_id=current_user.id).first_or_404()
return render_template('reading/confirm_delete.html', reading=reading)
@reading.route('/<int:reading_id>/delete', methods=['POST'])
@login_required
def delete_reading(reading_id):
reading = Reading.query.filter_by(id=reading_id, user_id=current_user.id).first_or_404()
db.session.delete(reading)
db.session.commit()
flash('Reading deleted successfully!', 'success')
return redirect(url_for('main.dashboard'))
from flask import Blueprint, render_template, redirect, request, url_for, flash
from pytz import timezone, utc
from app.models import Reading, db
from app.forms import ReadingForm, DeleteForm
from flask_login import login_required, current_user
from datetime import datetime
reading = Blueprint('reading', __name__)
def get_user_timezone():
"""Fetch the user's timezone, defaulting to UTC."""
return timezone(current_user.profile.timezone if current_user.profile and current_user.profile.timezone else 'UTC')
def localize_timestamp(timestamp, user_tz):
"""Convert a UTC timestamp to the user's local timezone."""
return utc.localize(timestamp).astimezone(user_tz)
def save_reading_from_form(reading, form, user_tz):
"""Update a reading with form data and convert the timestamp to UTC."""
local_timestamp = form.timestamp.data
reading.timestamp = user_tz.localize(local_timestamp.replace(tzinfo=None)).astimezone(utc)
reading.systolic = form.systolic.data
reading.diastolic = form.diastolic.data
reading.heart_rate = form.heart_rate.data
db.session.commit()
@reading.route('/add', methods=['GET', 'POST'])
@login_required
def add_reading():
form = ReadingForm()
user_tz = get_user_timezone()
if form.validate_on_submit():
new_reading = Reading(
user_id=current_user.id,
timestamp=user_tz.localize(form.timestamp.data.replace(tzinfo=None)).astimezone(utc),
systolic=form.systolic.data,
diastolic=form.diastolic.data,
heart_rate=form.heart_rate.data,
)
db.session.add(new_reading)
db.session.commit()
flash("Reading added successfully.", "success")
destination = 'reading.add_reading' if request.form.get('action') == 'another' else 'main.dashboard'
return redirect(url_for(destination))
if request.method == 'GET':
form.timestamp.data = localize_timestamp(datetime.utcnow(), user_tz)
return render_template('reading/add_reading.html', form=form)
@reading.route('/<int:reading_id>/edit', methods=['GET', 'POST'])
@login_required
def edit_reading(reading_id):
reading = Reading.query.filter_by(id=reading_id, user_id=current_user.id).first_or_404()
user_tz = get_user_timezone()
form = ReadingForm(obj=reading)
if request.method == 'GET':
form.timestamp.data = localize_timestamp(reading.timestamp, user_tz)
if form.validate_on_submit():
save_reading_from_form(reading, form, user_tz)
flash('Reading updated successfully!', 'success')
return redirect(url_for('main.dashboard'))
return render_template('reading/edit_reading.html', form=form, reading=reading)
@reading.route('/<int:reading_id>/confirm_delete', methods=['GET'])
@login_required
def confirm_delete(reading_id):
reading = Reading.query.filter_by(id=reading_id, user_id=current_user.id).first_or_404()
reading.local_timestamp = localize_timestamp(reading.timestamp, get_user_timezone())
return render_template('reading/confirm_delete.html', reading=reading, form=DeleteForm())
@reading.route('/<int:reading_id>/delete', methods=['POST'])
@login_required
def delete_reading(reading_id):
reading = Reading.query.filter_by(id=reading_id, user_id=current_user.id).first_or_404()
if not DeleteForm().validate_on_submit():
flash('Please try deleting the reading again.', 'danger')
return redirect(url_for('reading.confirm_delete', reading_id=reading.id))
db.session.delete(reading)
db.session.commit()
flash('Reading deleted successfully!', 'success')
return redirect(url_for('main.dashboard'))
app/routes/user.py
+85 -81
View File
@@ -1,81 +1,85 @@
import io
from flask import Blueprint, make_response, render_template, redirect, url_for, flash
from werkzeug.http import http_date
from app.models import Profile, db
from app.forms import ProfileForm
from flask_login import login_required, current_user
import base64
from datetime import datetime
from PIL import Image
user = Blueprint('user', __name__)
@user.route('/profile', methods=['GET', 'POST'])
@login_required
def profile():
profile = current_user.profile or Profile(user_id=current_user.id)
form = ProfileForm(obj=profile)
if form.validate_on_submit():
# Update profile fields
profile.name = form.name.data
profile.email = form.email.data
profile.systolic_threshold = form.systolic_threshold.data or profile.systolic_threshold
profile.diastolic_threshold = form.diastolic_threshold.data or profile.diastolic_threshold
profile.dark_mode = form.dark_mode.data
profile.timezone = form.timezone.data
# Handle profile picture upload
if form.profile_pic.data:
file_data = form.profile_pic.data.read()
# Resize and compress the image
try:
image = Image.open(io.BytesIO(file_data))
image = image.convert("RGB") # Ensure it's in RGB format
image.thumbnail((200, 200)) # Resize to a maximum of 200x200 pixels
# Save the resized image to a buffer
buffer = io.BytesIO()
image.save(buffer, format="JPEG", quality=80) # Compress with quality=80
buffer.seek(0)
# Encode the compressed image as base64
profile.profile_pic = base64.b64encode(buffer.read()).decode('utf-8')
except Exception as e:
flash(f"Error processing profile picture: {e}", 'danger')
db.session.add(profile)
db.session.commit()
flash('Profile updated successfully!', 'success')
return redirect(url_for('user.profile'))
return render_template('profile.html', form=form, profile=profile)
@user.route('/profile/image/<int:user_id>')
def profile_image(user_id):
# Ensure the reading belongs to the logged-in user
if user_id != current_user.id:
flash('You are not authorized to delete this reading.', 'danger')
return redirect(url_for('main.dashboard'))
profile = Profile.query.filter_by(user_id=user_id).first()
if profile and profile.profile_pic:
image_data = base64.b64decode(profile.profile_pic)
response = make_response(image_data)
response.headers.set('Content-Type', 'image/jpeg')
response.headers.set('Cache-Control', 'public, max-age=86400') # Cache for 1 day
response.headers.set('ETag', str(hash(profile.profile_pic))) # Unique ETag for the image
# Use actual profile update time instead of utcnow() which defeats caching
last_modified = profile.updated_at or datetime.utcnow()
response.headers.set('Last-Modified', http_date(last_modified.timestamp()))
return response
else:
# Serve the default SVG if no profile picture is found
with open('app/static/images/default-profile.svg', 'r') as f:
default_image = f.read()
response = make_response(default_image)
response.headers.set('Content-Type', 'image/svg+xml')
import io
from flask import Blueprint, make_response, render_template, redirect, url_for, flash
from werkzeug.http import http_date
from app.models import Profile, db
from app.forms import ProfileForm
from flask_login import login_required, current_user
import base64
from datetime import datetime
from PIL import Image
from flask_login import login_required, current_user
user = Blueprint('user', __name__)
@user.route('/profile', methods=['GET', 'POST'])
@login_required
@login_required
def profile():
profile = current_user.profile or Profile(user_id=current_user.id)
form = ProfileForm(obj=profile)
if form.validate_on_submit():
# Update profile fields
profile.name = form.name.data
profile.email = form.email.data
profile.systolic_threshold = form.systolic_threshold.data or profile.systolic_threshold
profile.diastolic_threshold = form.diastolic_threshold.data or profile.diastolic_threshold
profile.dark_mode = form.dark_mode.data
profile.timezone = form.timezone.data
# Handle profile picture upload
if form.profile_pic.data:
file_data = form.profile_pic.data.read()
# Resize and compress the image
try:
image = Image.open(io.BytesIO(file_data))
image = image.convert("RGB") # Ensure it's in RGB format
image.thumbnail((200, 200)) # Resize to a maximum of 200x200 pixels
# Save the resized image to a buffer
buffer = io.BytesIO()
image.save(buffer, format="JPEG", quality=80) # Compress with quality=80
buffer.seek(0)
# Encode the compressed image as base64
profile.profile_pic = base64.b64encode(buffer.read()).decode('utf-8')
except Exception as e:
flash(f"Error processing profile picture: {e}", 'danger')
db.session.add(profile)
db.session.commit()
flash('Profile updated successfully!', 'success')
return redirect(url_for('user.profile'))
return render_template('profile.html', form=form, profile=profile)
@user.route('/profile/image/<int:user_id>')
@login_required
def profile_image(user_id):
# Ensure the reading belongs to the logged-in user
if user_id != current_user.id:
flash('You are not authorized to delete this reading.', 'danger')
return redirect(url_for('main.dashboard'))
profile = Profile.query.filter_by(user_id=user_id).first()
if profile and profile.profile_pic:
image_data = base64.b64decode(profile.profile_pic)
response = make_response(image_data)
response.headers.set('Content-Type', 'image/jpeg')
response.headers.set('Cache-Control', 'public, max-age=86400') # Cache for 1 day
response.headers.set('ETag', str(hash(profile.profile_pic))) # Unique ETag for the image
# Use actual profile update time instead of utcnow() which defeats caching
last_modified = profile.updated_at or datetime.utcnow()
response.headers.set('Last-Modified', http_date(last_modified.timestamp()))
return response
else:
# Serve the default SVG if no profile picture is found
with open('app/static/images/default-profile.svg', 'r') as f:
default_image = f.read()
response = make_response(default_image)
response.headers.set('Content-Type', 'image/svg+xml')
return response