dodil/erp-brick-payroll
DODIL ERP brick: payroll
src/brick_payroll/application/gather.py
"""What a calculation reads, and how it becomes each person's input to gross to net.
`gather` asks the neighbours (hr's extract once per run, time-off's absences once per run, then per person only what that
person needs: the employment for the payee and cost centre, the work pattern for a part period, the holiday balance for a
leaver). `inputs` turns that and this brick's own rows into PersonInputs, each with a fingerprint of everything it was
built from - so the approval can tell, person by person, whether a payslip was built on yesterday's facts (PRL-25)."""
import hashlib
import json
from datetime import date, timedelta
from fractions import Fraction
from typing import Optional
from brick_payroll.application import statutory_pay as sp
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 ZERO, F, money, rate4
from brick_payroll.infrastructure import seams
SKIP_COMPONENTS = ("bank",) # a bank change is payments' business (HR-59), never a payslip line
def _sha(obj) -> str:
return hashlib.sha256(json.dumps(obj, sort_keys=True, default=str).encode()).hexdigest()
def gather(run: pr.PayRun, actor: str) -> dict:
"""{batch_id, state, people: {(person_id, n): {lines, starter, leaver, employment, absences, pattern, balance}}}."""
seams.produce_extract(run.company_code, run.period_code, actor)
ex = seams.extract(run.company_code, run.period_code, actor)
if ex is None:
raise RuleBroken(f"hr has no payroll extract {seams.batch_id(run.company_code, run.period_code)}", gate="extract",
fix="ask hr to check the legal entity and period")
people: dict = {}
for ln in ex.get("lines") or []:
if ln.get("component") in SKIP_COMPONENTS or ln.get("change_kind") == "bank_change":
continue
people.setdefault((int(ln["person_id"]), int(ln.get("n") or 1)), {"lines": []})["lines"].append(ln)
for s in ex.get("starters") or []:
people.setdefault((int(s["person_id"]), int(s.get("n") or 1)), {"lines": []})["starter"] = s
for s in ex.get("leavers") or []:
people.setdefault((int(s["person_id"]), int(s.get("n") or 1)), {"lines": []})["leaver"] = s
absences = [a for a in seams.absences(run.company_code, run.period_from, run.period_to, actor)
if a.get("status") in ("approved", "recorded", "closed")]
for key, p in sorted(people.items()):
pid = key[0]
last = date.fromisoformat(p["leaver"]["last_day"]) if p.get("leaver") else None
on = min(run.pay_date, last) if last else run.pay_date
p["employment"] = seams.employment(pid, on, actor)
p["absences"] = sorted((a for a in absences if int(a["person_id"]) == pid), key=lambda a: (a["absence_ref"], a.get("revision") or 0))
maternity = [a for a in p["absences"] if a.get("pay_treatment") == "statutory_maternity"]
if any(_part_period(run, ln) for ln in p["lines"]) or maternity:
p["pattern"] = [{**d, "date": d["date"].isoformat()} for d in seams.work_pattern(pid, run.period_from, run.period_to, actor)]
for a in p["absences"]:
if a.get("pay_treatment") == "sick" and a.get("sickness"): # the whole spell's pattern: waiting days and the week count start on its first day
first = date.fromisoformat(a["sickness"]["first_day"])
end = max(sp.spell_end(a), first + timedelta(days=6))
p.setdefault("spells", {})[a["absence_ref"]] = [{**d, "date": d["date"].isoformat()} for d in seams.work_pattern(pid, first, end, actor)]
for a in maternity:
p.setdefault("leave", {})[a["absence_ref"]] = _leave_facts(run, pid, a, actor)
if last:
p["balance"] = seams.leave_balance(pid, last, actor)
return {"batch_id": ex.get("batch_id") or seams.batch_id(run.company_code, run.period_code), "state": ex.get("state"), "people": people}
def _leave_facts(run: pr.PayRun, pid: int, a: dict, actor: str) -> dict:
"""Where a maternity leave began (time-off serves the days of a period only, so look back in windows of at most 93 days from the
earliest legal start, 11 weeks before the expected week) and whether she had 26 weeks' service by the qualifying week."""
ewc = date.fromisoformat((a.get("statutory") or {})["ewc"])
start, lo = None, ewc - timedelta(weeks=11)
while start is None and lo <= run.period_to:
hi = min(lo + timedelta(days=92), run.period_to)
for row in seams.absences(run.company_code, lo, hi, actor, person_id=pid):
days = [d["date"] for d in row.get("days") or []]
if row["absence_ref"] == a["absence_ref"] and days:
start = min(days)
lo = hi + timedelta(days=1)
q_end = st.qualifying_week(ewc)[1]
twenty_six = seams.employment(pid, q_end - timedelta(days=181), actor)
return {"start": start, "employed_26_weeks": bool(twenty_six.get("employed"))}
def _d(v) -> Optional[date]:
return date.fromisoformat(v) if isinstance(v, str) and v else (v if isinstance(v, date) else None)
def _part_period(run: pr.PayRun, ln: dict) -> bool:
lo, hi = _d(ln.get("prorate_from")) or run.period_from, _d(ln.get("prorate_to")) or run.period_to
return lo > run.period_from or hi < run.period_to
def payroll_id(person_id: int, n: int) -> str:
return f"{person_id}-{n}"
def _annual(ln: dict) -> Fraction:
basis = ln.get("rate_basis")
if basis == "year":
return F(ln["rate_amount"])
if basis == "month":
return F(ln["rate_amount"]) * 12
if basis == "week":
return F(ln["rate_amount"]) * 52
raise paye.UnsupportedHere(f"pay on a rate per {basis} needs approved hours from timesheets (PRL-22), which this brick does not read yet", gate="pay_basis")
def earnings(run: pr.PayRun, key: tuple, p: dict, elements: dict, divisor: int, deduct_overtaken: bool, rates: paye.Rates, history: dict,
notes: dict) -> list:
"""The person's gross lines: each salary slice prorated by working days, unpaid leave at the day rate, a leaver's
holiday in lieu. Anything this slice does not calculate is refused by name (UnsupportedHere)."""
who = payroll_id(*key)
by_map = {e["maps_from"]: e for e in elements.values() if e.get("maps_from")}
out = []
base = []
for ln in sorted(p["lines"], key=lambda x: (x.get("component") or "", x.get("prorate_from") or "")):
if ln.get("for_period") and ln["for_period"] != run.period_code:
raise paye.UnsupportedHere(f"{who}: a retro line for {ln['for_period']} is not calculated yet", gate="retro_line")
el = by_map.get(f"hr:{ln.get('component')}")
if el is None:
raise RuleBroken(f"{who}: hr's pay component {ln.get('component')!r} maps to no pay element", gate="element_mapped",
fix="add a pay element whose maps_from is hr:" + str(ln.get("component")))
if (ln.get("currency") or "GBP") != "GBP":
raise RuleBroken(f"{who}: pay in {ln.get('currency')} on a UK PAYE scheme", gate="currency")
lo, hi = max(_d(ln.get("prorate_from")) or run.period_from, run.period_from), min(_d(ln.get("prorate_to")) or run.period_to, run.period_to)
if _part_period(run, ln):
days = [{**d, "date": date.fromisoformat(d["date"])} for d in p.get("pattern") or []]
slice_wd, period_wd = pr.working_days(days, lo, hi), pr.working_days(days, run.period_from, run.period_to)
q, rate, amount = pr.salary_line(ln["rate_basis"], F(ln["rate_amount"]), run.frequency, slice_wd, period_wd)
ref = f"hr extract {seams.batch_id(run.company_code, run.period_code)} {ln.get('component')} {lo}..{hi}: {slice_wd} of {period_wd} working days"
else:
q, rate, amount = pr.salary_line(ln["rate_basis"], F(ln["rate_amount"]), run.frequency, None, None)
ref = f"hr extract {seams.batch_id(run.company_code, run.period_code)} {ln.get('component')} {lo}..{hi}"
base.append((lo, hi, _annual(ln)))
out.append(pr.Earning(el["element_code"], q, rate4(F(rate)), amount, "hr_extract", ref, el["taxable"], el["niable"], el["pensionable"]))
def day_rate(on: date) -> Fraction:
for lo, hi, annual in base:
if lo <= on <= hi:
return F(rate4(annual / divisor))
raise RuleBroken(f"{who}: an absence on {on} falls outside every pay slice hr gave", gate="absence_in_pay")
for a in p.get("absences") or []:
treatment = a.get("pay_treatment")
if treatment == "paid":
continue
if treatment == "sick" and a.get("sickness"):
out += sp.sick(run, who, p, a, by_map, rates, history, base, day_rate, notes)
continue
if treatment == "statutory_maternity":
out += sp.maternity(run, who, p, a, by_map, rates, history, base, notes)
continue
if treatment != "unpaid":
raise paye.UnsupportedHere(f"{who}: {treatment} absence {a['absence_ref']} needs statutory pay (SPP, SAP, ...), not calculated yet",
gate="statutory_pay")
el = by_map.get("tof:unpaid")
days = [(date.fromisoformat(d["date"]), Fraction(1) if (d.get("part") or "full") == "full" else Fraction(1, 2)) for d in a.get("days") or []]
days = [(d, w) for d, w in days if run.period_from <= d <= run.period_to]
if not days:
continue
rates = {day_rate(d) for d, _ in days}
if len(rates) != 1:
raise paye.UnsupportedHere(f"{who}: unpaid absence {a['absence_ref']} spans a pay change", gate="statutory_pay")
n_days = sum((w for _, w in days), Fraction(0))
r = rates.pop()
out.append(pr.Earning(el["element_code"], money(n_days), rate4(r), -money(n_days * r), "absence",
f"{a['absence_ref']} r{a.get('revision') or 1}", el["taxable"], el["niable"], el["pensionable"]))
if p.get("leaver"):
last = date.fromisoformat(p["leaver"]["last_day"])
bal = p.get("balance") or {}
if bal:
if bal.get("unit") not in (None, "days", "day"):
raise paye.UnsupportedHere(f"{who}: a holiday balance in {bal.get('unit')} is not calculated yet", gate="holiday_unit")
owed = F(bal.get("available") or 0)
if owed > 0 or (owed < 0 and deduct_overtaken):
el = by_map.get("tof:balance")
r = day_rate(last)
out.append(pr.Earning(el["element_code"], rate4(owed), rate4(r), money(owed * r), "absence", f"ANNUAL balance on {last}",
el["taxable"], el["niable"], el["pensionable"]))
return out
def inputs(run: pr.PayRun, gathered: dict, employees: dict, facts: dict, schemes: dict, elements: dict, ytd: dict,
rates_rows: list, divisor: int, deduct_overtaken: bool, history: dict) -> tuple:
"""(PersonInputs, missing): every person's input with its fingerprint, and the people with no tax code or NI category."""
rates_sha = _sha(sorted((r["rate_key"], r["region"], r["frequency"], str(r["valid_from"]), str(r["value"])) for r in rates_rows))
rates = paye.Rates(rates_rows)
out, missing = [], []
for key, p in sorted(gathered["people"].items()):
pid = payroll_id(*key)
emp = employees.get(pid)
f = facts.get(pid) or {}
tc, ni, mem = f.get("tax_code"), f.get("ni"), f.get("membership")
if emp is None or emp["status"] == "pending_details" or tc is None or ni is None:
missing.append(pid)
continue
after_p45 = emp["status"] == "left"
if after_p45: # a payment after the P45 (a late bonus) is taxed on 0T month 1 and reported with the leaving date
tc = {"tax_code": "0T", "basis": "week1_month1", "source": "after_p45", "valid_from": emp.get("leaving_date")}
scheme_row = schemes.get(mem["scheme_code"]) if mem else None
if mem and scheme_row is None:
raise RuleBroken(f"{pid}: pension scheme {mem['scheme_code']} is not in force on {run.pay_date}", gate="pension_scheme")
scheme = paye.Scheme(scheme_row["scheme_code"], scheme_row["basis"], scheme_row["employee_pct"], scheme_row["employer_pct"],
scheme_row["tax_relief"]) if scheme_row else None
parts = {"extract": _sha(p.get("lines")), "starter_leaver": _sha([p.get("starter"), p.get("leaver")]), "absences": _sha(p.get("absences")),
"employment": _sha({k: (p.get("employment") or {}).get(k) for k in ("employed", "company_code", "cost_centre", "bp_id")}),
"work_pattern": _sha([p.get("pattern"), p.get("spells"), p.get("leave")]), "holiday_balance": _sha(p.get("balance")),
"earnings_history": _sha([history["gross"].get(pid), history["ssp_days"].get(pid)]), "year_to_date": _sha(ytd.get(pid)),
"payroll_record": _sha([tc, ni, mem, scheme_row, emp.get("p45_pay_to_date"), emp.get("p45_tax_to_date")]), "rates": rates_sha}
cumulative = tc["basis"] == "cumulative"
employment = p.get("employment") or {}
notes: dict = {}
if tc["source"] == "emergency":
notes["warnings"] = ["no starter statement or P45 yet: the emergency code is used until HMRC sends one"]
if after_p45:
notes["warnings"] = [*notes.get("warnings", []), f"paid after the P45 (left {emp.get('leaving_date')}): taxed on 0T month 1"]
left_on = date.fromisoformat(p["leaver"]["last_day"]) if p.get("leaver") else (emp.get("leaving_date") if after_p45 else None)
inp = pr.PersonInput(
payroll_id=pid, person_id=key[0], earnings=earnings(run, key, p, elements, divisor, deduct_overtaken, rates, history, notes), notes=notes,
tax_code=paye.parse_tax_code(tc["tax_code"]), basis=tc["basis"], ni_category=ni["category"], scheme=scheme,
prev=ytd.get(pid) or {}, p45_pay=money(emp.get("p45_pay_to_date") or 0) if cumulative else ZERO,
p45_tax=money(emp.get("p45_tax_to_date") or 0) if cumulative else ZERO, cost_centre=employment.get("cost_centre") or "",
bp_id=int(employment["bp_id"]) if employment.get("bp_id") is not None else None,
leaving_date=left_on, inputs_sha=_sha(parts), parts=parts)
out.append(inp)
return out, missing
def stale_causes(run: pr.PayRun, fresh: list, missing: list) -> dict:
"""payroll_id -> what changed since the calculation, for every payslip whose inputs differ (and every person who joined or
left hr's extract since)."""
now = {i.payroll_id: i for i in fresh}
causes = {}
for p in run.payslips:
f = now.get(p.payroll_id)
if f is None:
causes[p.payroll_id] = "no tax details any more" if p.payroll_id in missing else "no longer in hr's extract"
elif f.inputs_sha != p.inputs_sha:
old = (p.ytd or {}).get("inputs") or {}
changed = [k for k, v in f.parts.items() if old.get(k) != v] or ["inputs"]
causes[p.payroll_id] = "changed since the calculation: " + ", ".join(changed)
have = {p.payroll_id for p in run.payslips}
for pid in sorted(set(now) | set(missing)):
if pid not in have:
causes[pid] = "joined hr's extract since the calculation"
return causes