"""Reports (HR-18): starters, leavers, headcount and FTE by team and worker type, for any date — derived from
the slices on read (no snapshot table), so a past month answers as the org was then. A manager sees their own
line only. The onboarding monitor (HR-29): lead time, overdue steps, time to first access, deviations per
playbook version — derived from runs and steps on read, over the runs the caller may see."""
from datetime import date, datetime
from typing import Literal, Optional
from erp_platform.bricks import Refused
from brick_hr.application import support as S
from brick_hr.application import use_cases as uc
from brick_hr.domain import monitoring, services
from brick_hr.domain.payroll import period_bounds
HR = uc.HR
Group = Literal["org_unit", "worker_type", "both"]
def _scope(view: S.View) -> Optional[set]:
if view.sees_all:
return None if view.companies is None else view.people_in_companies()
return {p for p in view.line if view.person_in_companies(p)} | ({view.me} if view.me else set())
@HR.use_case(kind="query", route="GET /reports/headcount", permission=S.REPORT_READ)
def headcount(caller: dict, on: Optional[S.Day] = None, group_by: Group = "org_unit") -> dict:
"""Headcount and FTE on a date (today by default), by org unit, worker type or both — as the org was on that
date. FTE = schedule minutes ÷ full-time base × employment %."""
with S.rules():
with uc.repository() as uow:
view = S.View(uow, caller)
return services.headcount(uow.persons.load_many(), on or view.today, view.config, view.orgs, group_by,
_scope(view))
@HR.use_case(kind="query", route="GET /reports/starters-and-leavers", permission=S.REPORT_READ)
def starters_and_leavers(caller: dict, period: Optional[S.Code] = None, start: Optional[S.Day] = None,
end: Optional[S.Day] = None, group_by: Group = "org_unit") -> dict:
"""Starters (engagements starting in the period) and leavers (last days in the period), by org unit, worker
type or both, with the headcount and FTE at the period's end. A period is YYYY-MM, or give start and end."""
with S.rules():
if period:
start, end = period_bounds(period)
if start is None or end is None:
raise Refused(422, "give a period (YYYY-MM) or a start and an end")
with uc.repository() as uow:
view = S.View(uow, caller)
people = uow.persons.load_many()
scope = _scope(view)
out = services.starters_and_leavers(people, start, end, group_by, scope)
out["at_end"] = services.headcount(people, end, view.config, view.orgs, group_by, scope)
return out
Direction = Literal["onboarding", "offboarding", "change", "conversion"]
@HR.use_case(kind="query", route="GET /reports/onboarding", permission=S.RUN_MONITOR)
def onboarding_monitor(caller: dict, direction: Direction = "onboarding", since: Optional[S.Day] = None) -> dict:
"""Monitor the onboarding programme (HR-29, APQC 7.3.1.3): run lead time (median and longest, in days, and the
age of the runs still open), steps overdue by owner group and by team, the days from the start date to the
first access granted (negative = before day one), and per playbook version the runs that used it and its
deviations from the standard (variation / warning). Over the runs you may see (a manager: their line);
`since` keeps the runs created on or after a date. Read-only: everything is derived from the runs."""
with S.rules():
with uc.repository() as uow:
view = S.View(uow, caller)
runs = [r for r in uow.runs.load_many() if r.direction == direction and view.sees_person(r.person_id)
and (since is None or monitoring.business_date(r.created_at, view.tz) >= since)]
seats: dict = {}
for st in uow.persons.seats():
seats.setdefault((st.person_id, st.n), []).append(st)
def facts(r):
start = r.anchors.get("start_date")
start = start.date() if isinstance(start, datetime) else start
rows = sorted(seats.get((r.person_id, r.n), []), key=lambda x: x.valid_from)
on = next((x for x in rows if start and x.covers(start)), rows[0] if rows else None)
return monitoring.RunFacts(start, on.org_unit_id if on else None)
playbooks = uow.playbooks.versions({(r.playbook_id, r.playbook_version) for r in runs if r.playbook_id})
from brick_hr.application import process
from brick_hr.domain import deviations as dv
def deviations(pid, ver):
pb = playbooks.get((pid, ver))
if pb is None or process.reference_for(view.config, pb) is None:
return None
return dv.summary(process.deviations_of(view.config, pb))
return {"direction": direction, "since": since.isoformat() if since else None,
**monitoring.monitor(runs, facts, view.now, view.tz, deviations)}
__all__ = ["headcount", "starters_and_leavers", "onboarding_monitor", "date"]