PacketStation/tests/server/pmhelpers.py

"""Helpers for the phase 8 tests: one database per test module, the Project Management world, and the texts of the
phase 8 contract (periods, PM steps, health, KPI labels, table titles and heads).

Built on shophelpers, qchelpers and closehelpers (imported, never edited). Nothing here reads VISUAL or M-Files.
The labor hours are rows written straight into `visual_labor`, as the feed tests do.

The PM world (dates are calendar days in the site time zone, America/Denver):

    project L670 "Pump skids" (Acme Gas, PO-9, due today + 20), alias OLD670
      unit 001 "Skid A" due yesterday       -> late    k_run (in progress, timer 6 h, ticket 5.2 h, target 4.5 h),
                                                       k_short (warehouse pulling, an open short with a past ETA)
      unit 002 "Skid B" due today + 4       -> risk    k_rev (review after a Q.C. reject), k_qc (at Q.C.)
      unit 003 "Skid C" due today + 30      -> risk    k_hold (warehouse hold for 5 days), k_inc (warehouse waiting)
      unit 004 "Skid D" due today + 30      -> risk    k_co (ready, change order CO-1 open 3 days), k_pulled
      unit 005 "Skid E" due today + 30      -> done    k_filed (filed, created 10 days ago), k_oldfiled (filed 120
                                                       days ago, created 140 days ago, a synthetic history)
      unit 006 "Skid F" no due (project's)  -> ok      k_draft, k_ready, k_asg (assigned), k_complete
      unit 007 "Skid G" due today + 30      -> empty
    project L489 "BPX" (due today + 9), unit 052 with no packets -> empty
    project L333 closed, unit 001 with no packets: not in scope

History rows inserted with old timestamps (append-only tables, so inserted, never updated):
    k_draft: return_to_clerical 400 and 45 days ago (sent back)
    k_oldfiled: send_to_warehouse -139 d, wh_send_back -138 d (sent back), send_to_warehouse -137 d, wh_release -130 d,
                send_to_qc -126 d, qc_reject -125 d (sent back; a rejected qc_history row), send_to_qc -124 d,
                qc_accept -123 d (an accepted qc_history row), file -120 d.

So: open 12, late units 1, units at risk 3, holds 1, change orders 1, shorts 1, over target 1; by period
    rejects 30: 1, 90: 1, 365: 2, all: 2; filed 30: 1, 90: 1, 365: 2, all: 2;
    sent back 30: 1, 90: 2, 365: 4, all: 5; lead time 90: 10.0 d, 365: 15.0 d.
"""

from __future__ import annotations

import secrets
from collections.abc import Iterator
from contextlib import contextmanager
from dataclasses import dataclass, field
from datetime import date, datetime, timedelta
from pathlib import Path
from typing import Any
from zoneinfo import ZoneInfo

import pytest
from fakedir import FakeDirectory
from fastapi.testclient import TestClient
from pkthelpers import step, tiny_pdf, upload
from pshelpers import ORIGIN_HEADERS, PG_URL, app_conn, fake_visual, make_settings
from shophelpers import (
    SIGN,
    Shop,
    draft,
    issue_co,
    ok,
    open_shop,
    pull_all,
    ready_skipped,
    run,
    set_running_since,
    start,
    to_station,
    to_warehouse,
)

from packetstation.settings import Settings

SITE_TZ = "America/Denver"

# ---------------------------------------------------------------- the contract's texts

PERIODS = {"30": "Last 30 days", "90": "Last 90 days", "365": "Last 12 months", "all": "All time"}

PM_STEPS = (
    ("draft", "Ready to release (Clerical)"),
    ("wh_incoming", "Warehouse: waiting"),
    ("wh_pulling", "Warehouse: being pulled"),
    ("wh_hold", "Warehouse: on hold"),
    ("wh_pulled", "Warehouse: pulled, ready for production"),
    ("ready", "Ready to assign"),
    ("station", "At a station"),
    ("review", "Supervisor review"),
    ("qc", "At Q.C."),
    ("complete", "Back for data book"),
)
STEP_LABEL = dict(PM_STEPS)

