PacketStation/server/packetstation/policy/actions.py

"""The Action enum, the Who predicates and the rule table. Plain data; the engine reads it.

A port of the old guard.js POLICY_TRANSITIONS, POLICY_FIELDS, SIGN_OFF and CO_HOLD tables, with the
tightenings listed in docs/POLICY.md.
"""

from __future__ import annotations

from dataclasses import dataclass
from enum import StrEnum

# Statuses in workflow order (the packet_status enum declaration order).
STATUSES: tuple[str, ...] = (
    "draft",
    "warehouse",
    "ready",
    "assigned",
    "in_progress",
    "review",
    "qc",
    "complete",
    "filed",
)
SHOP: frozenset[str] = frozenset({"ready", "assigned", "in_progress", "review"})
ACTIVE: frozenset[str] = frozenset({"assigned", "in_progress"})

# Short labels for messages.
STATUS_LABELS: dict[str, str] = {
    "draft": "Draft",
    "warehouse": "With warehouse",
    "ready": "Ready to assign",
    "assigned": "At a station",
    "in_progress": "In progress",
    "review": "Supervisor review",
    "qc": "At Q.C.",
    "complete": "Complete",
    "filed": "Filed",
}

# Department codes that are not shop departments.
SUPPORT_CODES: frozenset[str] = frozenset({"CL", "WH", "QC", "PM"})


class Action(StrEnum):
    # transitions
    SEND_TO_WAREHOUSE = "send_to_warehouse"
    RELEASE_TO_SHOP = "release_to_shop"
    SEND_DRAFT_TO_QC = "send_draft_to_qc"
    WH_SEND_BACK = "wh_send_back"
    RETURN_TO_CLERICAL = "return_to_clerical"
    WH_RELEASE = "wh_release"
    SEND_BACK_TO_WAREHOUSE = "send_back_to_warehouse"
    ASSIGN = "assign"
    UNASSIGN = "unassign"
    PULL_BACK = "pull_back"
    START = "start"
    SUBMIT = "submit"
    RETURN_TO_STATION = "return_to_station"
    SEND_TO_QC = "send_to_qc"
    ROUTE_FROM_REVIEW = "route_from_review"
    QC_REJECT = "qc_reject"
    QC_ACCEPT = "qc_accept"
    QC_ACCEPT_AND_ROUTE = "qc_accept_and_route"
    FILE = "file"
    ADMIN_REOPEN = "admin_reopen"
    # edits (no status change)
    EDIT_HEADER = "edit_header"
    EDIT_DRAWING_REF = "edit_drawing_ref"
    EDIT_DEPT = "edit_dept"
    MOVE_UNIT = "move_unit"
    EDIT_WOS = "edit_wos"
    EDIT_BOM = "edit_bom"
    WH_RECEIVE_SHORT = "wh_receive_short"
    EDIT_HEATS = "edit_heats"
    EDIT_HEAT_TABLE = "edit_heat_table"
    EDIT_WELDS = "edit_welds"
    EDIT_CHECKLIST = "edit_checklist"
    EDIT_STAGE_FIELDS = "edit_stage_fields"
    CLOCK = "clock"
    EDIT_QC = "edit_qc"
    SET_WH_STAGE = "set_wh_stage"
    EDIT_DOCUMENT_SET = "edit_document_set"
    DOC_ADD = "doc_add"
    DOC_REMOVE = "doc_remove"
    REV_APPROVE = "rev_approve"
    SET_VCLOSED = "set_vclosed"
    SET_DATABOOK_AT = "set_databook_at"
    REMOVE = "remove"
    CO_ISSUE = "co_issue"
    CO_APPROVE = "co_approve"
    CO_REJECT = "co_reject"
    CO_HOLD_TOGGLE = "co_hold_toggle"
    CO_VERIFY = "co_verify"
    ITP_EDIT_LINE = "itp_edit_line"
    ITP_BLOCK_SIGN = "itp_block_sign"
    ITP_SIGN_OFF = "itp_sign_off"
    ITP_REOPEN = "itp_reopen"
    VIS_UPDATE_DRAFT = "vis_update_draft"
    VIS_OPEN_CO = "vis_open_co"
    VIS_AUTO_RELEASE = "vis_auto_release"
    VIS_RETIRE_DRAFT = "vis_retire_draft"


