jacobh2
6a79435baf
A hand-rolled month-arithmetic module survived the move to dateutil with no caller left but its own tests, and two predicates alongside it were never wired up at all. An error import was also being deferred inside a function for a circular dependency that does not exist.
221 lines
8.7 KiB
Python
221 lines
8.7 KiB
Python
"""Analytics service: days-left-at-age-4 and planned-vs-actual reconciliation.
|
|
|
|
The headline calculation (§6.5 ARCHITECTURE.md):
|
|
|
|
projected_used_by_B4 = days_used + planned_days(date < B4)
|
|
projected_left_at_B4 = total_days - projected_used_by_B4
|
|
savable_after_B4 = min(projected_left_at_B4, CAP) # age-4 ceiling
|
|
expiring_at_B4 = max(projected_left_at_B4 - CAP, 0) # LOST days
|
|
|
|
Where `CAP` is 96 (per child) normally, or 48 (per parent) for a `couple`
|
|
workspace with two parents — see `app.utils.tip_predicates.
|
|
resolve_age4_savable_cap`, the single function that decides which applies.
|
|
|
|
All rule constants come from config.LeaveRules — no magic numbers.
|
|
dateutil.relativedelta handles leap-year safe 4th-birthday arithmetic.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from datetime import date
|
|
from uuid import UUID
|
|
|
|
from dateutil.relativedelta import relativedelta
|
|
|
|
from app.config import settings
|
|
from app.repositories.child_repository import ChildRepository
|
|
from app.repositories.parental_leave_repository import ParentalLeaveRepository
|
|
from app.repositories.personal_data_repository import PersonalDataRepository
|
|
from app.repositories.planned_leave_repository import PlannedLeaveRepository
|
|
from app.repositories.workspace_repository import WorkspaceRepository
|
|
from app.utils.errors import DubbeldagError, NotFoundError
|
|
from app.utils.tip_predicates import days_left_at_age_4_projection, resolve_age4_savable_cap
|
|
|
|
|
|
@dataclass
|
|
class Age4Projection:
|
|
"""Per-parent projection of days remaining at the child's 4th birthday."""
|
|
|
|
personal_data_uuid: UUID
|
|
child_id: UUID
|
|
birth_date: date
|
|
fourth_birthday: date
|
|
total_days: int
|
|
days_used: int # from latest FK snapshot
|
|
planned_days_before_b4: int # from calendar
|
|
projected_left_at_b4: int
|
|
savable_after_b4: int # min(projected_left, savable_cap_at_b4)
|
|
expiring_at_b4: int # max(projected_left - savable_cap_at_b4, 0) — LOST if not used
|
|
# The cap actually applied — 96 (per child) or 48 (per parent, couple
|
|
# workspace with 2 parents). See `resolve_age4_savable_cap`. Surfaced so
|
|
# the frontend can render "X kvar / Y sparbara" without re-deriving the
|
|
# rule itself.
|
|
savable_cap_at_b4: int
|
|
|
|
|
|
@dataclass
|
|
class WorkspaceOverview:
|
|
workspace_id: UUID
|
|
child_id: UUID
|
|
parent_projections: list[Age4Projection]
|
|
combined_days_used: int
|
|
combined_days_left: int
|
|
combined_expiring_at_b4: int
|
|
|
|
|
|
class AnalyticsService:
|
|
def __init__(
|
|
self,
|
|
parental_leave_repo: ParentalLeaveRepository,
|
|
planned_leave_repo: PlannedLeaveRepository,
|
|
child_repo: ChildRepository,
|
|
personal_data_repo: PersonalDataRepository,
|
|
workspace_repo: WorkspaceRepository,
|
|
) -> None:
|
|
self._leave = parental_leave_repo
|
|
self._planned = planned_leave_repo
|
|
self._children = child_repo
|
|
self._parents = personal_data_repo
|
|
self._workspace_repo = workspace_repo
|
|
|
|
async def _resolve_savable_cap(self, workspace_id: UUID) -> int:
|
|
"""The single call site AnalyticsService uses for the age-4 cap —
|
|
|
|
wraps `resolve_age4_savable_cap` with the DB reads it needs
|
|
(workspace planning_mode + member count) so callers never resolve
|
|
the cap any other way.
|
|
"""
|
|
workspace = await self._workspace_repo.get_by_id(workspace_id)
|
|
planning_mode = workspace.planning_mode if workspace is not None else "solo"
|
|
members = await self._parents.list_by_workspace(workspace_id)
|
|
return resolve_age4_savable_cap(planning_mode, len(members))
|
|
|
|
async def get_days_left_at_age_4(
|
|
self, personal_data_uuid: UUID, child_id: UUID
|
|
) -> Age4Projection:
|
|
"""Compute the headline projection for a single parent+child pair."""
|
|
rules = settings.rules
|
|
|
|
child = await self._children.get_by_id(child_id)
|
|
if child is None:
|
|
raise NotFoundError(f"Child {child_id} not found")
|
|
|
|
# Compute the 4th birthday with leap-year-safe arithmetic.
|
|
b4 = child.birth_date + relativedelta(years=4)
|
|
|
|
latest = await self._leave.get_latest_by_parent(personal_data_uuid, child_id)
|
|
if latest is None:
|
|
# No FK data yet — assume full entitlement with 0 used.
|
|
days_used = 0
|
|
total_days = rules.days_per_parent
|
|
else:
|
|
days_used = latest.days_used
|
|
total_days = latest.total_days
|
|
|
|
# Count föräldrapenning planned days that fall before the 4th birthday.
|
|
# Vacation days do NOT count against the FP pool.
|
|
today = date.today()
|
|
sjukpenning_before_b4 = await self._planned.count_by_parent_type(
|
|
personal_data_uuid,
|
|
child_id,
|
|
"sjukpenning",
|
|
from_date=today,
|
|
to_date=b4,
|
|
)
|
|
lagsta_before_b4 = await self._planned.count_by_parent_type(
|
|
personal_data_uuid,
|
|
child_id,
|
|
"lagsta",
|
|
from_date=today,
|
|
to_date=b4,
|
|
)
|
|
planned_fp_before_b4 = sjukpenning_before_b4 + lagsta_before_b4
|
|
|
|
savable_cap = await self._resolve_savable_cap(child.workspace_id)
|
|
|
|
# Shared with the plan-proposal side (`allocation_service._Ledger.
|
|
# build_age4_projections`) so the two surfaces can never re-derive
|
|
# this arithmetic differently — see `days_left_at_age_4_projection`.
|
|
projected_left, expiring = days_left_at_age_4_projection(
|
|
days_used=days_used,
|
|
planned_fp_before_b4=planned_fp_before_b4,
|
|
total_days=total_days,
|
|
max_savable=savable_cap,
|
|
)
|
|
savable = min(projected_left, savable_cap)
|
|
|
|
return Age4Projection(
|
|
personal_data_uuid=personal_data_uuid,
|
|
child_id=child_id,
|
|
birth_date=child.birth_date,
|
|
fourth_birthday=b4,
|
|
total_days=total_days,
|
|
days_used=days_used,
|
|
planned_days_before_b4=planned_fp_before_b4,
|
|
projected_left_at_b4=projected_left,
|
|
savable_after_b4=savable,
|
|
expiring_at_b4=expiring,
|
|
savable_cap_at_b4=savable_cap,
|
|
)
|
|
|
|
async def get_overview(self, workspace_id: UUID, child_id: UUID) -> WorkspaceOverview:
|
|
"""Compute projections for all parents in a workspace for one child."""
|
|
members = await self._parents.list_by_workspace(workspace_id)
|
|
if not members:
|
|
raise NotFoundError(f"No parents found in workspace {workspace_id}")
|
|
|
|
projections = []
|
|
for pd in members:
|
|
proj = await self.get_days_left_at_age_4(pd.uuid, child_id)
|
|
projections.append(proj)
|
|
|
|
combined_used = sum(p.days_used + p.planned_days_before_b4 for p in projections)
|
|
combined_left = sum(p.projected_left_at_b4 for p in projections)
|
|
combined_expiring = sum(p.expiring_at_b4 for p in projections)
|
|
|
|
return WorkspaceOverview(
|
|
workspace_id=workspace_id,
|
|
child_id=child_id,
|
|
parent_projections=projections,
|
|
combined_days_used=combined_used,
|
|
combined_days_left=combined_left,
|
|
combined_expiring_at_b4=combined_expiring,
|
|
)
|
|
|
|
async def validate_dubbeldagar(self, child_id: UUID, new_date: date) -> None:
|
|
"""Raise DubbeldagError if adding a dubbeldagar day would break rules.
|
|
|
|
Called from the planning service before persisting a new day.
|
|
"""
|
|
child = await self._children.get_by_id(child_id)
|
|
if child is None:
|
|
raise NotFoundError(f"Child {child_id} not found")
|
|
|
|
# Window: child's birth_date ... birth_date + 15 months
|
|
window_end = child.birth_date + relativedelta(
|
|
months=settings.rules.dubbeldagar_months_limit
|
|
)
|
|
if new_date > window_end:
|
|
raise DubbeldagError(
|
|
f"Dubbeldagar only allowed within the child's first "
|
|
f"{settings.rules.dubbeldagar_months_limit} months "
|
|
f"(deadline {window_end})",
|
|
message_key="planApply.rejection.dubbeldagWindowExceeded",
|
|
data={
|
|
"months_limit": settings.rules.dubbeldagar_months_limit,
|
|
"deadline": window_end.isoformat(),
|
|
},
|
|
)
|
|
|
|
# Quota: count already-planned dubbeldagar from child's birth_date onward.
|
|
already_planned = await self._planned.count_dubbeldagar(
|
|
child_id, from_date=child.birth_date
|
|
)
|
|
if already_planned >= settings.rules.max_dubbeldagar:
|
|
raise DubbeldagError(
|
|
f"Dubbeldagar quota ({settings.rules.max_dubbeldagar} days) already reached",
|
|
message_key="planApply.rejection.dubbeldagQuotaReached",
|
|
data={"max_dubbeldagar": settings.rules.max_dubbeldagar},
|
|
)
|