HEALTH = {"ok": "On track", "risk": "At risk", "late": "Late", "done": "Complete", "empty": "No packets yet"}

KPI_KEYS = ("open", "late_units", "risk_units", "holds", "cos", "shorts", "rejects", "over", "filed", "lead")
KPI_TABS = ("aging", "projects", "projects", "quality", "quality", "quality", "quality", "hours", "projects", "aging")
KPI_ALERT = {"late_units", "risk_units", "holds", "cos", "shorts", "rejects", "over"}


def kpi_labels(period: str = "90") -> tuple[str, ...]:
    lower = PERIODS[period].lower()
    return (
        "Open packets",
        "Late units",
        "Units at risk",
        "Warehouse holds",
        "Open change orders",
        "Open shorts",
        f"Q.C. rejects · {lower}",
        "Packets over target hours (tickets)",
        f"Filed · {lower}",
        "Average lead time, created → filed",
    )


HOURS_TAIL = [
    "Packets",
    "Timer h",
    "Ticket h",
    "Target h",
    "Ticket variance h",
    "Ticket h / target",
    "Over target (tickets)",
]
HEADS: dict[str, list[str]] = {
    "projects": [
        "Project / unit",
        "Name",
        "Customer",
        "Due",
        "Left",
        "Units",
        "Packets",
        "Draft",
        "Warehouse",
        "Shop",
        "Q.C.",
        "Done",
        "% done",
        "Health",
    ],
    "aging_steps": ["Step", "Packets", "Average", "Longest", "Oldest packet"],
    "aging_open": ["Job", "Unit", "Step", "With", "In this step", "Total age", "Unit due", "Flags"],
    "aging_history": ["Step", "Moves", "Average", "Longest"],
    "hours_dept": ["Department", *HOURS_TAIL],
    "hours_worker": ["Worker", *HOURS_TAIL],
    "hours_project": ["Project", *HOURS_TAIL],
    "hours_over": ["Job", "Station / worker", "Status", "Timer h", "Ticket h", "Target h", "Over by (tickets)"],
    "quality_rejects": ["Job", "Date", "Inspector", "Reason / NCR", "Station / worker"],
    "quality_shorts": ["Job", "Item", "Short qty", "Reason", "PO / req", "ETA", "Age", "Packet is"],
    "quality_cos": ["Job", "Change order", "Change", "Waiting on", "Open for"],
    "quality_holds": ["Job", "Reason", "Note", "Put on hold by", "On hold for"],
    "quality_sentback": ["Job", "When", "Action", "By", "Reason"],
    "jobs": [
        "Job",
        "Project",
        "Unit",
        "Operation",
        "Title",
        "Department",
        "Step",
        "With",
        "Station",
        "Worker",
        "Drawing",
        "Rev",
        "Unit due",
        "Days in step",
        "Timer h",
        "Ticket h",
        "Target h",
        "Created",
        "Updated",
        "Change orders open",
        "Shorts open",
    ],
}
ATTENTION_HEAD = ["Unit", "Description", "Due", "Left", "Packets", "% done", "Health"]


def titles(period: str = "90") -> dict[str, str]:
    p = PERIODS[period]
    return {
        "projects": "Projects and units",
        "aging_steps": "Time in the current step (open packets)",
        "aging_open": "Every open packet, longest waiting first",
        "aging_history": f"Average time per step, moves completed ({p})",
        "hours_dept": "By department",
        "hours_worker": "By welder / worker",
        "hours_project": "By project",
        "hours_over": "Packets over target",
        "quality_rejects": f"Q.C. rejects ({p})",
        "quality_shorts": "Open shorts",
        "quality_cos": "Open change orders",
        "quality_holds": "Warehouse holds",
        "quality_sentback": f"Sent back ({p})",
        "jobs": "All jobs in scope",
    }


