"""Deviations — a domain service: how a customer's playbook differs from the reference playbook of its
localisation pack, step by step KEY (never by position), and what each difference means.
Every deviation is one of three classes (design of record: blog repo design/bpmn-driven-lifecycle.md §1):
variation inside the defaults: a step added, renamed, re-timed, given to another group, a condition changed.
Shown, signed with the rest, built.
warning removes a best-practice safeguard. Allowed, and flagged so the customer signs it knowingly:
* an approval of the reference removed (or no longer waited for by a step that waited for it);
* a safeguard step of the pack removed (SAFEGUARDS: the right-to-work check, the written
statement, access ending on the last day, the equipment back);
* a step that takes access away timed later than the reference (access lingers after they leave);
* a step carrying a pack gate removed with it;
* approvals never chased (approval_chase_max = 0).
foundation breaks the kept foundation. Never stored: refused at save with the reason. For hr:
1. a destructive step that no approval precedes (an account is never deleted by a timer);
2. a delete of the account that no completed ownership transfer precedes (nothing they own is lost);
3. a conversion playbook that takes access away (a conversion keeps it);
4. a pack GATE dropped from a step that keeps it (the right-to-work check before the account, the
status determination before a company freelancer's contract): the law, not a preference;
5. two steps with one key, or a key that is not a slug (ids the BPMN, the scenario paths and the
signed change list join on).
1-3 and 5 are Playbook.validate's own refusals; 4 is found here, by comparing with the reference.
Plain Python, no I/O: the application gives it the two playbooks and the pack's GATES and SAFEGUARDS.
"""
from dataclasses import dataclass
from typing import Any, Optional
from brick_hr.domain.playbook import ACCESS_REMOVAL, Playbook, PlaybookStep
CLASSES = ("variation", "warning", "foundation")
FIELDS = ("kind", "title", "responsible_group", "approval_group", "timing", "condition", "capability",
"input_template", "depends_on", "draft_instruction", "template_kind", "fan_out", "outcomes")
# the settings a deviation report covers (the pack's default is SETTINGS_DEFAULTS), and the steps each one touches
WATCHED_SETTINGS = {
"contract_signer_person_id": ("document",),
"task_chase_every_days": ("task",), "task_chase_max": ("task",),
"approval_chase_every_days": ("approval",), "approval_chase_max": ("approval",),
"intake_chase_every_days": ("intake",), "intake_chase_max": ("intake",),
"chase_unit": ("task", "approval", "intake"),
}
@dataclass(frozen=True)
class Deviation:
key: Optional[str] # the step key; None for a setting of the whole playbook
change: str # added | removed | changed | setting
field: Optional[str]
before: Any
after: Any
cls: str # variation | warning | foundation
reason: str
def as_row(self) -> dict:
return {"key": self.key, "change": self.change, "field": self.field, "before": self.before,
"after": self.after, "class": self.cls, "reason": self.reason}
def _view(pb: Playbook, s: PlaybookStep) -> dict:
"""A step's fields with every step NUMBER replaced by its key, so two playbooks compare by name."""
t = dict(s.timing or {})
anchor = t.get("anchor") or "run.created"
if anchor.startswith("step:"):
ref = pb.step(int(anchor[5:].split(".")[0]))
t["anchor"] = f"step:{ref.key if ref else anchor[5:].split('.')[0]}.done"
t = {"anchor": t.get("anchor") or "run.created", "offset": int(t.get("offset") or 0),
"unit": t.get("unit") or "days", "time": t.get("time")}
return {"kind": s.kind, "title": s.title, "responsible_group": s.responsible_group,
"approval_group": s.approval_group, "timing": t, "condition": s.condition or None,
"capability": s.capability, "input_template": s.input_template or None,
"depends_on": pb.keys_of(s.depends_on), "draft_instruction": s.draft_instruction,
"template_kind": s.template_kind, "fan_out": s.fan_out, "outcomes": sorted(s.outcomes or []) or None}
def _gates(cond: Optional[dict]) -> set:
return {g.get("record") for g in (cond or {}).get("requires") or [] if g.get("record")}
def _minutes(t: dict) -> int:
hh, _, mm = (t.get("time") or "00:00").partition(":")
return int(t["offset"]) * 24 * 60 + int(hh or 0) * 60 + int(mm or 0)
def _later(new: dict, old: dict) -> bool:
"""Is a timing rule later than the reference's? (Working days count as days: it is a direction, not a date.)"""
if new["anchor"] != old["anchor"]:
return old["anchor"] == "last_day"
return _minutes(new) > _minutes(old)
def compare(pb: Playbook, ref: Playbook, *, gates: dict, safeguards: dict) -> list:
"""Every difference between a customer's playbook and its reference, by step key, classified."""
out: list = []
mine, theirs = {s.key: s for s in pb.steps}, {s.key: s for s in ref.steps}
approvals = {k for k, s in theirs.items() if s.kind == "approval"}
for key in sorted(set(theirs) - set(mine), key=lambda k: theirs[k].step_no):
r = theirs[key]
if r.kind == "approval":
after = [theirs[d].title for d in theirs if r.step_no in theirs[d].depends_on]
out.append(Deviation(key, "removed", None, r.title, None, "warning",
f"removes the approval '{r.title}'" + (f" that {', '.join(after)} waited for"
if after else "")))
elif key in safeguards:
out.append(Deviation(key, "removed", None, r.title, None, "warning",
f"removes '{r.title}': best practice is that {safeguards[key]}"))
elif _gates(r.condition):
out.append(Deviation(key, "removed", None, r.title, None, "warning",
f"removes '{r.title}' and with it the {', '.join(sorted(_gates(r.condition)))} "
"gate on it"))
else:
out.append(Deviation(key, "removed", None, r.title, None, "variation", f"'{r.title}' is not done"))
for key in sorted(set(mine) - set(theirs), key=lambda k: mine[k].step_no):
s = mine[key]
out.append(Deviation(key, "added", None, None, s.title, "variation",
f"adds '{s.title}' ({s.kind}, {s.responsible_group})"))
for key in sorted(set(mine) & set(theirs), key=lambda k: mine[k].step_no):
a, b = _view(pb, mine[key]), _view(ref, theirs[key])
title = mine[key].title
for f in FIELDS:
if a[f] == b[f]:
continue
cls, why = "variation", f"'{title}': {f.replace('_', ' ')} changed"
if f == "kind" and b[f] == "approval":
cls, why = "warning", f"'{title}' is no longer an approval"
elif f == "depends_on":
dropped = sorted(set(b[f]) - set(a[f]) & approvals)
if dropped:
cls, why = "warning", (f"'{title}' no longer waits for the approval "
f"{', '.join(repr(theirs[d].title) for d in dropped)}")
elif f == "timing" and mine[key].capability in ACCESS_REMOVAL and _later(a[f], b[f]):
cls, why = "warning", f"'{title}' takes access away later than the reference: access lingers"
elif f == "condition":
lost = sorted(_gates(b[f]) - _gates(a[f]))
if lost:
cls = "foundation"
why = "; ".join(f"'{title}' drops the {g} gate: {gates.get(g) or 'a gate of the pack'}"
for g in lost)
out.append(Deviation(key, "changed", f, b[f], a[f], cls, why))
return out
def settings(values: dict, defaults: dict, pb: Playbook) -> list:
"""The watched settings that differ from the pack's defaults, where this playbook has a step they touch."""
kinds = {s.kind for s in pb.steps}
out = []
for key, touches in WATCHED_SETTINGS.items():
v, d = str(values.get(key) or ""), str(defaults.get(key) or "")
if v == d or not kinds & set(touches):
continue
cls, why = "variation", f"{key.replace('_', ' ')} is {v or 'unset'} (default {d or 'unset'})"
if key == "approval_chase_max" and v == "0":
cls, why = "warning", "approvals are never chased: a forgotten approval holds the run"
if key == "contract_signer_person_id":
why = (f"contracts are signed for the company by person {v}, unless a run names its own director"
if v else "contracts are signed by the director named on each run")
out.append(Deviation(None, "setting", key, d or None, v or None, cls, why))
return out
def foundation(devs: list) -> list:
return [d for d in devs if d.cls == "foundation"]
def marks(devs: list) -> dict:
"""step key → (class, the reasons in words): the strongest class of each step's deviations, for the BPMN."""
out: dict = {}
for d in devs:
if d.key is None or d.change == "removed":
continue
cls, why = out.get(d.key, ("variation", []))
if CLASSES.index(d.cls) > CLASSES.index(cls):
cls = d.cls
out[d.key] = (cls, why + [d.reason])
return {k: (c, "; ".join(w)) for k, (c, w) in out.items()}
def summary(devs: list) -> dict:
return {c: sum(1 for d in devs if d.cls == c) for c in CLASSES}
__all__ = ["Deviation", "compare", "settings", "foundation", "marks", "summary", "CLASSES", "WATCHED_SETTINGS"]