class Who(StrEnum):
    """The closed set of who predicates. custody.allows() evaluates them."""

    CL = "CL"
    CL_SUP = "CL_SUP"
    WH = "WH"
    WH_SUP = "WH_SUP"
    QC = "QC"
    QC_SUP = "QC_SUP"
    CL_OR_QC = "CL_OR_QC"
    HOLDER = "HOLDER"
    DEPT_LEAD = "DEPT_LEAD"
    DEPT_SUP = "DEPT_SUP"
    WORKS_ON = "WORKS_ON"
    CO_VERIFIER = "CO_VERIFIER"
    ADMIN = "ADMIN"
    SYSTEM = "SYSTEM"


@dataclass(frozen=True)
class Rule:
    who: Who
    from_status: frozenset[str]  # empty = any status (edits); the filed lock is applied separately
    to_status: str | None  # None for edits
    dept_change: str = "no"  # no | required | allowed
    co_hold: bool = False
    preconditions: tuple[str, ...] = ()
    signature: str | None = None  # the meaning text when a signature is required
    reason_required: bool = False
    filed_ok: bool = False  # only SET_DATABOOK_AT and ADMIN_REOPEN
    system_only: bool = False

    @property
    def transition(self) -> bool:
        return self.to_status is not None


def _s(*statuses: str | frozenset[str]) -> frozenset[str]:
    out: set[str] = set()
    for s in statuses:
        if isinstance(s, str):
            out.add(s)
        else:
            out.update(s)
    return frozenset(out)


ANY: frozenset[str] = frozenset()
_REASON = ("reason_given",)
_DRAFT_OUT = ("heat_map_answered", "drawing_attached")

