109 lines
4.4 KiB
Python
109 lines
4.4 KiB
Python
"""Shared, timezone-aware ranges and bounded chart summaries."""
|
|
from dataclasses import dataclass
|
|
from datetime import date, datetime, time, timedelta
|
|
from math import ceil
|
|
|
|
from flask_login import current_user
|
|
from pytz import timezone, utc
|
|
from sqlalchemy import func
|
|
|
|
from app.models import Reading
|
|
|
|
|
|
def user_timezone():
|
|
profile = current_user.profile
|
|
return timezone(profile.timezone if profile and profile.timezone else 'UTC')
|
|
|
|
|
|
@dataclass
|
|
class DateRange:
|
|
start: date
|
|
end: date
|
|
tz: object
|
|
|
|
@property
|
|
def start_utc(self):
|
|
return self.tz.localize(datetime.combine(self.start, time.min)).astimezone(utc).replace(tzinfo=None)
|
|
|
|
@property
|
|
def end_utc(self):
|
|
# Localize each midnight independently: a local day may be 23 or 25 hours.
|
|
return self.tz.localize(datetime.combine(self.end + timedelta(days=1), time.min)).astimezone(utc).replace(tzinfo=None)
|
|
|
|
@property
|
|
def params(self):
|
|
return {'start_date': self.start.isoformat(), 'end_date': self.end.isoformat()}
|
|
|
|
|
|
def parse_range(args, tz, default_days=7):
|
|
today = datetime.now(tz).date()
|
|
start, end = args.get('start_date'), args.get('end_date')
|
|
if start or end:
|
|
try:
|
|
selected = DateRange(date.fromisoformat(start), date.fromisoformat(end), tz)
|
|
except (TypeError, ValueError):
|
|
raise ValueError('Choose a valid start and end date.')
|
|
else:
|
|
try:
|
|
days = int(args.get('days', default_days))
|
|
except (TypeError, ValueError):
|
|
raise ValueError('Choose a 7, 30 or 90 day period.')
|
|
if days not in (7, 30, 90):
|
|
raise ValueError('Choose a 7, 30 or 90 day period.')
|
|
selected = DateRange(today - timedelta(days=days - 1), today, tz)
|
|
if selected.start > selected.end:
|
|
raise ValueError('Start date must be on or before end date.')
|
|
if selected.start.year < 1900 or selected.end.year > 2100:
|
|
raise ValueError('Choose dates between 1900 and 2100.')
|
|
return selected
|
|
|
|
|
|
def readings_query(user_id, selected=None):
|
|
query = Reading.query.filter_by(user_id=user_id)
|
|
if selected:
|
|
query = query.filter(Reading.timestamp >= selected.start_utc, Reading.timestamp < selected.end_utc)
|
|
return query
|
|
|
|
|
|
def summary(query):
|
|
row = query.with_entities(func.count(Reading.id), func.avg(Reading.systolic),
|
|
func.avg(Reading.diastolic), func.avg(Reading.heart_rate)).one()
|
|
return dict(count=row[0], systolic=round(row[1], 1) if row[0] else None,
|
|
diastolic=round(row[2], 1) if row[0] else None,
|
|
pulse=round(row[3], 1) if row[0] else None)
|
|
|
|
|
|
def annotate(readings, tz):
|
|
for reading in readings:
|
|
reading.local_timestamp = utc.localize(reading.timestamp).astimezone(tz)
|
|
return readings
|
|
|
|
|
|
def chart_data(query, selected):
|
|
"""Stream raw values into <=180 buckets; never average bucket averages."""
|
|
width = max(1, ceil(((selected.end - selected.start).days + 1) / 180))
|
|
buckets = {}
|
|
rows = query.with_entities(Reading.timestamp, Reading.systolic, Reading.diastolic,
|
|
Reading.heart_rate).yield_per(500)
|
|
for ts, systolic, diastolic, pulse in rows:
|
|
day = utc.localize(ts).astimezone(selected.tz).date()
|
|
index = (day - selected.start).days // width
|
|
sums = buckets.setdefault(index, [0, 0, 0, 0])
|
|
for i, value in enumerate((systolic, diastolic, pulse, 1)):
|
|
sums[i] += value
|
|
points = []
|
|
span = (selected.end - selected.start).days
|
|
for index, (sys, dia, pulse, count) in sorted(buckets.items()):
|
|
day = selected.start + timedelta(days=index * width)
|
|
end = min(day + timedelta(days=width - 1), selected.end)
|
|
values = [round(v / count, 1) for v in (sys, dia, pulse)]
|
|
x = round(48 + (day - selected.start).days / span * 704, 1) if span else 400
|
|
points.append(dict(date=day.isoformat(), end=end.isoformat(), count=count,
|
|
systolic=values[0], diastolic=values[1], pulse=values[2], x=x))
|
|
paths = []
|
|
for key, label in [('systolic', 'Systolic'), ('diastolic', 'Diastolic'), ('pulse', 'Pulse')]:
|
|
coords = [(p['x'], round(244 - p[key] / 260 * 208, 1)) for p in points]
|
|
paths.append(dict(key=key, label=label, coordinates=coords,
|
|
path=' '.join(f'{"M" if i == 0 else "L"}{x},{y}' for i, (x, y) in enumerate(coords))))
|
|
return dict(points=points, paths=paths, bucket_days=width)
|