"""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) for a genuine single parent, or 48 (per parent, non-poolable) as soon as the child has 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, field from datetime import date from uuid import UUID from dateutil.relativedelta import relativedelta from app.config import settings from app.entities.child import Child from app.entities.parental_leave import ParentalLeaveFetch from app.entities.planned_leave import PlannedLeaveDay from app.entities.vacation_days import VacationDays 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.vacation_days_repository import VacationDaysRepository 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, default_tier_pool_share, resolve_age4_savable_cap, ) @dataclass class VacationYearBalance: """One parent's vacation (semester) balance for a single calendar year. Part of the per-parent, per-year balance series (`ParentYearBalance`) — see that dataclass's docstring for the carry-forward algorithm. """ year: int earned: int # entitlement for this year: the `vacation_days` row's total_days, or the statutory default is_assumed: bool # True when no `vacation_days` row exists for this (parent, year) carried_in: int # days carried in from prior years, still valid planned_use: int # committed 'vacation' planned_leave_day rows in this year (see PlanningService._vacation_days_left) left: int # max(earned + carried_in - planned_use, 0) carried_out: int # days saved forward into next year's carried_in # EVERY day lost at this year's end, from both causes combined — the sum # of `expiring_age` and `expiring_over_cap` below. Never a subset: a UI # showing only `expiring` must not silently under-report a forfeiture. expiring: int # subset of `expiring`: bank entries that hit the 5-year # (rules.vacation_carry_forward_years) age cap this year. expiring_age: int # subset of `expiring`: days above (earned - floor) that couldn't be # banked — includes an unused floor itself when usage is low. expiring_over_cap: int @dataclass class ParentalLeaveYearBalance: """One parent's föräldrapenning (FP) balance for a single calendar year, for one child — see `ParentYearBalance`.""" year: int sjukpenning_used: int # FP sjukpenning-tier rows dated in this year lagsta_used: int # FP lägstanivå-tier rows dated in this year sjukpenning_left: int # running balance at this year's end lagsta_left: int # running balance at this year's end @dataclass class ParentYearBalance: """One (parent, child, calendar year) slice of the balance series shown on the analytics overview — vacation and FP side by side. **Vacation carry-forward algorithm** (Semesterlagen): a parent must take at least `PersonalData.min_vacation_days_per_year` (the "floor") each year; only days earned *above* the floor may be saved, and each saved day expires `LeaveRules.vacation_carry_forward_years` (5) years after the year it was earned. Modelled as a FIFO bank of `(year_earned, count)` entries: - Each year, `planned_use` is drawn from the bank oldest-entry-first, then from the year's own entitlement (which is never itself banked). - At year end, `carried_out = min(left, max(earned - floor, 0) + old_carry_forward)` joins the bank as a new entry dated this year — `old_carry_forward` (the surviving pre-existing bank balance) keeps its original earn-year for its own future expiry, it is never re-dated. - A bank entry in its final valid year (earned exactly `vacation_carry_forward_years` years ago) that still has an unconsumed balance drops out instead of carrying forward, and is reported as `expiring_age` for that year. - Separately, whatever of `left` cannot be absorbed by `carried_out` (days above `earned - floor` with nowhere to bank — including an unused floor itself, when usage is low) is `expiring_over_cap` for that year. - `expiring` is `expiring_age + expiring_over_cap`: EVERY vacation day lost at this year's end, however it was lost. Earlier versions of this field counted only the age-cap portion, so a family could lose days purely to the over-cap cause (e.g. earned 30, floor 20, used 15 → 5 days over-cap lost) while the dashboard showed "Expiring 0" — that silent gap is why the breakdown exists now. `saved_days_in` on a `vacation_days` row is normally only used to *seed* the bank for the first year of the series (year-earned = year − 1, since we have no earlier history to derive it from). For any LATER year in the series, a non-zero `saved_days_in` on that year's row is instead treated as a Settings-UI override: it replaces the computed `carried_in` (and resets the bank to a single entry dated year − 1) rather than being added to it — the user is telling us the true figure directly. A zero (or absent) `saved_days_in` on a later year never overrides anything. """ personal_data_uuid: UUID child_id: UUID year: int vacation: VacationYearBalance parental_leave: ParentalLeaveYearBalance @dataclass class YearTotals: """Combined (all parents) balance for one calendar year, on the overview.""" year: int pl_left: int vacation_left: int @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, single parent) or 48 (per # parent, non-poolable, as soon as the child has 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 # Per-calendar-year vacation + FP balance series, birth year through the # year of the 4th birthday inclusive. See `ParentYearBalance`. year_balances: list[ParentYearBalance] = field(default_factory=list) @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 # Combined (all parents) per-year totals — see `YearTotals`. year_totals: list[YearTotals] = field(default_factory=list) class AnalyticsService: def __init__( self, parental_leave_repo: ParentalLeaveRepository, planned_leave_repo: PlannedLeaveRepository, child_repo: ChildRepository, personal_data_repo: PersonalDataRepository, workspace_repo: WorkspaceRepository, vacation_days_repo: VacationDaysRepository, ) -> 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 self._vacation = vacation_days_repo async def get_parent_count(self, workspace_id: UUID) -> int: """Number of (non-deleted) parents on a workspace. Shared read so callers that need a parent count (the age-4 cap here, `PlanningService._tier_pool_left`'s default-tier-share lookup) never query membership differently. """ members = await self._parents.list_by_workspace(workspace_id) return len(members) async def get_parent_count_for_child(self, child_id: UUID) -> int: """Number of parents on the workspace that owns this child. Lets a caller that only has a `child_id` (no `workspace_id` in scope) still resolve a parent count — see `PlanningService. _tier_pool_left`, which only ever receives `(parent_uuid, child_id)`. """ child = await self._children.get_by_id(child_id) if child is None: raise NotFoundError(f"Child {child_id} not found") return await self.get_parent_count(child.workspace_id) 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 read it needs (member count) so callers never resolve the cap any other way. The cap is a fact about how many parents the child has, not a workspace display preference — see `resolve_age4_savable_cap`'s docstring. """ return resolve_age4_savable_cap(await self.get_parent_count(workspace_id)) async def get_days_left_at_age_4( self, personal_data_uuid: UUID, child_id: UUID, workspace_id: UUID | None = None ) -> Age4Projection: """Compute the headline projection for a single parent+child pair. `workspace_id`, when given, must match the child's own — a foreign workspace's child 404s exactly like a nonexistent one. """ rules = settings.rules child = await self._children.get_by_id(child_id) if child is None or (workspace_id is not None and child.workspace_id != workspace_id): 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) default_share: tuple[int, int] | None = None if latest is None: # No FK data yet — assume full entitlement (0 used) for this # parent's share, via the same shared helper `AllocationInputLoader. # load`/`PlanningService._tier_pool_left` use, so a sole parent # (their whole-pool share) and a partnered one (half) never # disagree with the Plan tab's own no-snapshot default (Bug A). days_used = 0 parent_count = await self.get_parent_count(child.workspace_id) sjuk_share, lagsta_share = default_tier_pool_share(rules, parent_count) total_days = sjuk_share + lagsta_share default_share = (sjuk_share, lagsta_share) 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) year_balances = await self._compute_year_balances( personal_data_uuid, child, b4, savable_cap, latest, default_share ) 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, year_balances=year_balances, ) async def _compute_year_balances( self, personal_data_uuid: UUID, child: Child, fourth_birthday: date, savable_cap: int, latest_fetch: ParentalLeaveFetch | None, default_share: tuple[int, int] | None, ) -> list[ParentYearBalance]: """Assemble the per-calendar-year vacation + FP balance series for one (parent, child) pair — see `ParentYearBalance`'s docstring for the vacation carry-forward algorithm. `years` runs from the child's birth year through the year of the 4th birthday inclusive. """ years = list(range(child.birth_date.year, fourth_birthday.year + 1)) personal_data = await self._parents.get_by_id(personal_data_uuid) floor = ( personal_data.min_vacation_days_per_year if personal_data is not None else settings.rules.min_vacation_days_per_year ) vacation_rows = await self._vacation.list(personal_data_uuid=personal_data_uuid) vacation_by_year = {v.year: v for v in vacation_rows} # Vacation is budgeted per (parent, calendar year), not per child — # see `PlanningService._vacation_days_left`'s docstring. `planned_use` # below deliberately counts every persisted 'vacation' row for this # parent across ALL of their children in that year, matching that # method exactly, so this row's `left` never disagrees with the Plan # tab's own vacation-balance check. all_parent_days = await self._planned.list(parent_personal_data_uuid=personal_data_uuid) vacation_use_by_year: dict[int, int] = {} for d in all_parent_days: if d.day_type == "vacation": vacation_use_by_year[d.date.year] = vacation_use_by_year.get(d.date.year, 0) + 1 vacation_balances = self._vacation_year_series( years, vacation_by_year, vacation_use_by_year, floor ) fp_rows = [ d for d in all_parent_days if d.child_id == child.id and d.day_type in ("sjukpenning", "lagsta") ] pl_balances = self._pl_year_series( years, fp_rows, child.birth_date, fourth_birthday, savable_cap, latest_fetch, default_share, ) return [ ParentYearBalance( personal_data_uuid=personal_data_uuid, child_id=child.id, year=year, vacation=vacation_balances[i], parental_leave=pl_balances[i], ) for i, year in enumerate(years) ] @staticmethod def _vacation_year_series( years: list[int], vacation_by_year: dict[int, VacationDays], use_by_year: dict[int, int], floor: int, ) -> list[VacationYearBalance]: """FIFO carry-forward bank simulation — see `ParentYearBalance`'s docstring for the full algorithm this implements.""" carry_years = settings.rules.vacation_carry_forward_years default_earned = settings.rules.statutory_vacation_days_per_year # (year_earned, remaining_count) entries, oldest first. bank: list[list[int]] = [] result: list[VacationYearBalance] = [] for i, year in enumerate(years): row = vacation_by_year.get(year) earned = row.total_days if row is not None else default_earned is_assumed = row is None if i == 0: # Seed the bank for the first year in the series — there is # no earlier history to derive a carry-in from ourselves, so # this is purely the user-entered figure (Settings UI), # attributed to year - 1 for its own future expiry. seed = row.saved_days_in if row is not None else 0 bank = [[year - 1, seed]] if seed > 0 else [] carried_in = seed elif row is not None and row.saved_days_in != 0: # Later-year override: a non-zero saved_days_in on THIS # year's row supersedes the computed carry-in — see # `ParentYearBalance`'s docstring. bank = [[year - 1, row.saved_days_in]] if row.saved_days_in > 0 else [] carried_in = row.saved_days_in else: carried_in = sum(count for _, count in bank) planned_use = use_by_year.get(year, 0) left = max(earned + carried_in - planned_use, 0) # FIFO consumption: oldest bank entries first, then this year's # own entitlement (never itself banked). remaining_to_consume = planned_use for entry in bank: if remaining_to_consume <= 0: break take = min(entry[1], remaining_to_consume) entry[1] -= take remaining_to_consume -= take bank = [e for e in bank if e[1] > 0] # Entries still valid to carry into next year vs. ones in their # final valid year (expire now if any balance survived). old_carry_forward = sum(c for ye, c in bank if ye >= year - (carry_years - 1)) expiring_age_raw = sum(c for ye, c in bank if ye == year - carry_years) bank = [e for e in bank if e[0] >= year - (carry_years - 1)] current_year_saveable = max(earned - floor, 0) carried_out = min(left, current_year_saveable + old_carry_forward) new_entry_amount = max(carried_out - old_carry_forward, 0) if new_entry_amount > 0: bank.append([year, new_entry_amount]) # `left` minus what actually got banked is EVERY day lost this # year — from both causes. `expiring_age_raw` (bank entries that # just hit the age cap) is a subset of that total by construction # (those days are inside `left` and excluded from `carried_out` # via `old_carry_forward`), but clamp it defensively so the two # portions can never sum to more than `total_lost` in an edge # case (e.g. `left` itself clamped to 0 by the `max(..., 0)` # above, when planned_use outstrips earned + carried_in). total_lost = max(left - carried_out, 0) expiring_age = min(expiring_age_raw, total_lost) expiring_over_cap = total_lost - expiring_age result.append( VacationYearBalance( year=year, earned=earned, is_assumed=is_assumed, carried_in=carried_in, planned_use=planned_use, left=left, carried_out=carried_out, expiring=total_lost, expiring_age=expiring_age, expiring_over_cap=expiring_over_cap, ) ) return result @staticmethod def _pl_year_series( years: list[int], fp_rows: list[PlannedLeaveDay], birth_date: date, fourth_birthday: date, savable_cap: int, latest_fetch: ParentalLeaveFetch | None, default_share: tuple[int, int] | None, ) -> list[ParentalLeaveYearBalance]: """Running FP (sjukpenning/lagsta) balance at each year's end. Starts from the latest snapshot's tier lefts (or, with no snapshot, this parent's default share — see `default_tier_pool_share`), then subtracts planned FP rows dated after the snapshot (or from the child's birth when there is none) cumulatively, year by year. The age-4 clamp (`resolve_age4_savable_cap`) is applied once, in the year of the 4th birthday, cutting lagsta first — mirrors `allocation_ service._Ledger._maybe_clamp_age4` exactly. """ if latest_fetch is not None: sjuk_left = latest_fetch.tier_data.sjukpenning_left lagsta_left = latest_fetch.tier_data.lagsta_niva_left cutoff = latest_fetch.fetched_at.date() has_snapshot = True else: assert default_share is not None sjuk_left, lagsta_left = default_share cutoff = birth_date has_snapshot = False used_by_year: dict[int, dict[str, int]] = {} deduct_by_year: dict[int, dict[str, int]] = {} for d in fp_rows: y = d.date.year bucket = used_by_year.setdefault(y, {"sjukpenning": 0, "lagsta": 0}) bucket[d.day_type] += 1 is_post_snapshot = d.date > cutoff if has_snapshot else d.date >= cutoff if is_post_snapshot: deduct = deduct_by_year.setdefault(y, {"sjukpenning": 0, "lagsta": 0}) deduct[d.day_type] += 1 result: list[ParentalLeaveYearBalance] = [] for year in years: used = used_by_year.get(year, {"sjukpenning": 0, "lagsta": 0}) deduct = deduct_by_year.get(year, {"sjukpenning": 0, "lagsta": 0}) sjuk_left = max(sjuk_left - deduct["sjukpenning"], 0) lagsta_left = max(lagsta_left - deduct["lagsta"], 0) if year == fourth_birthday.year: remaining = sjuk_left + lagsta_left capped = min(remaining, savable_cap) excess = remaining - capped if excess > 0: cut_lagsta = min(excess, lagsta_left) lagsta_left -= cut_lagsta excess -= cut_lagsta sjuk_left -= excess result.append( ParentalLeaveYearBalance( year=year, sjukpenning_used=used["sjukpenning"], lagsta_used=used["lagsta"], sjukpenning_left=sjuk_left, lagsta_left=lagsta_left, ) ) return result 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) # Combined (all parents) per-year totals — see `YearTotals`. totals_by_year: dict[int, list[int]] = {} # year -> [pl_left, vacation_left] for proj in projections: for yb in proj.year_balances: acc = totals_by_year.setdefault(yb.year, [0, 0]) acc[0] += yb.parental_leave.sjukpenning_left + yb.parental_leave.lagsta_left acc[1] += yb.vacation.left year_totals = [ YearTotals(year=year, pl_left=acc[0], vacation_left=acc[1]) for year, acc in sorted(totals_by_year.items()) ] 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, year_totals=year_totals, ) 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}, )