A = Action
RULES: dict[Action, Rule] = {
    # transitions
    A.SEND_TO_WAREHOUSE: Rule(Who.CL, _s("draft"), "warehouse", preconditions=_DRAFT_OUT),
    A.RELEASE_TO_SHOP: Rule(
        Who.CL, _s("draft"), "ready", dept_change="allowed", preconditions=(*_DRAFT_OUT, "target_dept_given")
    ),
    A.SEND_DRAFT_TO_QC: Rule(Who.CL, _s("draft"), "qc", preconditions=_DRAFT_OUT),
    A.WH_SEND_BACK: Rule(Who.WH, _s("warehouse"), "draft", preconditions=_REASON, reason_required=True),
    A.RETURN_TO_CLERICAL: Rule(Who.DEPT_LEAD, _s("ready"), "draft", preconditions=_REASON, reason_required=True),
    A.WH_RELEASE: Rule(
        Who.WH, _s("warehouse"), "ready", co_hold=True, preconditions=("bom_all_pulled_or_short", "heats_recorded")
    ),
    A.SEND_BACK_TO_WAREHOUSE: Rule(
        Who.DEPT_LEAD, _s("ready"), "warehouse", preconditions=_REASON, reason_required=True
    ),
    A.ASSIGN: Rule(Who.DEPT_LEAD, _s("ready"), "assigned", preconditions=("welder_set",)),
    A.UNASSIGN: Rule(Who.DEPT_LEAD, _s("assigned"), "ready"),
    A.PULL_BACK: Rule(Who.DEPT_LEAD, _s("in_progress"), "ready"),
    A.START: Rule(Who.WORKS_ON, _s("assigned"), "in_progress"),
    A.SUBMIT: Rule(
        Who.WORKS_ON,
        _s("assigned", "in_progress"),
        "review",
        co_hold=True,
        preconditions=("checklist_complete", "welds_recorded", "filler_set"),
        signature="Submit work to the supervisor",
    ),
    A.RETURN_TO_STATION: Rule(Who.DEPT_SUP, _s("review"), "assigned", preconditions=_REASON, reason_required=True),
    A.SEND_TO_QC: Rule(
        Who.DEPT_SUP,
        _s("review"),
        "qc",
        co_hold=True,
        preconditions=("qc_book_spool_location",),
        signature="Send to Q.C.",
    ),
    A.ROUTE_FROM_REVIEW: Rule(
        Who.DEPT_SUP,
        _s("review"),
        "ready",
        dept_change="required",
        co_hold=True,
        preconditions=("target_dept_given",),
        signature="Route to another department",
    ),
    A.QC_REJECT: Rule(Who.QC, _s("qc"), "review", preconditions=_REASON, signature="Q.C. reject", reason_required=True),
    A.QC_ACCEPT: Rule(
        Who.QC,
        _s("qc"),
        "complete",
        co_hold=True,
        preconditions=("qc_missing_empty", "route_given"),
        signature="Q.C. accept",
    ),
    A.QC_ACCEPT_AND_ROUTE: Rule(
        Who.QC,
        _s("qc"),
        "ready",
        dept_change="required",
        co_hold=True,
        preconditions=("qc_missing_empty", "target_dept_given"),
        signature="Q.C. accept and route",
    ),
    A.FILE: Rule(
        Who.CL,
        _s("complete"),
        "filed",
        co_hold=True,
        preconditions=("vclosed_if_databook",),
        signature="File in the data book",
    ),
    A.ADMIN_REOPEN: Rule(
        Who.ADMIN,
        _s("filed"),
        "complete",
        preconditions=_REASON,
        signature="Reopen a filed packet",
        reason_required=True,
        filed_ok=True,
    ),
    # edits
    A.EDIT_HEADER: Rule(Who.CL, _s("draft"), None),
    A.EDIT_DRAWING_REF: Rule(Who.CL, ANY, None),
    A.EDIT_DEPT: Rule(Who.CL, _s("draft"), None),
    A.MOVE_UNIT: Rule(Who.CL, ANY, None),
    A.EDIT_WOS: Rule(Who.CL, _s("draft"), None),
    A.EDIT_BOM: Rule(Who.HOLDER, _s("draft", "warehouse", SHOP), None),
    A.WH_RECEIVE_SHORT: Rule(Who.WH, ANY, None, preconditions=("short_open",)),
    A.EDIT_HEATS: Rule(Who.HOLDER, _s("warehouse", SHOP), None),
    A.EDIT_HEAT_TABLE: Rule(Who.HOLDER, _s("warehouse", SHOP), None),
    A.EDIT_WELDS: Rule(Who.HOLDER, SHOP, None),
    A.EDIT_CHECKLIST: Rule(Who.HOLDER, SHOP, None),
    A.EDIT_STAGE_FIELDS: Rule(Who.HOLDER, SHOP, None),
    A.CLOCK: Rule(Who.HOLDER, SHOP, None),
    A.EDIT_QC: Rule(Who.QC, _s("qc"), None),
    A.SET_WH_STAGE: Rule(Who.WH, _s("warehouse"), None),
    A.EDIT_DOCUMENT_SET: Rule(Who.CL, ANY, None, signature="Attach a drawing by hand"),
    A.DOC_ADD: Rule(Who.CL_OR_QC, ANY, None),
    A.DOC_REMOVE: Rule(Who.CL_OR_QC, ANY, None),
    A.REV_APPROVE: Rule(
        Who.CL_SUP,
        _s("draft"),
        None,
        preconditions=_REASON,
        signature="Approve the M-Files revision",
        reason_required=True,
    ),
    A.SET_VCLOSED: Rule(Who.CL, _s("complete"), None),
    A.SET_DATABOOK_AT: Rule(Who.CL_OR_QC, ANY, None, filed_ok=True),
    A.REMOVE: Rule(Who.CL, _s("draft"), None, preconditions=_REASON, reason_required=True),
    A.CO_ISSUE: Rule(Who.CL, ANY, None, signature="Issue a change order"),
    A.CO_APPROVE: Rule(Who.CL, ANY, None, signature="Approve a change order"),
    A.CO_REJECT: Rule(
        Who.CL, ANY, None, preconditions=_REASON, signature="Reject a change order", reason_required=True
    ),
    A.CO_HOLD_TOGGLE: Rule(Who.CL, ANY, None),
    A.CO_VERIFY: Rule(Who.CO_VERIFIER, ANY, None, signature="Verify a change order for the department"),
    A.ITP_EDIT_LINE: Rule(Who.QC, ANY, None),
    A.ITP_BLOCK_SIGN: Rule(Who.QC, ANY, None, signature="Sign an ITP block"),
    A.ITP_SIGN_OFF: Rule(Who.QC, ANY, None, signature="Sign off the ITP"),
    # QC supervisor; an admin passes through the bypass with an IT reason.
    A.ITP_REOPEN: Rule(Who.QC_SUP, ANY, None, preconditions=_REASON, reason_required=True),
    A.VIS_UPDATE_DRAFT: Rule(Who.SYSTEM, _s("draft"), None, system_only=True),
    A.VIS_OPEN_CO: Rule(Who.SYSTEM, frozenset(STATUSES) - {"draft", "filed"}, None, system_only=True),
    A.VIS_AUTO_RELEASE: Rule(Who.SYSTEM, _s("draft"), "warehouse", preconditions=_DRAFT_OUT, system_only=True),
    A.VIS_RETIRE_DRAFT: Rule(Who.SYSTEM, _s("draft"), None, system_only=True),
}