NOTES = {
    "projects": "Tap a project to show its units. Done = accepted by Q.C. or filed. At risk = due within 7 days and"
    " under 75% done, or a hold or open change order.",
    "aging_steps": "How long packets have been sitting where they are right now.",
    "aging_open": "Over 3 days in one step is flagged.",
    "aging_history": "From the step history recorded on every move.",
    "hours_over": "VISUAL ticket hours against the work-order target hours decide over target. The station timer is"
    " shown beside them. A packet with no ticket hours is not over target.",
}

TAB_TABLES = {
    "aging": ("aging_steps", "aging_open", "aging_history"),
    "hours": ("hours_dept", "hours_worker", "hours_project", "hours_over"),
    "quality": ("quality_rejects", "quality_shorts", "quality_cos", "quality_holds", "quality_sentback"),
}

REJECT_REASON = "Undercut on weld 1, regrind and reweld"
REJECT_NCR = "NCR-2026-0031"
HOLD_REASON = "Waiting on purchasing / PO"
HOLD_NOTE = "PO 4471"
FORMULA_TITLE = "=HYPERLINK(1)"


# ---------------------------------------------------------------- one database per module


def today_site() -> date:
    return datetime.now(ZoneInfo(SITE_TZ)).date()


@contextmanager
def module_settings(tmp: Path) -> Iterator[Settings]:
    """The conftest `pg_settings` for a whole module: a fresh database with ps_migrate and ps_app, migrated to head,
    dropped afterwards. Lets a module build its world once."""
    import psycopg
    from psycopg import sql
    from psycopg.conninfo import conninfo_to_dict

    from packetstation import db, migrate

    assert PG_URL
    info = conninfo_to_dict(PG_URL)
    mig_pw, app_pw = secrets.token_urlsafe(18), secrets.token_urlsafe(18)
    (tmp / "mig").write_text(mig_pw + "\n")
    (tmp / "app").write_text(app_pw + "\n")
    name = f"ps_test_{secrets.token_hex(4)}"
    s = make_settings(
        tmp,
        pg_host=info.get("host", "localhost"),
        pg_port=int(info.get("port", 5432)),
        pg_database=name,
        pg_migrate_password_file=tmp / "mig",
        pg_app_password_file=tmp / "app",
        pg_superuser_url=PG_URL,
        # A world takes minutes to build over a slow tunnel: no idle lock or idle sign-out while it does.
        session_lock_minutes=600,
        session_idle_minutes=720,
    )
    migrate.bootstrap_roles(s)
    with psycopg.connect(PG_URL, autocommit=True) as su:
        for user, pw in (("ps_migrate", mig_pw), ("ps_app", app_pw)):
            su.execute(sql.SQL("ALTER ROLE {} LOGIN PASSWORD {}").format(sql.Identifier(user), sql.Literal(pw)))
    with db.connect_as("migrate", s, autocommit=True) as conn:
        migrate.run(conn, db.MIGRATIONS_DIR)
    try:
        yield s
    finally:
        db.close_pool()
        with psycopg.connect(PG_URL, autocommit=True) as su:
            su.execute(sql.SQL("DROP DATABASE IF EXISTS {} WITH (FORCE)").format(sql.Identifier(name)))


@contextmanager
def module_shop(tmp: Path, fake_dir: FakeDirectory | None = None) -> Iterator[Shop]:
    """The app on a module database with the fake directory and DEV VISUAL settings; every test user signed in."""
    from packetstation import guard

    with pytest.MonkeyPatch.context() as mp:
        mp.setattr(guard, "load_visual_settings", lambda: fake_visual(False))
        with module_settings(tmp) as s:
            gen = open_shop(s, fake_dir or FakeDirectory())
            shop = next(gen)
            try:
                yield shop
            finally:
                gen.close()


