"""Domain services: rules that need more than one aggregate's state to answer, and change none.
ManagerOf (person, date) → the derived manager: the chief of the assignment's unit, walking up; the
override wins when set. Never stored.
FteOf (assignment, date) → schedule minutes ÷ the schedule's full-time base × employment %.
Headcount (date or period, grouping) → headcount, FTE, starters, leavers, derived from slices.
person_card what the business partner may know about a person: name and work contact, roles, works_for.
notice_warning the UK pack's statutory-minimum-notice check (warned, never refused).
"""
from dataclasses import dataclass
from datetime import date
from decimal import Decimal
from typing import Callable, Iterable, Optional
from brick_hr.domain.configuration import Config
from brick_hr.domain.org import Org
from brick_hr.domain.person import Person
@dataclass(frozen=True)
class Seat:
"""One person's assignment on a date, as the org services need it (a read model, not the aggregate)."""
person_id: int
n: int
valid_from: date
valid_to: date
org_id: Optional[str]
org_unit_id: str
position_id: Optional[str]
job_code: Optional[str]
location_code: Optional[str]
schedule_code: Optional[str]
employment_pct: Optional[Decimal]
manager_override_person_id: Optional[int]
worker_type: str
cost_centre: Optional[str] = None
# hr v4 (HR-63, HR-64): the engagement's company and dates, carried on each of its seats
company_code: Optional[str] = None
employed_from: Optional[date] = None
employed_to: Optional[date] = None
def covers(self, d: date) -> bool:
return self.valid_from <= d <= self.valid_to
class Directory:
"""Every person's active assignment slices (Seats), for the services that look across people."""
def __init__(self, seats: Iterable[Seat]):
self.seats = list(seats)
def seat(self, person_id: int, d: date) -> Optional[Seat]:
return next((s for s in self.seats if s.person_id == person_id and s.covers(d)), None)
def seats_on(self, d: date) -> list:
return [s for s in self.seats if s.covers(d)]
def holders(self, position_id: str, d: date) -> list:
return sorted(s.person_id for s in self.seats if s.position_id == position_id and s.covers(d))
def positions_held(self, person_id: int, d: date) -> set:
return {s.position_id for s in self.seats if s.person_id == person_id and s.covers(d) and s.position_id}
def manager_of(person_id: int, d: date, directory: Directory, orgs: dict) -> Optional[int]:
"""ManagerOf: the holder of the chief position of the person's unit on `d`, walking up while a unit has no
chief, a vacant one, or the person themselves. A manager override on the assignment wins."""
seat = directory.seat(person_id, d)
if seat is None:
return None
if seat.manager_override_person_id:
return int(seat.manager_override_person_id)
org: Optional[Org] = orgs.get(seat.org_id) if seat.org_id else None
if org is None:
return None
for unit in org.ancestors(seat.org_unit_id, d):
if not unit.chief_position_id:
continue
holders = [h for h in directory.holders(unit.chief_position_id, d) if h != person_id]
if holders:
return holders[0]
return None
def line_of(manager_id: int, today: date, directory: Directory, orgs: dict, people: Iterable[int]) -> set:
"""The people whose derived manager `manager_id` is today, or on a future slice of their own assignment —
so a manager sees the starter they hired before the start date, and a reorganisation that hands someone to
them is visible from its date and not before."""
out = set()
for pid in people:
dates = {today} | {s.valid_from for s in directory.seats if s.person_id == pid and s.valid_from > today}
if any(manager_of(pid, d, directory, orgs) == manager_id for d in dates):
out.add(pid)
return out
def fte_of(seat: Seat, config: Config) -> Decimal:
"""FteOf: schedule minutes ÷ the full-time base × employment %, to 4 places."""
pct = Decimal(seat.employment_pct if seat.employment_pct is not None else 100) / Decimal(100)
sched = config.schedule(seat.schedule_code) if seat.schedule_code else None
ratio = Decimal(1)
if sched and int(sched.get("full_time_weekly_minutes") or 0):
ratio = Decimal(int(sched["weekly_minutes"])) / Decimal(int(sched["full_time_weekly_minutes"]))
return (ratio * pct).quantize(Decimal("0.0001"))
def headcount(people: Iterable[Person], on: date, config: Config, orgs: dict, group_by: str = "org_unit",
only: Optional[set] = None) -> dict:
"""Headcount and FTE on a date, grouped by org unit, worker type or both — as the org was THEN (HR-18)."""
groups: dict = {}
total, fte_total = 0, Decimal(0)
for p in people:
if only is not None and p.person_id not in only:
continue
at = p.assignment_at(on)
if at is None:
continue
n, s = at
e = p.engagements[n]
seat = _seat(p, n, s)
key = _group_key(group_by, s.data.get("org_unit_id"), e.worker_type)
g = groups.setdefault(key, {"headcount": 0, "fte": Decimal(0)})
f = fte_of(seat, config)
g["headcount"] += 1
g["fte"] += f
total += 1
fte_total += f
return {"on": on.isoformat(), "group_by": group_by, "headcount": total, "fte": str(fte_total),
"groups": [{"group": k, "headcount": v["headcount"], "fte": str(v["fte"])}
for k, v in sorted(groups.items())]}
def _group_key(group_by: str, unit: Optional[str], worker_type: str) -> str:
if group_by == "worker_type":
return worker_type
if group_by == "both":
return f"{unit}|{worker_type}"
return unit or ""
def starters_and_leavers(people: Iterable[Person], start: date, end: date, group_by: str = "org_unit",
only: Optional[set] = None) -> dict:
"""Starters: engagements starting in the period. Leavers: actual_end in the period. Derived from slices."""
starters: dict = {}
leavers: dict = {}
for p in people:
if only is not None and p.person_id not in only:
continue
for e in p.engagements.values():
if start <= e.start_date <= end:
a = p.assignments[e.n].at(e.start_date)
k = _group_key(group_by, a.data.get("org_unit_id") if a else None, e.worker_type)
starters.setdefault(k, []).append(p.person_id)
if e.actual_end is not None and start <= e.actual_end <= end:
a = p.assignments[e.n].at(e.actual_end)
k = _group_key(group_by, a.data.get("org_unit_id") if a else None, e.worker_type)
leavers.setdefault(k, []).append(p.person_id)
return {"from": start.isoformat(), "to": end.isoformat(), "group_by": group_by,
"starters": sum(len(v) for v in starters.values()), "leavers": sum(len(v) for v in leavers.values()),
"groups": [{"group": k, "starters": len(starters.get(k, [])), "leavers": len(leavers.get(k, []))}
for k in sorted(set(starters) | set(leavers))]}
def _seat(p: Person, n: int, s) -> Seat:
d = s.data
e = p.engagements[n]
pct = d.get("employment_pct")
return Seat(p.person_id, n, s.valid_from, s.valid_to, d.get("org_id"), d.get("org_unit_id"),
d.get("position_id"), d.get("job_code"), d.get("location_code"), d.get("schedule_code"),
Decimal(str(pct)) if pct is not None else None,
int(d["manager_override_person_id"]) if d.get("manager_override_person_id") else None,
e.worker_type, d.get("cost_centre"), e.legal_entity_ref, e.start_date, e.actual_end or e.planned_end)
def seats_of(p: Person) -> list:
out = []
for n, t in p.assignments.items():
out += [_seat(p, n, s) for s in t.active()]
return out
# ── the business partner's view of a person (HR-50) ─────────────────────────────────────────────
PRIVATE = ("personal_email", "personal_phone")
def person_card(p: Person, today: date, config: Config, country: str = "") -> dict:
"""The ONLY fields a person's business partner carries: legal and preferred name, work email and phone,
the roles of the engagement active today, and the company they work through. Never a personal contact,
address, NI number, bank detail or right-to-work field — those are not even in reach here."""
d = p.detail_at(today)
data = d.data if d else {}
active = p.engagement_on(today)
open_e = p.open_engagement(today)
roles = []
if active is not None:
wt = config.worker_type(active.worker_type) or {}
roles = sorted(set(wt.get("bp_roles") or []))
works_for = (open_e or active).supplier_bp_id if (open_e or active) else None
legal = " ".join(x for x in (data.get("legal_given_name"), data.get("legal_family_name")) if x)
return {"person_id": str(p.person_id), "legal_name": legal, "preferred_name": data.get("preferred_name") or "",
"work_email": data.get("work_email") or "", "work_phone": data.get("work_phone") or "",
"country": country, "roles": roles, "works_for": int(works_for) if works_for else None}
# ── the UK pack's notice check (Employment Rights Act 1996 s.86) ─────────────────────────────────
def statutory_notice_weeks(start: date, last_day: date) -> int:
"""At least one week after one month's employment; one week per complete year from two years, up to 12."""
months = (last_day.year - start.year) * 12 + (last_day.month - start.month) - (1 if last_day.day < start.day else 0)
if months < 1:
return 0
years = months // 12
if years < 2:
return 1
return min(years, 12)
def notice_warning(start: date, last_day: date, told_on: date) -> Optional[str]:
weeks = statutory_notice_weeks(start, last_day)
given = (last_day - told_on).days
if weeks and given < weeks * 7:
return (f"the last day gives {given} day(s) of notice; the statutory minimum for this length of service is "
f"{weeks} week(s) (Employment Rights Act 1996 s.86) — the contract may say more and the employee "
"may waive it: check before you rely on it")
return None
__all__ = ["Seat", "Directory", "manager_of", "line_of", "fte_of", "headcount", "starters_and_leavers",
"seats_of", "person_card", "notice_warning", "statutory_notice_weeks", "Callable"]