dodil/erp-brick-payroll
DODIL ERP brick: payroll
src/brick_payroll/application/statutory_pay.py
"""Statutory sick and maternity pay as payslip lines: turns the facts time-off served (and hr's work pattern) into SICK_DEDUCTION +
SSP and MATERNITY_DEDUCTION + SMP lines by domain/statutory.py, with every line naming its absence, revision and rate rows.
Average weekly earnings come from this payroll's own approved payslips in the relevant period; a person with none there (a new
starter, a migrated employee) is averaged from the contracted pay hr gave (annual / 52) and the payslip says so (`awe_basis`).
That fallback is a choice, written in AGENTS.md."""
from datetime import date, timedelta
from fractions import Fraction
from typing import Callable
from brick_payroll.domain import pay_run as pr
from brick_payroll.domain import paye
from brick_payroll.domain import statutory as st
from brick_payroll.domain.errors import RuleBroken
from brick_payroll.domain.money import F, money, rate4
def _iso(v) -> date:
return v if isinstance(v, date) else date.fromisoformat(v)
def spell_end(a: dict) -> date:
"""The last day a sickness covers as far as we know: its last day off, else its last day in the days served."""
s = a["sickness"]
if s.get("last_day_off"):
return _iso(s["last_day_off"])
days = [_iso(d["date"]) for d in a.get("days") or []]
return max(days) if days else _iso(s["first_day"])
def history_window(people: dict) -> tuple:
"""(payroll person ids, from, to) the history of gross pay must cover for every statutory absence in the gathered people,
or None: sickness looks back 8 weeks from its first day, maternity from the qualifying week's relevant period."""
ids, lo, hi = [], None, None
for key, p in people.items():
for a in p.get("absences") or []:
if a.get("pay_treatment") == "sick" and a.get("sickness"):
first = _iso(a["sickness"]["first_day"])
span = (first - timedelta(days=56), first - timedelta(days=1))
elif a.get("pay_treatment") == "statutory_maternity" and (a.get("statutory") or {}).get("ewc"):
span = st.relevant_period(_iso(a["statutory"]["ewc"]))
else:
continue
ids.append(int(key[0]))
lo, hi = min(lo or span[0], span[0]), max(hi or span[1], span[1])
return sorted(set(ids)), lo, hi
def _awe(paydays: list, frequency: str, annual: Fraction):
got = st.average_weekly_earnings(paydays, frequency)
if got is not None:
return got, "payslips"
return annual / 52, "contract"
def _needs(by_map: dict, key: str, who: str) -> dict:
el = by_map.get(key)
if el is None:
raise RuleBroken(f"{who}: no pay element maps {key}", gate="element_mapped", fix=f"add a pay element whose maps_from is {key}")
return el
def _earning(el: dict, qty, rate, amount, source_kind: str, ref: str, used: str = "") -> pr.Earning:
return pr.Earning(el["element_code"], qty, rate, amount, source_kind, ref, el["taxable"], el["niable"], el["pensionable"], used)
def sick(run: pr.PayRun, who: str, p: dict, a: dict, by_map: dict, rates: paye.Rates, history: dict, base: list, day_rate: Callable, notes: dict) -> list:
"""The sickness deduction on every qualifying day of the spell in the period, and SSP on the payable ones."""
s = a["sickness"]
first, last = _iso(s["first_day"]), spell_end(a)
pattern = [{**d, "date": _iso(d["date"])} for d in (p.get("spells") or {}).get(a["absence_ref"]) or []]
qualifying = sorted(d["date"] for d in pattern if d.get("employed") and int(d.get("contracted_minutes") or 0) > 0 and first <= d["date"] <= last)
week = [d["date"] for d in pattern if d.get("employed") and int(d.get("contracted_minutes") or 0) > 0 and first <= d["date"] < first + timedelta(days=7)]
group = s.get("linked_group") or a["absence_ref"]
pid = who
paydays = [(d, g) for d, g in (history["gross"].get(pid) or []) if first - timedelta(days=56) <= d < first]
annual = next((x[2] for x in base if x[0] <= first <= x[1]), base[0][2] if base else Fraction(0))
awe, basis = _awe(paydays, run.frequency, annual)
rates.used = set()
facts = st.SspFacts(a["absence_ref"], int(a.get("revision") or 1), first, last, bool(s.get("piw")), bool(s.get("linked_to")), group,
bool(a.get("evidence_ok", True)), qualifying, len(week), int((history["ssp_days"].get(pid) or {}).get(group, 0)))
res = st.ssp(facts, run.period_from, run.period_to, awe, rates)
used = ";".join(sorted(rates.used))
ref = f"{a['absence_ref']} r{a.get('revision') or 1}"
out = []
if res.deduct_days:
rs = {day_rate(d) for d in res.deduct_days}
if len(rs) != 1:
raise paye.UnsupportedHere(f"{who}: sickness {a['absence_ref']} spans a pay change", gate="statutory_pay")
r = rs.pop()
n = Fraction(len(res.deduct_days))
out.append(_earning(_needs(by_map, "tof:sick", who), money(n), rate4(r), -money(n * r), "absence", ref))
if res.paid_days:
daily = rate4(res.daily)
n = len(res.paid_days)
out.append(_earning(_needs(by_map, "tof:ssp", who), money(n), daily, money(F(daily) * n), "statutory", f"{ref} group {group}", used))
notes.setdefault("ssp", []).append({**res.notes, "awe_basis": basis, "reason": res.reason or None, "from": min(res.paid_days).isoformat() if res.paid_days else None,
"to": max(res.paid_days).isoformat() if res.paid_days else None, "pay_amount": str(sum((o.amount for o in out if o.element_code == "SSP"), money(0)))})
return out
def maternity(run: pr.PayRun, who: str, p: dict, a: dict, by_map: dict, rates: paye.Rates, history: dict, base: list, notes: dict) -> list:
"""Salary off for the days on leave (pro rata, so a whole month on leave is 0.00) and SMP on the calendar days of the period."""
sf = a.get("statutory") or {}
if not sf.get("ewc"):
raise paye.UnsupportedHere(f"{who}: maternity {a['absence_ref']} carries no expected week of childbirth", gate="statutory_pay")
days = sorted(_iso(d["date"]) for d in a.get("days") or [] if run.period_from <= _iso(d["date"]) <= run.period_to)
if not days:
return []
ewc = _iso(sf["ewc"])
leave = (p.get("leave") or {}).get(a["absence_ref"]) or {}
start = _iso(leave["start"]) if leave.get("start") else days[0]
pattern = [{**d, "date": _iso(d["date"])} for d in p.get("pattern") or []]
tail = [d["date"] for d in pattern if d["date"] > days[-1]]
last = run.period_to if all(int(next(x for x in pattern if x["date"] == t).get("contracted_minutes") or 0) == 0 for t in tail) else days[-1]
lo, hi = st.relevant_period(ewc)
paydays = [(d, g) for d, g in (history["gross"].get(who) or []) if lo <= d <= hi]
annual = next((x[2] for x in base if x[0] <= days[0] <= x[1]), base[0][2] if base else Fraction(0))
awe, basis = _awe(paydays, run.frequency, annual)
rates.used = set()
facts = st.SmpFacts(a["absence_ref"], int(a.get("revision") or 1), ewc, _iso(sf["evidence_received_on"]) if sf.get("evidence_received_on") else None,
start, last, bool(leave.get("employed_26_weeks")))
res = st.smp(facts, run.period_from, run.period_to, awe, rates)
used = ";".join(sorted(rates.used))
ref = f"{a['absence_ref']} r{a.get('revision') or 1}"
working = pr.working_days(pattern, run.period_from, run.period_to)
if not working:
raise RuleBroken(f"{who}: hr's work pattern has no working day in the period, so salary on maternity leave cannot be prorated", gate="work_pattern")
per = annual / paye.PERIODS[run.frequency]
out = [_earning(_needs(by_map, "tof:statutory_maternity", who), money(len(days)), rate4(per), -money(per * Fraction(len(days), working)), "absence", ref)]
if res.amount:
total = money(res.amount)
out.append(_earning(_needs(by_map, "tof:smp", who), money(res.days), rate4(F(total) / res.days), total, "statutory", ref, used))
notes.setdefault("smp", []).append({**res.notes, "awe_basis": basis, "eligible": res.eligible, "reason": res.reason or None, "days": res.days,
"weeks_in_period": res.weeks_in_period, "pay_amount": str(money(res.amount))})
return out