def anon(shop: Shop) -> TestClient:
    """A client with no session."""
    return TestClient(shop.main.app, headers=ORIGIN_HEADERS)


def get_ok(c: TestClient, url: str, **params: Any) -> dict[str, Any]:
    r = c.get(url, params={k: v for k, v in params.items() if v is not None})
    assert r.status_code == 200, (url, params, r.text)
    out: dict[str, Any] = r.json()
    return out


# ---------------------------------------------------------------- SQL setup


def make_project(
    settings: Settings,
    number: str,
    name: str,
    *,
    customer: str | None = None,
    po: str | None = None,
    due: date | None = None,
    status: str = "open",
) -> int:
    with app_conn(settings) as conn:
        row = conn.execute(
            "INSERT INTO projects (number, name, customer, po, due, status)"
            " VALUES (%s, %s, %s, %s, %s, %s::project_status) RETURNING id",
            (number, name, customer, po, due, status),
        ).fetchone()
    return int(row[0])


def make_unit(settings: Settings, project_id: int, lot: str, name: str | None = None, due: date | None = None) -> int:
    with app_conn(settings) as conn:
        row = conn.execute(
            "INSERT INTO units (project_id, lot, name, due) VALUES (%s, %s, %s, %s) RETURNING id",
            (project_id, lot, name, due),
        ).fetchone()
    return int(row[0])


def add_alias(settings: Settings, project_id: int, alias: str) -> None:
    run(
        settings,
        "INSERT INTO project_aliases (alias, project_id, kind) VALUES (%s, %s, 'import')",
        (alias.upper(), project_id),
    )


def transition_at(
    settings: Settings,
    packet_id: int,
    days_ago: float,
    action: str,
    from_status: str | None,
    to_status: str | None,
    *,
    by: str = "clerk.clara",
    reason: str | None = None,
    dept: str = "03",
) -> None:
    """One packet_transitions row with an old timestamp (the table is append-only: insert, never update)."""
    run(
        settings,
        "INSERT INTO packet_transitions (packet_id, at, by, action, from_status, to_status, from_dept, to_dept,"
        " reason) VALUES (%s, now() - make_interval(secs => %s), (SELECT id FROM users WHERE sam_account = %s),"
        " %s, %s::packet_status, %s::packet_status, %s, %s, %s)",
        (packet_id, days_ago * 86400, by, action, from_status, to_status, dept, dept, reason),
    )


def qc_history_at(settings: Settings, packet_id: int, days_ago: float, result: str, note: str | None = None) -> None:
    run(
        settings,
        "INSERT INTO qc_history (packet_id, at, result, inspector, note, dept_code, snapshot)"
        " VALUES (%s, now() - make_interval(secs => %s), %s, 'Quinn QC', %s, '03', '{}')",
        (packet_id, days_ago * 86400, result, note),
    )


def labor_rows(settings: Settings, packet_id: int, rows: list[tuple[str, int, str]]) -> None:
    """Make a packet a VISUAL packet and give its work orders VISUAL ticket hours: rows of (wo_base, seq, hours).
    Each work order gets the seq of its row; visual_labor gets one row per entry, as the feed writes them."""
    with app_conn(settings) as conn:
        conn.execute("UPDATE packets SET source_kind = 'visual' WHERE id = %s", (packet_id,))
        for base, seq, hours in rows:
            conn.execute(
                "UPDATE packet_work_orders SET seq = %s WHERE packet_id = %s AND wo_base = %s", (seq, packet_id, base)
            )
            conn.execute(
                "INSERT INTO visual_labor (wo_base, lot, split, sub, seq, hours, tickets, last_ticket)"
                " SELECT wo_base, lot, split, sub, seq, %s, 2, current_date FROM packet_work_orders"
                " WHERE packet_id = %s AND wo_base = %s"
                " ON CONFLICT (wo_base, lot, split, sub, seq) DO UPDATE SET hours = EXCLUDED.hours",
                (hours, packet_id, base),
            )


