402 lines
12 KiB
Python
402 lines
12 KiB
Python
"""Badge evaluation engine and built-in catalogue."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
|
|
from .models import Badge, BadgeCriterion, Child
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
|
|
def _b(
|
|
id_suffix: str, name: str, description: str, icon: str, tier: str, point_bonus: int, metric: str, value: int
|
|
) -> Badge:
|
|
"""Helper to build a built-in badge."""
|
|
criteria = [BadgeCriterion(metric=metric, operator=">=", value=value)] if metric else []
|
|
badge = Badge(
|
|
name=name,
|
|
description=description,
|
|
icon=icon,
|
|
tier=tier,
|
|
point_bonus=point_bonus,
|
|
criteria=criteria,
|
|
builtin=True,
|
|
enabled=True,
|
|
notify_on_earn=True,
|
|
)
|
|
badge.id = f"builtin.{id_suffix}"
|
|
return badge
|
|
|
|
|
|
BUILTIN_CATALOGUE: list[Badge] = [
|
|
# Bronze
|
|
_b(
|
|
"first_chore",
|
|
"First Chore",
|
|
"Complete your very first chore",
|
|
"mdi:check-circle",
|
|
"bronze",
|
|
0,
|
|
"first_chore",
|
|
1,
|
|
),
|
|
_b("first_reward", "First Reward", "Claim your first reward", "mdi:gift", "bronze", 0, "first_reward", 1),
|
|
_b("100_points", "100 Points", "Earn 100 lifetime points", "mdi:star", "bronze", 0, "total_points", 100),
|
|
_b(
|
|
"10_chores",
|
|
"10 Chores Completed",
|
|
"Complete 10 chores",
|
|
"mdi:checkbox-marked-circle",
|
|
"bronze",
|
|
0,
|
|
"total_chores",
|
|
10,
|
|
),
|
|
# Silver
|
|
_b("500_points", "500 Points", "Earn 500 lifetime points", "mdi:star-circle", "silver", 25, "total_points", 500),
|
|
_b(
|
|
"50_chores",
|
|
"50 Chores Completed",
|
|
"Complete 50 chores",
|
|
"mdi:checkbox-multiple-marked-circle",
|
|
"silver",
|
|
25,
|
|
"total_chores",
|
|
50,
|
|
),
|
|
_b(
|
|
"3_day_streak", "3-Day Streak", "Complete chores 3 days in a row", "mdi:fire", "silver", 25, "current_streak", 3
|
|
),
|
|
_b(
|
|
"first_perfect_week",
|
|
"First Perfect Week",
|
|
"Complete a perfect week",
|
|
"mdi:calendar-star",
|
|
"silver",
|
|
50,
|
|
"perfect_weeks",
|
|
1,
|
|
),
|
|
# Gold
|
|
_b("1000_points", "1000 Points", "Earn 1000 lifetime points", "mdi:trophy", "gold", 100, "total_points", 1000),
|
|
_b(
|
|
"100_chores",
|
|
"100 Chores Completed",
|
|
"Complete 100 chores",
|
|
"mdi:trophy-variant",
|
|
"gold",
|
|
100,
|
|
"total_chores",
|
|
100,
|
|
),
|
|
_b(
|
|
"7_day_streak",
|
|
"7-Day Streak",
|
|
"Complete chores 7 days in a row",
|
|
"mdi:lightning-bolt",
|
|
"gold",
|
|
50,
|
|
"current_streak",
|
|
7,
|
|
),
|
|
_b(
|
|
"5_perfect_weeks",
|
|
"5 Perfect Weeks",
|
|
"Achieve 5 perfect weeks",
|
|
"mdi:calendar-multiple-check",
|
|
"gold",
|
|
100,
|
|
"perfect_weeks",
|
|
5,
|
|
),
|
|
# Platinum
|
|
_b(
|
|
"5000_points",
|
|
"5000 Points",
|
|
"Earn 5000 lifetime points",
|
|
"mdi:diamond-stone",
|
|
"platinum",
|
|
250,
|
|
"total_points",
|
|
5000,
|
|
),
|
|
_b(
|
|
"30_day_streak",
|
|
"30-Day Streak",
|
|
"Complete chores 30 days in a row",
|
|
"mdi:crown",
|
|
"platinum",
|
|
250,
|
|
"current_streak",
|
|
30,
|
|
),
|
|
_b(
|
|
"10_perfect_weeks",
|
|
"10 Perfect Weeks",
|
|
"Achieve 10 perfect weeks",
|
|
"mdi:rainbow",
|
|
"platinum",
|
|
250,
|
|
"perfect_weeks",
|
|
10,
|
|
),
|
|
]
|
|
|
|
|
|
def resolve_metric(metric: str, child: Child, storage) -> int:
|
|
"""Resolve a metric value for a child. Returns int (0 for unknown metrics)."""
|
|
if metric == "total_points":
|
|
return int(child.total_points_earned or 0)
|
|
if metric == "total_chores":
|
|
return int(child.total_chores_completed or 0)
|
|
if metric == "current_streak":
|
|
return int(child.current_streak or 0)
|
|
if metric == "best_streak":
|
|
return int(child.best_streak or 0)
|
|
if metric == "perfect_weeks":
|
|
return len(child.awarded_perfect_weeks or [])
|
|
if metric == "first_chore":
|
|
return 1 if (child.total_chores_completed or 0) >= 1 else 0
|
|
if metric in ("total_rewards", "first_reward"):
|
|
approved_count = sum(1 for c in storage.get_reward_claims() if c.child_id == child.id and c.approved)
|
|
if metric == "first_reward":
|
|
return 1 if approved_count >= 1 else 0
|
|
return approved_count
|
|
return 0
|
|
|
|
|
|
TRIGGER_METRICS: dict[str, set[str]] = {
|
|
"chore_completed": {"total_chores", "first_chore"},
|
|
"points_changed": {"total_points"},
|
|
"reward_redeemed": {"total_rewards", "first_reward"},
|
|
"streak_updated": {"current_streak", "best_streak"},
|
|
"perfect_week": {"perfect_weeks"},
|
|
}
|
|
|
|
|
|
def criterion_met(value: int, operator: str, threshold: int) -> bool:
|
|
"""Evaluate a single badge criterion. Unknown operators default to >=."""
|
|
op = operator or ">="
|
|
if op == ">=":
|
|
return value >= threshold
|
|
if op == "==":
|
|
return value == threshold
|
|
if op == "<=":
|
|
return value <= threshold
|
|
if op == ">":
|
|
return value > threshold
|
|
if op == "<":
|
|
return value < threshold
|
|
if op == "!=":
|
|
return value != threshold
|
|
return value >= threshold
|
|
|
|
|
|
def badge_relevant_to_trigger(badge: Badge, trigger: str) -> bool:
|
|
"""Return True if any of the badge's criteria reference a metric in the trigger's set.
|
|
|
|
'manual' trigger always returns True (re-evaluates everything).
|
|
Badges with no criteria are manual-award-only and never auto-fire.
|
|
"""
|
|
if trigger == "manual":
|
|
return True
|
|
if not badge.criteria:
|
|
return False
|
|
relevant = TRIGGER_METRICS.get(trigger, set())
|
|
return any(c.metric in relevant for c in badge.criteria)
|
|
|
|
|
|
class BadgeCoordinator:
|
|
"""Evaluates badge criteria and emits awards."""
|
|
|
|
def __init__(self, hass, storage, points_coord, notifications=None) -> None:
|
|
self.hass = hass
|
|
self.storage = storage
|
|
self.points_coord = points_coord
|
|
self.notifications = notifications
|
|
|
|
async def evaluate_for_child(
|
|
self,
|
|
child_id: str,
|
|
trigger: str,
|
|
*,
|
|
silent: bool = False,
|
|
) -> list:
|
|
"""Evaluate all enabled badges for this child given a trigger event.
|
|
|
|
Returns the list of newly-created AwardedBadge instances.
|
|
Set silent=True for retroactive backfill (suppresses notifications,
|
|
events, and point bonuses).
|
|
"""
|
|
from .models import AwardedBadge
|
|
|
|
child = self.storage.get_child(child_id)
|
|
if child is None:
|
|
return []
|
|
|
|
new_awards: list[AwardedBadge] = []
|
|
|
|
for badge in self.storage.get_badges():
|
|
if not badge.enabled:
|
|
continue
|
|
if badge.assigned_to and child_id not in badge.assigned_to:
|
|
continue
|
|
if not badge_relevant_to_trigger(badge, trigger):
|
|
continue
|
|
if self.storage.has_awarded(child_id, badge.id):
|
|
continue
|
|
if not badge.criteria:
|
|
continue
|
|
results = [
|
|
criterion_met(resolve_metric(c.metric, child, self.storage), c.operator, c.value)
|
|
for c in badge.criteria
|
|
]
|
|
combinator = (getattr(badge, "combinator", "AND") or "AND").upper()
|
|
passed = any(results) if combinator == "OR" else all(results)
|
|
if not passed:
|
|
continue
|
|
|
|
bonus = 0 if silent else badge.point_bonus
|
|
award = AwardedBadge(
|
|
child_id=child_id,
|
|
badge_id=badge.id,
|
|
silent=silent,
|
|
bonus_credited=bonus,
|
|
)
|
|
self.storage.add_awarded_badge(award)
|
|
new_awards.append(award)
|
|
|
|
if not silent:
|
|
if bonus > 0:
|
|
await self.points_coord.async_add_points(
|
|
child_id,
|
|
bonus,
|
|
reason=f"Badge: {badge.name}",
|
|
)
|
|
self.hass.bus.async_fire(
|
|
"taskmate_badge_earned",
|
|
{
|
|
"child_id": child_id,
|
|
"badge_id": badge.id,
|
|
"awarded_id": award.id,
|
|
"name": badge.name,
|
|
"icon": badge.icon,
|
|
"tier": badge.tier,
|
|
"point_bonus": bonus,
|
|
},
|
|
)
|
|
|
|
if new_awards:
|
|
await self.storage.async_save()
|
|
if not silent:
|
|
await self._dispatch_notifications(new_awards)
|
|
|
|
return new_awards
|
|
|
|
async def award_manually(self, child_id: str, badge_id: str):
|
|
"""Manually award a badge to a child via parent action."""
|
|
from .models import AwardedBadge
|
|
|
|
child = self.storage.get_child(child_id)
|
|
badge = self.storage.get_badge(badge_id)
|
|
if child is None or badge is None:
|
|
return None
|
|
if self.storage.has_awarded(child_id, badge_id):
|
|
return None
|
|
|
|
bonus = badge.point_bonus
|
|
award = AwardedBadge(
|
|
child_id=child_id,
|
|
badge_id=badge_id,
|
|
manually_awarded=True,
|
|
silent=False,
|
|
bonus_credited=bonus,
|
|
)
|
|
self.storage.add_awarded_badge(award)
|
|
if bonus > 0:
|
|
await self.points_coord.async_add_points(
|
|
child_id,
|
|
bonus,
|
|
reason=f"Badge: {badge.name}",
|
|
)
|
|
self.hass.bus.async_fire(
|
|
"taskmate_badge_earned",
|
|
{
|
|
"child_id": child_id,
|
|
"badge_id": badge_id,
|
|
"awarded_id": award.id,
|
|
"name": badge.name,
|
|
"icon": badge.icon,
|
|
"tier": badge.tier,
|
|
"point_bonus": bonus,
|
|
},
|
|
)
|
|
await self._dispatch_notifications([award])
|
|
await self.storage.async_save()
|
|
return award
|
|
|
|
async def revoke(self, awarded_id: str) -> bool:
|
|
"""Revoke an awarded badge; reverse bonus_credited if > 0."""
|
|
matching = [a for a in self.storage.get_awarded_badges() if a.id == awarded_id]
|
|
if not matching:
|
|
return False
|
|
award = matching[0]
|
|
badge = self.storage.get_badge(award.badge_id)
|
|
badge_name = badge.name if badge else "(removed)"
|
|
|
|
self.storage.remove_awarded_badge(awarded_id)
|
|
if award.bonus_credited > 0:
|
|
await self.points_coord.async_remove_points(
|
|
award.child_id,
|
|
award.bonus_credited,
|
|
reason=f"Badge revoked: {badge_name}",
|
|
)
|
|
await self.storage.async_save()
|
|
return True
|
|
|
|
async def rebuild_all(self) -> int:
|
|
"""Re-evaluate all enabled badges across all children, silently.
|
|
|
|
Used by rebuild_badges service and one-time backfill.
|
|
Returns total count of new silent awards.
|
|
"""
|
|
total = 0
|
|
for child in self.storage.get_children():
|
|
new_awards = await self.evaluate_for_child(
|
|
child.id,
|
|
"manual",
|
|
silent=True,
|
|
)
|
|
total += len(new_awards)
|
|
return total
|
|
|
|
async def _dispatch_notifications(self, awards: list) -> None:
|
|
"""Combine awards into one or per-badge ping(s) via the central dispatcher.
|
|
|
|
Per-badge `notify_on_earn` still applies — badges with that flag off are
|
|
excluded from the message but their event is still emitted by the badge
|
|
coordinator's regular bus.fire.
|
|
"""
|
|
per_child: dict[str, list] = {}
|
|
for award in awards:
|
|
badge = self.storage.get_badge(award.badge_id)
|
|
if not badge or not badge.notify_on_earn:
|
|
continue
|
|
per_child.setdefault(award.child_id, []).append(badge)
|
|
|
|
for child_id, badges in per_child.items():
|
|
child = self.storage.get_child(child_id)
|
|
if child is None or self.notifications is None:
|
|
continue
|
|
badge_name = badges[0].name if len(badges) == 1 else ", ".join(b.name for b in badges)
|
|
await self.notifications.fire(
|
|
"badge_earned",
|
|
{
|
|
"child_name": child.name,
|
|
"badge_name": badge_name,
|
|
"badge_id": ",".join(b.id for b in badges),
|
|
"tier": badges[0].tier,
|
|
},
|
|
)
|