# The user transitions (not the system ones), in table order.
TRANSITIONS: tuple[Action, ...] = tuple(a for a, r in RULES.items() if r.transition and not r.system_only)

# What an edit changes, for the "Your access doesn't allow changing <what>" message.
EDIT_WHAT: dict[Action, str] = {
    A.EDIT_HEADER: "the packet details",
    A.EDIT_DRAWING_REF: "the drawing number",
    A.EDIT_DEPT: "the department",
    A.MOVE_UNIT: "the unit",
    A.EDIT_WOS: "the work orders",
    A.EDIT_BOM: "the materials",
    A.WH_RECEIVE_SHORT: "short parts",
    A.EDIT_HEATS: "heat tags",
    A.EDIT_HEAT_TABLE: "the heat table",
    A.EDIT_WELDS: "welds",
    A.EDIT_CHECKLIST: "the checklist",
    A.EDIT_STAGE_FIELDS: "the stage details",
    A.CLOCK: "the clock",
    A.EDIT_QC: "the Q.C. record",
    A.SET_WH_STAGE: "the warehouse stage",
    A.EDIT_DOCUMENT_SET: "the drawing pages",
    A.DOC_ADD: "the documents",
    A.DOC_REMOVE: "the documents",
    A.REV_APPROVE: "the drawing revision",
    A.SET_VCLOSED: "the VISUAL close",
    A.SET_DATABOOK_AT: "the data book date",
    A.REMOVE: "whether it is removed",
    A.CO_ISSUE: "change orders",
    A.CO_APPROVE: "change orders",
    A.CO_REJECT: "change orders",
    A.CO_HOLD_TOGGLE: "change orders",
    A.CO_VERIFY: "change orders",
    A.ITP_EDIT_LINE: "the ITP",
    A.ITP_BLOCK_SIGN: "the ITP",
    A.ITP_SIGN_OFF: "the ITP",
    A.ITP_REOPEN: "the ITP",
}