def quick_draft(
    shop: Shop, unit_id: int, title: str, *, n: int, send: str | None = None, **over: Any
) -> dict[str, Any]:
    """A hand draft by clerk.clara with neutral values (drawing D-61nn, work order Wnn XY, no BOM) unless
    overridden. `send` "warehouse" sends it to the warehouse; a department code releases it there (ready). Either
    attaches a one-page drawing and answers the heat map first. A bare draft costs one request."""
    body: dict[str, Any] = {
        "unit_id": unit_id,
        "dept_code": "03",
        "op": "WELD-PART",
        "title": title,
        "customer": "Plains Midstream",
        "drawing": f"D-61{n:02d}",
        "rev": "A",
        "wps": "WPS-11",
        "work_orders": [
            {"wo_base": f"W{n:02d}XY", "lot": "001", "split": "0", "sub": "0", "qty": 1, "target_hrs": 1},
        ],
        "bom": [],
    }
    body.update(over)
    c = shop("clerk.clara")
    r = c.post("/api/packets", json=body)
    assert r.status_code == 201, r.text
    pkt: dict[str, Any] = r.json()
    if send:
        assert c.patch(f"/api/packets/{pkt['id']}", json={"set": {"heat_map": True}, "base": {}}).status_code == 200
        a = upload(c, tiny_pdf(1))
        assert a.status_code == 201, a.text
        pkt = ok(
            c.post(
                f"/api/packets/{pkt['id']}/pages", json={"asset_id": a.json()["id"], "type": "drawing", "sign": SIGN}
            )
        )
        if send == "warehouse":
            pkt = ok(step(c, pkt, "send_to_warehouse"))
            assert pkt["status"] == "warehouse", pkt["status"]
        else:
            pkt = ok(step(c, pkt, "release_to_shop", dept=send))
            assert pkt["status"] == "ready", pkt["status"]
    return pkt


# ---------------------------------------------------------------- the PM world


@dataclass
class PmWorld:
    shop: Shop
    today: date
    projects: dict[str, int] = field(default_factory=dict)  # "L670", "L489", "L333"
    units: dict[int, int] = field(default_factory=dict)  # 1..7 -> L670 units; 52 -> L489 052
    k: dict[str, int] = field(default_factory=dict)  # packet ids by key

    @property
    def settings(self) -> Settings:
        return self.shop.settings

    def unit_number(self, n: int) -> str:
        return f"L670{n:03d}"


OPEN_KEYS = (
    "k_run",
    "k_short",
    "k_rev",
    "k_qc",
    "k_hold",
    "k_inc",
    "k_co",
    "k_pulled",
    "k_draft",
    "k_ready",
    "k_asg",
    "k_complete",
)
FILED_KEYS = ("k_filed", "k_oldfiled")

# The PM step of each open packet, and the expected pipeline counts.
STEP_OF = {
    "k_draft": "draft",
    "k_inc": "wh_incoming",
    "k_short": "wh_pulling",
    "k_hold": "wh_hold",
    "k_pulled": "wh_pulled",
    "k_ready": "ready",
    "k_co": "ready",
    "k_asg": "station",
    "k_run": "station",
    "k_rev": "review",
    "k_qc": "qc",
    "k_complete": "complete",
}
PIPELINE_COUNTS = {key: sum(1 for v in STEP_OF.values() if v == key) for key, _ in PM_STEPS}

UNIT_HEALTH = {1: "late", 2: "risk", 3: "risk", 4: "risk", 5: "done", 6: "ok", 7: "empty"}

