feat: US-001 - Habits JSON schema and helper functions
This commit is contained in:
Echo
385
dashboard/habits_helpers.py
Normal file
385
dashboard/habits_helpers.py
Normal file
@@ -0,0 +1,385 @@
|
||||
"""
|
||||
Habit Tracker Helper Functions
|
||||
|
||||
This module provides core helper functions for calculating streaks,
|
||||
checking relevance, and computing stats for habits.
|
||||
"""
|
||||
|
||||
from datetime import datetime, timedelta
|
||||
from typing import Dict, List, Any, Optional
|
||||
|
||||
|
||||
def calculate_streak(habit: Dict[str, Any]) -> int:
|
||||
"""
|
||||
Calculate the current streak for a habit based on its frequency type.
|
||||
|
||||
Args:
|
||||
habit: Dict containing habit data with frequency, completions, etc.
|
||||
|
||||
Returns:
|
||||
int: Current streak count (days, weeks, or months depending on frequency)
|
||||
"""
|
||||
frequency_type = habit.get("frequency", {}).get("type", "daily")
|
||||
completions = habit.get("completions", [])
|
||||
|
||||
if not completions:
|
||||
return 0
|
||||
|
||||
# Sort completions by date (newest first)
|
||||
sorted_completions = sorted(
|
||||
[c for c in completions if c.get("date")],
|
||||
key=lambda x: x["date"],
|
||||
reverse=True
|
||||
)
|
||||
|
||||
if not sorted_completions:
|
||||
return 0
|
||||
|
||||
if frequency_type == "daily":
|
||||
return _calculate_daily_streak(sorted_completions)
|
||||
elif frequency_type == "specific_days":
|
||||
return _calculate_specific_days_streak(habit, sorted_completions)
|
||||
elif frequency_type == "x_per_week":
|
||||
return _calculate_x_per_week_streak(habit, sorted_completions)
|
||||
elif frequency_type == "weekly":
|
||||
return _calculate_weekly_streak(sorted_completions)
|
||||
elif frequency_type == "monthly":
|
||||
return _calculate_monthly_streak(sorted_completions)
|
||||
elif frequency_type == "custom":
|
||||
return _calculate_custom_streak(habit, sorted_completions)
|
||||
|
||||
return 0
|
||||
|
||||
|
||||
def _calculate_daily_streak(completions: List[Dict[str, Any]]) -> int:
|
||||
"""Calculate streak for daily habits (consecutive days)."""
|
||||
streak = 0
|
||||
today = datetime.now().date()
|
||||
expected_date = today
|
||||
|
||||
for completion in completions:
|
||||
completion_date = datetime.fromisoformat(completion["date"]).date()
|
||||
|
||||
if completion_date == expected_date:
|
||||
streak += 1
|
||||
expected_date = completion_date - timedelta(days=1)
|
||||
elif completion_date < expected_date:
|
||||
# Gap found, streak breaks
|
||||
break
|
||||
|
||||
return streak
|
||||
|
||||
|
||||
def _calculate_specific_days_streak(habit: Dict[str, Any], completions: List[Dict[str, Any]]) -> int:
|
||||
"""Calculate streak for specific days habits (only count relevant days)."""
|
||||
relevant_days = set(habit.get("frequency", {}).get("days", []))
|
||||
if not relevant_days:
|
||||
return 0
|
||||
|
||||
streak = 0
|
||||
today = datetime.now().date()
|
||||
current_date = today
|
||||
|
||||
# Find the most recent relevant day
|
||||
while current_date.weekday() not in relevant_days:
|
||||
current_date -= timedelta(days=1)
|
||||
|
||||
for completion in completions:
|
||||
completion_date = datetime.fromisoformat(completion["date"]).date()
|
||||
|
||||
if completion_date == current_date:
|
||||
streak += 1
|
||||
# Move to previous relevant day
|
||||
current_date -= timedelta(days=1)
|
||||
while current_date.weekday() not in relevant_days:
|
||||
current_date -= timedelta(days=1)
|
||||
elif completion_date < current_date:
|
||||
# Check if we missed a relevant day
|
||||
temp_date = current_date
|
||||
found_gap = False
|
||||
while temp_date > completion_date:
|
||||
if temp_date.weekday() in relevant_days:
|
||||
found_gap = True
|
||||
break
|
||||
temp_date -= timedelta(days=1)
|
||||
if found_gap:
|
||||
break
|
||||
|
||||
return streak
|
||||
|
||||
|
||||
def _calculate_x_per_week_streak(habit: Dict[str, Any], completions: List[Dict[str, Any]]) -> int:
|
||||
"""Calculate streak for x_per_week habits (consecutive weeks meeting target)."""
|
||||
target_count = habit.get("frequency", {}).get("count", 1)
|
||||
|
||||
streak = 0
|
||||
today = datetime.now().date()
|
||||
|
||||
# Group completions by week
|
||||
week_counts = {}
|
||||
for completion in completions:
|
||||
completion_date = datetime.fromisoformat(completion["date"]).date()
|
||||
# Get ISO week (year, week_number)
|
||||
week_key = completion_date.isocalendar()[:2]
|
||||
week_counts[week_key] = week_counts.get(week_key, 0) + 1
|
||||
|
||||
# Start from current week and count backwards
|
||||
current_week = today.isocalendar()[:2]
|
||||
|
||||
while current_week in week_counts:
|
||||
if week_counts[current_week] >= target_count:
|
||||
streak += 1
|
||||
# Move to previous week
|
||||
year, week = current_week
|
||||
if week == 1:
|
||||
year -= 1
|
||||
week = 52
|
||||
else:
|
||||
week -= 1
|
||||
current_week = (year, week)
|
||||
else:
|
||||
break
|
||||
|
||||
return streak
|
||||
|
||||
|
||||
def _calculate_weekly_streak(completions: List[Dict[str, Any]]) -> int:
|
||||
"""Calculate streak for weekly habits (consecutive weeks with at least one check)."""
|
||||
today = datetime.now().date()
|
||||
|
||||
# Group completions by week
|
||||
weeks_with_checks = set()
|
||||
for completion in completions:
|
||||
completion_date = datetime.fromisoformat(completion["date"]).date()
|
||||
week_key = completion_date.isocalendar()[:2]
|
||||
weeks_with_checks.add(week_key)
|
||||
|
||||
streak = 0
|
||||
current_week = today.isocalendar()[:2]
|
||||
|
||||
while current_week in weeks_with_checks:
|
||||
streak += 1
|
||||
year, week = current_week
|
||||
if week == 1:
|
||||
year -= 1
|
||||
week = 52
|
||||
else:
|
||||
week -= 1
|
||||
current_week = (year, week)
|
||||
|
||||
return streak
|
||||
|
||||
|
||||
def _calculate_monthly_streak(completions: List[Dict[str, Any]]) -> int:
|
||||
"""Calculate streak for monthly habits (consecutive months with at least one check)."""
|
||||
today = datetime.now().date()
|
||||
|
||||
# Group completions by month
|
||||
months_with_checks = set()
|
||||
for completion in completions:
|
||||
completion_date = datetime.fromisoformat(completion["date"]).date()
|
||||
month_key = (completion_date.year, completion_date.month)
|
||||
months_with_checks.add(month_key)
|
||||
|
||||
streak = 0
|
||||
current_month = (today.year, today.month)
|
||||
|
||||
while current_month in months_with_checks:
|
||||
streak += 1
|
||||
year, month = current_month
|
||||
if month == 1:
|
||||
year -= 1
|
||||
month = 12
|
||||
else:
|
||||
month -= 1
|
||||
current_month = (year, month)
|
||||
|
||||
return streak
|
||||
|
||||
|
||||
def _calculate_custom_streak(habit: Dict[str, Any], completions: List[Dict[str, Any]]) -> int:
|
||||
"""Calculate streak for custom interval habits (every X days)."""
|
||||
interval = habit.get("frequency", {}).get("interval", 1)
|
||||
if interval <= 0:
|
||||
return 0
|
||||
|
||||
streak = 0
|
||||
expected_date = datetime.now().date()
|
||||
|
||||
for completion in completions:
|
||||
completion_date = datetime.fromisoformat(completion["date"]).date()
|
||||
|
||||
# Allow completion within the interval window
|
||||
days_diff = (expected_date - completion_date).days
|
||||
if 0 <= days_diff <= interval - 1:
|
||||
streak += 1
|
||||
expected_date = completion_date - timedelta(days=interval)
|
||||
else:
|
||||
break
|
||||
|
||||
return streak
|
||||
|
||||
|
||||
def should_check_today(habit: Dict[str, Any]) -> bool:
|
||||
"""
|
||||
Check if a habit is relevant for today based on its frequency type.
|
||||
|
||||
Args:
|
||||
habit: Dict containing habit data with frequency settings
|
||||
|
||||
Returns:
|
||||
bool: True if the habit should be checked today
|
||||
"""
|
||||
frequency_type = habit.get("frequency", {}).get("type", "daily")
|
||||
today = datetime.now().date()
|
||||
weekday = today.weekday() # 0=Monday, 6=Sunday
|
||||
|
||||
if frequency_type == "daily":
|
||||
return True
|
||||
|
||||
elif frequency_type == "specific_days":
|
||||
relevant_days = set(habit.get("frequency", {}).get("days", []))
|
||||
return weekday in relevant_days
|
||||
|
||||
elif frequency_type == "x_per_week":
|
||||
# Always relevant for x_per_week (can check any day)
|
||||
return True
|
||||
|
||||
elif frequency_type == "weekly":
|
||||
# Always relevant (can check any day of the week)
|
||||
return True
|
||||
|
||||
elif frequency_type == "monthly":
|
||||
# Always relevant (can check any day of the month)
|
||||
return True
|
||||
|
||||
elif frequency_type == "custom":
|
||||
# Check if enough days have passed since last completion
|
||||
completions = habit.get("completions", [])
|
||||
if not completions:
|
||||
return True
|
||||
|
||||
interval = habit.get("frequency", {}).get("interval", 1)
|
||||
last_completion = max(completions, key=lambda x: x.get("date", ""))
|
||||
last_date = datetime.fromisoformat(last_completion["date"]).date()
|
||||
days_since = (today - last_date).days
|
||||
|
||||
return days_since >= interval
|
||||
|
||||
return False
|
||||
|
||||
|
||||
def get_completion_rate(habit: Dict[str, Any], days: int = 30) -> float:
|
||||
"""
|
||||
Calculate the completion rate as a percentage over the last N days.
|
||||
|
||||
Args:
|
||||
habit: Dict containing habit data
|
||||
days: Number of days to look back (default 30)
|
||||
|
||||
Returns:
|
||||
float: Completion rate as percentage (0-100)
|
||||
"""
|
||||
frequency_type = habit.get("frequency", {}).get("type", "daily")
|
||||
completions = habit.get("completions", [])
|
||||
|
||||
today = datetime.now().date()
|
||||
start_date = today - timedelta(days=days - 1)
|
||||
|
||||
# Count relevant days and checked days
|
||||
relevant_days = 0
|
||||
checked_dates = set()
|
||||
|
||||
for completion in completions:
|
||||
completion_date = datetime.fromisoformat(completion["date"]).date()
|
||||
if start_date <= completion_date <= today:
|
||||
checked_dates.add(completion_date)
|
||||
|
||||
# Calculate relevant days based on frequency type
|
||||
if frequency_type == "daily":
|
||||
relevant_days = days
|
||||
|
||||
elif frequency_type == "specific_days":
|
||||
relevant_day_set = set(habit.get("frequency", {}).get("days", []))
|
||||
current = start_date
|
||||
while current <= today:
|
||||
if current.weekday() in relevant_day_set:
|
||||
relevant_days += 1
|
||||
current += timedelta(days=1)
|
||||
|
||||
elif frequency_type == "x_per_week":
|
||||
target_per_week = habit.get("frequency", {}).get("count", 1)
|
||||
num_weeks = days // 7
|
||||
relevant_days = num_weeks * target_per_week
|
||||
|
||||
elif frequency_type == "weekly":
|
||||
num_weeks = days // 7
|
||||
relevant_days = num_weeks
|
||||
|
||||
elif frequency_type == "monthly":
|
||||
num_months = days // 30
|
||||
relevant_days = num_months
|
||||
|
||||
elif frequency_type == "custom":
|
||||
interval = habit.get("frequency", {}).get("interval", 1)
|
||||
relevant_days = days // interval if interval > 0 else 0
|
||||
|
||||
if relevant_days == 0:
|
||||
return 0.0
|
||||
|
||||
checked_days = len(checked_dates)
|
||||
return (checked_days / relevant_days) * 100
|
||||
|
||||
|
||||
def get_weekly_summary(habit: Dict[str, Any]) -> Dict[str, str]:
|
||||
"""
|
||||
Get a summary of the current week showing status for each day.
|
||||
|
||||
Args:
|
||||
habit: Dict containing habit data
|
||||
|
||||
Returns:
|
||||
Dict mapping day names to status: "checked", "skipped", "missed", or "upcoming"
|
||||
"""
|
||||
frequency_type = habit.get("frequency", {}).get("type", "daily")
|
||||
completions = habit.get("completions", [])
|
||||
|
||||
today = datetime.now().date()
|
||||
|
||||
# Start of current week (Monday)
|
||||
start_of_week = today - timedelta(days=today.weekday())
|
||||
|
||||
# Create completion map
|
||||
completion_map = {}
|
||||
for completion in completions:
|
||||
completion_date = datetime.fromisoformat(completion["date"]).date()
|
||||
if completion_date >= start_of_week:
|
||||
completion_type = completion.get("type", "check")
|
||||
completion_map[completion_date] = completion_type
|
||||
|
||||
# Build summary for each day of the week
|
||||
summary = {}
|
||||
day_names = ["Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday", "Sunday"]
|
||||
|
||||
for i, day_name in enumerate(day_names):
|
||||
day_date = start_of_week + timedelta(days=i)
|
||||
|
||||
if day_date > today:
|
||||
summary[day_name] = "upcoming"
|
||||
elif day_date in completion_map:
|
||||
if completion_map[day_date] == "skip":
|
||||
summary[day_name] = "skipped"
|
||||
else:
|
||||
summary[day_name] = "checked"
|
||||
else:
|
||||
# Check if this day was relevant
|
||||
if frequency_type == "specific_days":
|
||||
relevant_days = set(habit.get("frequency", {}).get("days", []))
|
||||
if day_date.weekday() not in relevant_days:
|
||||
summary[day_name] = "not_relevant"
|
||||
else:
|
||||
summary[day_name] = "missed"
|
||||
else:
|
||||
summary[day_name] = "missed"
|
||||
|
||||
return summary
|
||||
Reference in New Issue
Block a user