dodil/erp-brick-time-off
DODIL ERP brick: time off
src/brick_time_off/domain/calendar.py
"""Days, half days and the leave year: what a request costs and what a person is entitled to. Plain arithmetic over what hr says
about each day (employed, contracted minutes, public holiday, FTE) - this brick keeps no calendar, pattern or staff list.
A DAY has two HALVES (am, pm); a half is what a person races for (tof_day_claim). A half costs 0.5 days for someone who books in
days and its minutes / 60 hours for someone who books in hours. A day that is not worked (a weekend, a day off in the part-time
pattern) or is a public holiday has no halves: it costs nothing and claims nothing.
"""
from dataclasses import dataclass
from datetime import date, timedelta
from decimal import ROUND_CEILING, ROUND_HALF_UP, Decimal
from typing import Optional
from brick_time_off.domain.errors import RuleBroken
ZERO = Decimal("0")
Q4 = Decimal("0.0001")
Q2 = Decimal("0.01")
HALF = Decimal("0.5")
PARTS = ("am", "pm")
MAX_SPAN_DAYS = 366 # a request covers at most a year; hr answers a range of at most 400 days
def q4(v) -> Decimal:
return Decimal(str(v)).quantize(Q4, rounding=ROUND_HALF_UP)
def show(v, unit: str) -> str:
"""An exact quantity as a string: days to two places, hours to two places and more only when they are there (3.7500 -> 3.75)."""
d = q4(v)
if unit == "days":
return f"{d.quantize(Q2, rounding=ROUND_HALF_UP):.2f}"
text = f"{d:.4f}"
while text.endswith("0") and len(text.split(".")[1]) > 2:
text = text[:-1]
return text
@dataclass(frozen=True)
class Day:
"""hr's answer for one date (GET /persons/{id}/work-pattern)."""
date: date
employed: bool
minutes: int
holiday: Optional[str]
fte: Decimal
@dataclass(frozen=True)
class Slot:
"""One half day of a request: the row it claims."""
date: date
part: str
minutes: int
quantity: Decimal
def parts_of(first: date, last: date, from_part: str, to_part: str, on: date) -> tuple:
"""Which halves of `on` a request takes. One day: the part asked (full = both). A longer request starts on its first day
(full, or pm = from lunchtime) and ends on its last (full, or am = until lunchtime); every day between is whole."""
if first == last:
return PARTS if from_part == "full" else (from_part,)
if on == first:
return PARTS if from_part == "full" else ("pm",)
if on == last:
return PARTS if to_part == "full" else ("am",)
return PARTS
def check_parts(first: date, last: date, from_part: str, to_part: str, min_booking: str = "half_day") -> None:
"""The shape of the ask: a bounds error is the caller's, so it is not a RuleBroken."""
if last < first:
raise ValueError("the last day is before the first")
if (last - first).days >= MAX_SPAN_DAYS:
raise ValueError(f"a request covers at most {MAX_SPAN_DAYS} days")
if first == last:
if from_part != to_part:
raise ValueError("for one day, from_part and to_part are the same: full, am or pm")
else:
if from_part not in ("full", "pm"):
raise ValueError("a request that goes on starts the first day whole (full) or from lunchtime (pm)")
if to_part not in ("full", "am"):
raise ValueError("a request that goes on ends the last day whole (full) or at lunchtime (am)")
if min_booking == "full_day" and ((first == last and from_part != "full") or (first != last and (from_part == "pm" or to_part == "am"))):
raise ValueError("this company books whole days only (min_booking full_day)")
def slots_of(days: list, first: date, last: date, from_part: str, to_part: str, unit: str) -> list:
"""The halves a request takes, from hr's answer for its range. Only a day that is worked and not a public holiday has halves."""
out = []
for d in days:
if not first <= d.date <= last or d.minutes <= 0 or d.holiday:
continue
am = d.minutes // 2
for part in parts_of(first, last, from_part, to_part, d.date):
minutes = am if part == "am" else d.minutes - am
out.append(Slot(d.date, part, minutes, HALF if unit == "days" else q4(Decimal(minutes) / 60)))
return out
def cost_of(slots: list) -> Decimal:
return q4(sum((s.quantity for s in slots), ZERO))
def unemployed_day(days: list, first: date, last: date) -> Optional[date]:
"""The first date of the range on which hr says the person is not employed."""
for d in days:
if first <= d.date <= last and not d.employed:
return d.date
return None
# ── the leave year ────────────────────────────────────────────────────────────────────────────────
def leave_year(on: date, start: str = "01-01") -> tuple:
"""(label, first day, last day) of the leave year `on` falls in; `start` is MM-DD. The label is the year it starts in."""
try:
month, dom = (int(x) for x in start.split("-"))
first = date(on.year, month, dom)
except (ValueError, TypeError):
raise RuleBroken(f"the policy's leave year start {start!r} is not MM-DD") from None
if on < first:
first = date(on.year - 1, month, dom)
nxt = date(first.year + 1, month, dom)
return first.year, first, nxt - timedelta(days=1)
# ── entitlement: derived on read, never stored (TOF-15) ─────────────────────────────────────────────
@dataclass(frozen=True)
class Entitlement:
entitled: Decimal
unit: str
day_minutes: int
share: Decimal # employed FTE-days / days in the leave year
employed_days: int
holidays: int # public holidays inside the employed span
holiday_deduction: Decimal # a policy with the holidays INSIDE the allowance: those on a day the person works
policy_code: str = ""
def pick_policy(policies: list, fte: Decimal) -> Optional[dict]:
"""The policy whose `applies_to` fits the person's FTE on the first day of the leave year (fte_min / fte_below)."""
import json
for p in policies:
rule = json.loads(p.get("applies_json") or "{}")
if "fte_min" in rule and fte < Decimal(rule["fte_min"]):
continue
if "fte_below" in rule and fte >= Decimal(rule["fte_below"]):
continue
return p
return None
def entitlement(policy: dict, days: list, first: date, last: date, last_day: Optional[date] = None) -> Entitlement:
"""What the policy gives for the leave year [first, last], from hr's days. Prorated by calendar days employed and by FTE on each,
so a starter, a leaver and a change of FTE are right with nobody recalculating:
days policy: annual_days x (sum of FTE over employed days / days in the year), rounded up to the next half day
hours policy: (annual_days + holidays when included) x the contracted day x the same share
A leaver's entitlement runs to the last day and is never rounded up (to two decimals; hours to four)."""
n_year = (last - first).days + 1
live = [d for d in days if first <= d.date <= last and d.employed and (last_day is None or d.date <= last_day)]
share = sum((d.fte for d in live), ZERO) / n_year
day_minutes = max((d.minutes for d in live), default=0)
holidays = [d for d in live if d.holiday]
base = Decimal(str(policy["annual_days"]))
included = policy["bank_holidays"] == "included"
if included:
base += len(holidays)
unit = policy["unit"]
raw = base * share if unit == "days" else base * Decimal(day_minutes) / 60 * share
if last_day is not None:
ent = raw.quantize(Q2 if unit == "days" else Q4, rounding=ROUND_HALF_UP)
elif policy["rounding"] == "half_day_up":
step = HALF if unit == "days" else Decimal(day_minutes) / 60 / 2 or HALF
ent = (raw / step).to_integral_value(rounding=ROUND_CEILING) * step
else:
ent = raw
deduction = ZERO
if included:
deduction = sum((Decimal(d.minutes) / 60 if unit == "hours" else (Decimal(1) if d.minutes > 0 else ZERO)
for d in holidays if d.minutes > 0), ZERO)
return Entitlement(q4(ent), unit, day_minutes, share, len(live), len(holidays), q4(deduction), policy["policy_code"])
# ── the balance: the entitlement plus a pure sum of entries ─────────────────────────────────────────
AGAINST = ("take", "reverse", "closure") # what was spent (negative) and given back (positive)
OPENING = ("adjust", "opening", "accrue") # what was added or taken by hand, at the opening, or earned
# "carry" is the year's days brought in; "lapse" (the part of a year's rest above the cap, written in THAT year) and "expire" (carried days unused by
# their date, written in the NEW year) take days away and are shown apart, so a person sees what the year end did
@dataclass(frozen=True)
class Balance:
unit: str
entitled: Decimal
carried: Decimal
adjustments: Decimal
lapsed: Decimal
expired: Decimal
holiday_deduction: Decimal
closures: Decimal
closure_details: tuple
taken: Decimal
booked: Decimal
pending: Decimal
available: Decimal
def closure_deductions(closure_rows: list, days: list, holder: dict, unit: str, last_day: Optional[date] = None) -> tuple:
"""(what the company's closure days cost this person, details): each closure day on which hr says the person is employed and works, less the
halves some absence already holds - `holder` is {(date, part): absence_ref} of every claim the person has - so a day already in their own request
(or sickness) is shown as COVERED by it and never deducted twice. A day after a leaver's last day is not theirs. Derived on read, never stored."""
by_date = {d.date: d for d in days}
total, details = ZERO, []
for row in closure_rows:
d, day = row["closure_date"], by_date.get(row["closure_date"])
if day is None or not day.employed or (last_day is not None and d > last_day):
continue
slots = slots_of([day], d, d, "full", "full", unit)
if not slots:
continue
free = [s for s in slots if (s.date, s.part) not in holder]
cost = sum((s.quantity for s in free), ZERO)
total += cost
covered = sorted({holder[(s.date, s.part)] for s in slots if (s.date, s.part) in holder})
details.append({"date": d, "name": row["name"], "cost": q4(cost), "covered_by": covered})
return q4(total), tuple(details)
def balance(ent: Entitlement, entries: list, claims: list, on: date, closures: tuple = (ZERO, ())) -> Balance:
"""`entries`: [(kind, signed quantity)] of the person, type and leave year, in ANY order. `claims`: [(date, quantity, status)] of the
person's requests of this type and year. Available is the entitlement plus the sum of the entries, less what is still pending;
`taken` and `booked` only split what the ledger says was spent at the business date."""
carried = sum((q for k, q in entries if k == "carry"), ZERO)
adjustments = sum((q for k, q in entries if k in OPENING), ZERO)
spent = -sum((q for k, q in entries if k in AGAINST), ZERO)
taken = sum((q for d, q, s in claims if s == "approved" and d <= on), ZERO)
taken = min(taken, spent) if spent >= 0 else taken
pending = sum((q for _, q, s in claims if s == "requested"), ZERO)
lapsed = sum((q for k, q in entries if k == "lapse"), ZERO)
expired = sum((q for k, q in entries if k == "expire"), ZERO)
avail = ent.entitled + carried + adjustments + lapsed + expired - ent.holiday_deduction - closures[0] - spent - pending
return Balance(ent.unit, ent.entitled, carried, adjustments, lapsed, expired, ent.holiday_deduction, closures[0], closures[1], q4(taken), q4(spent - taken),
q4(pending), q4(avail))
# ── the year end (TOF-17) ───────────────────────────────────────────────────────────────────────
def expiry_date(first: date, mmdd: str) -> date:
"""The day carried days expire in the leave year that starts on `first`: the next `mmdd` on or after it (a year starting 1 April with 03-31 expires
them a year on)."""
try:
month, dom = (int(x) for x in mmdd.split("-"))
day = date(first.year, month, dom)
except (ValueError, TypeError):
raise RuleBroken(f"carry_over_expires {mmdd!r} is not MM-DD") from None
return day if day >= first else date(first.year + 1, month, dom)
def carry_split(available: Decimal, cap: Decimal) -> tuple:
"""(carried, lapsed) of what is left at a year's end: never negative, never over the cap; what is over the cap lapses."""
left = max(available, ZERO)
carry = min(left, max(cap, ZERO))
return q4(carry), q4(left - carry)
def unused_carry(carried: Decimal, spent_by_expiry: Decimal) -> Decimal:
"""Carried days are used FIRST: what was spent by the expiry date comes off the carried days before anything else, and what is left expires."""
return q4(max(carried - max(min(spent_by_expiry, carried), ZERO), ZERO))