# By period: Q.C. rejects, filed, sent back, lead time.
BY_PERIOD = {
    "30": {"rejects": 1, "filed": 1, "sentback": 1},
    "90": {"rejects": 1, "filed": 1, "sentback": 2, "lead": "10.0 d"},
    "365": {"rejects": 2, "filed": 2, "sentback": 4, "lead": "15.0 d"},
    "all": {"rejects": 2, "filed": 2, "sentback": 5},
}


def pm_world(shop: Shop) -> PmWorld:
    """Build the PM world described at the top of this file. Takes a few minutes over a slow tunnel."""
    from closehelpers import at_complete, filed
    from qchelpers import at_qc, reject

    s = shop.settings
    today = today_site()
    w = PmWorld(shop, today)
    pa = make_project(s, "L670", "Pump skids", customer="Acme Gas", po="PO-9", due=today + timedelta(days=20))
    add_alias(s, pa, "OLD670")
    w.projects["L670"] = pa
    dues = {
        1: today - timedelta(days=1),
        2: today + timedelta(days=4),
        3: today + timedelta(days=30),
        4: today + timedelta(days=30),
        5: today + timedelta(days=30),
        6: None,
        7: today + timedelta(days=30),
    }
    for n, due in dues.items():
        w.units[n] = make_unit(s, pa, f"{n:03d}", f"Skid {'ABCDEFG'[n - 1]}", due)
    pb = make_project(s, "L489", "BPX", customer="BPX", due=today + timedelta(days=9))
    w.projects["L489"] = pb
    w.units[52] = make_unit(s, pb, "052", "48in separator")
    pc = make_project(s, "L333", "Old closed job", status="closed")
    w.projects["L333"] = pc
    make_unit(s, pc, "001", "Closed unit")
    u = w.units
    wade = shop("wh.wade")

    # unit 001, late: a running packet over target on its ticket hours, a short with a past ETA
    run_pkt = start(shop, to_station(shop, ready_skipped(shop, u[1], title="Running spool")))
    w.k["k_run"] = run_pkt["id"]
    run(
        s,
        "UPDATE packet_stages SET clock_elapsed_ms = %s WHERE packet_id = %s AND closed_at IS NULL",
        (4 * 3600 * 1000, run_pkt["id"]),
    )
    set_running_since(s, run_pkt["id"], 2 * 3600)
    labor_rows(s, run_pkt["id"], [("L670001", 10, "3.0"), ("L670002", 20, "2.2")])

    short = to_warehouse(shop, draft(shop, u[1], title="Short spool"))
    yesterday = (today - timedelta(days=1)).isoformat()
    short = ok(
        wade.post(
            f"/api/packets/{short['id']}/bom/1/short",
            json={"qty": 1, "reason": "On order", "ref": "PO-77", "eta": yesterday, "expect_open": False},
        )
    )
    w.k["k_short"] = short["id"]

    # unit 002, at risk by the due rule: a packet back in review after a Q.C. reject, a packet at Q.C.
    rev = at_qc(shop, u[2], title="Rejected spool")
    rev = ok(reject(shop, rev, REJECT_REASON, ncr=REJECT_NCR))
    assert rev["status"] == "review", rev["status"]
    w.k["k_rev"] = rev["id"]
    w.k["k_qc"] = at_qc(shop, u[2], title="Inspected spool")["id"]

    # unit 003, at risk by a hold: a warehouse hold five days old, a packet waiting
    hold = to_warehouse(shop, draft(shop, u[3], title="Held spool"))
    ok(
        wade.post(
            f"/api/packets/{hold['id']}/wh",
            json={"stage": "hold", "reason": HOLD_REASON, "note": HOLD_NOTE, "expect_stage": "incoming"},
        )
    )
    run(
        s,
        "UPDATE packet_wh SET at = now() - interval '5 days' WHERE packet_id = %s AND closed_at IS NULL",
        (hold["id"],),
    )
    w.k["k_hold"] = hold["id"]
    w.k["k_inc"] = to_warehouse(shop, draft(shop, u[3], title="Waiting spool"))["id"]

    # unit 004, at risk by a change order: CO-1 open three days, a packet pulled
    co = issue_co(shop, ready_skipped(shop, u[4], title="Changed spool"), ["03"])
    run(s, "UPDATE change_orders SET issued_at = now() - interval '3 days' WHERE packet_id = %s", (co["id"],))
    w.k["k_co"] = co["id"]
    pulled = pull_all(shop, to_warehouse(shop, draft(shop, u[4], title="Pulled spool")))
    ok(wade.post(f"/api/packets/{pulled['id']}/wh", json={"stage": "pulled", "expect_stage": "pulling"}))
    w.k["k_pulled"] = pulled["id"]

    # unit 005, done: one filed now (created ten days ago), one filed 120 days ago with a synthetic history
    fp = filed(shop, u[5])
    run(s, "UPDATE packets SET created_at = now() - interval '10 days' WHERE id = %s", (fp["id"],))
    w.k["k_filed"] = fp["id"]
    old = draft(shop, u[5], title="Old filed spool")
    oid = old["id"]
    run(
        s,
        "UPDATE packets SET status = 'filed', status_at = now() - interval '120 days',"
        " created_at = now() - interval '140 days' WHERE id = %s",
        (oid,),
    )
    for days, action, frm, to, reason in (
        (139, "send_to_warehouse", "draft", "warehouse", None),
        (138, "wh_send_back", "warehouse", "draft", "Wrong BOM"),
        (137, "send_to_warehouse", "draft", "warehouse", None),
        (130, "wh_release", "warehouse", "ready", None),
        (126, "send_to_qc", "review", "qc", None),
        (125, "qc_reject", "qc", "review", "Porosity at W2"),
        (124, "send_to_qc", "review", "qc", None),
        (123, "qc_accept", "qc", "complete", None),
        (120, "file", "complete", "filed", None),
    ):
        transition_at(s, oid, days, action, frm, to, reason=reason)
    qc_history_at(s, oid, 125, "rejected", "Porosity at W2")
    qc_history_at(s, oid, 123, "accepted")
    w.k["k_oldfiled"] = oid

    # unit 006, on track: a draft (sent back twice long ago), ready, assigned, complete
    dr = draft(shop, u[6], title="Draft spool")
    w.k["k_draft"] = dr["id"]
    transition_at(s, dr["id"], 400, "return_to_clerical", "ready", "draft", by="sup.wes", reason="Wrong department")
    transition_at(s, dr["id"], 45, "return_to_clerical", "ready", "draft", by="sup.wes", reason="Missing drawing")
    w.k["k_ready"] = ready_skipped(shop, u[6], title="Ready spool")["id"]
    w.k["k_asg"] = to_station(shop, ready_skipped(shop, u[6], title="Assigned spool"), station="Stall 2")["id"]
    w.k["k_complete"] = at_complete(shop, u[6])["id"]
    return w


def by_key(rows: list[dict[str, Any]], key: str = "key") -> dict[str, dict[str, Any]]:
    return {str(r[key]): r for r in rows}


def table(answer: dict[str, Any], key: str) -> dict[str, Any]:
    found = [t for t in answer["tables"] if t["key"] == key]
    assert found, (key, [t["key"] for t in answer["tables"]])
    out: dict[str, Any] = found[0]
    return out


def rows_by_pid(t: dict[str, Any]) -> dict[int, dict[str, Any]]:
    return {int(r["pid"]): r for r in t["rows"] if r.get("pid") is not None}


def unit_rows(t: dict[str, Any]) -> dict[int, dict[str, Any]]:
    return {int(r["unit_id"]): r for r in t["rows"] if r.get("lvl") == 1 and r.get("unit_id") is not None}


def project_rows(t: dict[str, Any]) -> dict[int, dict[str, Any]]:
    return {int(r["project_id"]): r for r in t["rows"] if r.get("lvl") == 0 and r.get("project_id") is not None}