"""Synthetic VISUAL feed and M-Files drawing fixtures for tests and CI. No real job data. python -m tests.fixtures.generator --out tests/fixtures/generated Writes feed.1.json, feed.2.json, drawings.1.json, drawings.2.json, the one-page PDFs the drawing files name, and `state` = "1". The feed files have the exact shape of VisualKit's examples/packet_station_feed.py: rows carry the status as a word, materials carry the status code, numbers are JSON numbers, dates are ISO, lot/split/sub are text, seq and piece are ints. An optional `labor` list follows the FixtureSource contract. Dates are offsets from `today` (default: the date the generator runs), so the window rows stay inside the feed's default window (14 days back, 30 ahead) on the day the files are made. Everything else is fixed: the output is the same for the same `today`. """ from __future__ import annotations import argparse import io import json import sys from datetime import date, timedelta from pathlib import Path from typing import Any SITE = "PESCO" STATUS_WORDS = {"F": "Firmed", "R": "Released", "U": "Unreleased", "C": "Closed", "X": "Cancelled"} # The part of VISUAL's resource map the fixtures use (the seed in migration 0004 has these too). NEWRES is unmapped. RESOURCE_MAP: dict[str, str] = { "LAYOUT": "01", "WELD-PART": "03", "WELD-CODE": "06", "MFG": "fold", "FORKLIFT": "ignore", } # Go-live: the tests set feed settings golive_date = today + GOLIVE_OFFSET. Work order L900-13 is released before it. GOLIVE_OFFSET = -5 # Names the tests and the e2e scenario refer to. PROJECT = "L900" PROJECT_CLOSED_ORDER = "L901" UNIT1, UNIT2 = "L900001", "L900002" KEY_WELD_PART = "L900001|WELD-PART|L900-0101" # rev 1 -> 2 in state 2; the in-shop packet in the scenario KEY_LAYOUT_A = "L900001|LAYOUT|L900-0101" # same work order, another resource: a second packet KEY_TWO_OPS = "L900001|WELD-CODE|L900-0201" # two operations, same resource and drawing; qty change in state 2 KEY_NO_DWG = "L900001|WELD-PART|(no drawing)" KEY_REV_A = "L900002|WELD-PART|L900-0401" # non-numeric rev; two legs with duplicate pieces KEY_DRAFTING = "L900002|LAYOUT|L900-0501" # drawing in drafting; this leg is gone from VISUAL in state 2 KEY_NOT_FOUND = "L900002|WELD-PART|L900-0701" # drawing not in M-Files; tgt-only change in state 2 (minor) KEY_CLOSES = "L900001|WELD-CODE|L900-0901" # closed in VISUAL in state 2 KEY_CANCELS = "L900001|WELD-CODE|L900-1001" # cancelled in VISUAL in state 2 KEY_MOVES = "L900002|WELD-CODE|L900-1601" # moves to drawing L900-1602 in state 2 (retire-moved) KEY_MOVED_TO = "L900002|WELD-CODE|L900-1602" KEY_GOLIVE = "L900002|WELD-PART|L900-1301" # released before go-live when the tests set it KEY_L901 = "L901001|WELD-PART|L901-0101" KEY_RELEASED = "L900002|WELD-CODE|L900-0601" # Released (not Firmed): only a candidate when tracked DWG_WELD_PART = "L900-0101" # Every candidate state 1 makes on the first run (no tracked keys, no go-live). STATE1_KEYS = frozenset( { KEY_WELD_PART, KEY_LAYOUT_A, KEY_TWO_OPS, KEY_NO_DWG, KEY_REV_A, KEY_DRAFTING, KEY_NOT_FOUND, KEY_CLOSES, KEY_CANCELS, KEY_MOVES, KEY_GOLIVE, KEY_L901, } ) ORDER_ACME = "C24-0901" # open, top job L900002, also linked to base L900-04 ORDER_BAYOU = "C24-0902" # open, top job L900001 (the unit rule must pick it for L900001) ORDER_L901 = "C23-0450" # closed, the only order of L901 def _d(today: date, days: int | None) -> str | None: return None if days is None else (today + timedelta(days=days)).isoformat() def _num(x: float | int) -> int | float: """A JSON number as the feed writes it: integers without '.0'.""" return int(x) if float(x).is_integer() else float(x) class _Wo: """One work order leg with its operations, as the generator describes it.""" def __init__( self, base: str, lot: str, ops: list[tuple[int, str, float]], *, dwg: str, rev: str, st: str, rel: int, want: int, qty: float, desc: str, split: str = "0", sub: str = "0", part: str = "", ps: dict[int, str] | None = None, ) -> None: self.base, self.lot, self.split, self.sub = base, lot, split, sub self.ops = ops # (seq, resource, run hours per piece) self.dwg, self.rev, self.st, self.rel, self.want, self.qty = dwg, rev, st, rel, want, qty self.desc, self.part, self.ps = desc, part or f"FG-{base}", ps or {} def rows(self, today: date) -> list[dict[str, Any]]: out = [] for seq, res, run in self.ops: out.append( { "wo": self.base, "lot": self.lot, "split": self.split, "sub": self.sub, "seq": seq, "res": res, "dwg": self.dwg, "rev": self.rev, "ps": self.ps.get(seq, ""), "desc": self.desc, "qty": _num(self.qty), "tgt": _num(round(run * self.qty, 4)), "rel": _d(today, self.rel), "want": _d(today, self.want), "st": STATUS_WORDS[self.st], "barcode": f"*%2{self.base}${self.lot}${self.split}${self.sub}${seq}%*", "part": self.part, "site": SITE, "order": None, } ) return out def _mat( wo: _Wo, seq: int, piece: int, part: str | None, qty: float, *, issued: float = 0, whs: str = "SHOP", loc: str = "R1", comm: str = "PIPE", cons: str = "N", desc: str | None = None, dwg: str | None = None, st: str | None = None, closed: bool = False, ) -> dict[str, Any]: calc = qty if issued > calc: istat = 3 elif closed: istat = 0 elif issued == 0: istat = 2 else: istat = 1 return { "wo": wo.base, "lot": wo.lot, "split": wo.split, "sub": wo.sub, "seq": seq, "piece": piece, "part": part, "qty": _num(qty), "issued": _num(issued), "whs": whs if part else "", "loc": loc if part else "", "st": st or wo.st, "istat": istat, "dwg": wo.dwg if dwg is None else dwg, "desc": desc if desc is not None else (f"Part {part}" if part else "Split item"), "comm": comm if part else "", "cons": cons if part else "", } def _work_orders() -> dict[str, _Wo]: """State 1's work orders by a short name.""" w: dict[str, _Wo] = {} # A: two operations at different resources on one drawing (two packets, one BOM each), plus fold and ignore rows w["A"] = _Wo( "L900-01", "1", [(10, "LAYOUT", 0.5), (20, "WELD-PART", 1.5), (30, "MFG", 0.25), (40, "FORKLIFT", 0.1), (50, "MAT_PIPE", 0)], dwg="L900-0101", rev="1", st="F", rel=-3, want=20, qty=2, desc="Inlet spool 6 in", ps={10: "Cut to length per sheet 1", 20: "Weld per WPS-11\r\nBack purge", 30: "Weld per WPS-11\r\nBack purge"}, ) # A2: an Unreleased split of A's base (only on a packet when tracked) w["A2"] = _Wo( "L900-01", "1", [(20, "WELD-PART", 1.5)], dwg="L900-0101", rev="1", st="U", rel=-3, want=25, qty=1, desc="Inlet spool 6 in", split="1", ) # B: two operations at the same resource and drawing (two legs with different seq on one packet) w["B"] = _Wo( "L900-02", "1", [(10, "WELD-CODE", 2), (20, "WELD-CODE", 1)], dwg="L900-0201", rev="2", st="F", rel=-2, want=21, qty=2, desc="Header box", ) # C: no drawing w["C"] = _Wo( "L900-03", "1", [(10, "WELD-PART", 0.75)], dwg="", rev="", st="F", rel=-1, want=18, qty=4, desc="Pipe supports" ) # D and H: one candidate (same unit, resource, drawing) with a non-numeric rev; duplicate piece 1 across the legs w["D"] = _Wo( "L900-04", "2", [(10, "WELD-PART", 1)], dwg="L900-0401", rev="A", st="F", rel=-4, want=22, qty=1, desc="Outlet nozzle", ) w["H"] = _Wo( "L900-08-1", "2", [(10, "WELD-PART", 0.5)], dwg="L900-0401", rev="", st="F", rel=-2, want=24, qty=1, desc="Outlet nozzle", ) # E: drawing in drafting; gone from VISUAL in state 2 w["E"] = _Wo( "L900-05", "2", [(10, "LAYOUT", 1)], dwg="L900-0501", rev="1", st="F", rel=-1, want=19, qty=3, desc="Skid frame" ) # F: Released, with materials (R legs count as mat_known); not in the window set w["F"] = _Wo( "L900-06", "2", [(10, "WELD-CODE", 1)], dwg="L900-0601", rev="1", st="R", rel=-2, want=20, qty=1, desc="Drain leg", ) # G: drawing not in M-Files; tgt-only change in state 2 w["G"] = _Wo( "L900-07", "2", [(10, "WELD-PART", 2)], dwg="L900-0701", rev="1", st="F", rel=-3, want=23, qty=2, desc="Vent stack", ) # J closes in state 2, K is cancelled in state 2 w["J"] = _Wo( "L900-09", "1", [(10, "WELD-CODE", 1)], dwg="L900-0901", rev="1", st="F", rel=-5, want=15, qty=1, desc="Gauge board", ) w["K"] = _Wo( "L900-10", "1", [(10, "WELD-CODE", 1)], dwg="L900-1001", rev="1", st="F", rel=-5, want=15, qty=1, desc="Sample cooler", ) # A Closed and a Cancelled leg outside the window set (only on a packet when tracked) w["L"] = _Wo( "L900-11", "1", [(10, "WELD-CODE", 1)], dwg="L900-1101", rev="1", st="C", rel=-6, want=10, qty=1, desc="Closed spool", ) w["X"] = _Wo( "L900-12", "1", [(10, "WELD-CODE", 1)], dwg="L900-1201", rev="1", st="X", rel=-6, want=10, qty=1, desc="Cancelled spool", ) # Go-live: released before golive_date = today + GOLIVE_OFFSET (still inside the window) w["P"] = _Wo( "L900-13", "2", [(10, "WELD-PART", 1)], dwg="L900-1301", rev="1", st="F", rel=-10, want=12, qty=1, desc="Old release", ) # A non-numeric lot: no unit number, no candidate w["Q"] = _Wo( "L900-14", "7A", [(10, "WELD-PART", 1)], dwg="L900-1401", rev="1", st="F", rel=-2, want=20, qty=1, desc="Lettered lot", ) # An unmapped resource w["N"] = _Wo( "L900-15", "2", [(10, "NEWRES", 1), (20, "NEWRES", 1)], dwg="L900-1501", rev="1", st="F", rel=-2, want=20, qty=1, desc="New resource work", ) # M: moves to drawing L900-1602 in state 2 w["M"] = _Wo( "L900-16", "2", [(10, "WELD-CODE", 1)], dwg="L900-1601", rev="1", st="F", rel=-2, want=26, qty=1, desc="Bypass loop", ) # Project L901: only a closed order w["Z"] = _Wo( "L901-01", "1", [(10, "WELD-PART", 1)], dwg="L901-0101", rev="1", st="F", rel=-2, want=28, qty=1, desc="Closed-order spool", ) return w def _materials(w: dict[str, _Wo]) -> list[dict[str, Any]]: a, d, h, f, b = w["A"], w["D"], w["H"], w["F"], w["B"] lines = [ # A seq 10 (LAYOUT): a split-item header (piece 10, no part) with lines 11 and 12 _mat(a, 10, 10, None, 0, desc="Split item: flange set"), _mat(a, 10, 11, "FLG-6-150", 2, issued=1), _mat(a, 10, 12, "GSK-6-150", 2, cons="Y", comm="GASKET"), # A seq 20 (WELD-PART): a MAIN warehouse line, a WIP sub-assembly, a line on another drawing (dropped) _mat(a, 20, 1, "PIPE-6-SCH40", 12.5, issued=12.5, whs="MAIN", loc="A-01"), _mat(a, 20, 2, "L900-01-SUB", 1, whs="ASY", loc="", comm="WIP"), _mat(a, 20, 3, "ELB-6-90", 2, dwg="L900-9999"), _mat(a, 20, 4, "WOL-1-6", 1, dwg=""), # blank drawing on the line: it belongs # B (two operations) _mat(b, 10, 1, "PLT-0.5-A36", 4), _mat(b, 20, 2, "CPL-2-3000", 2, issued=3), # D and H: piece 1 on both legs (the second becomes item 1-1) _mat(d, 10, 1, "NOZ-4-300", 1), _mat(h, 10, 1, "NOZ-4-300", 1), _mat(h, 10, 2, "PAD-4", 1, closed=True, issued=1), # F: a Released leg's line _mat(f, 10, 1, "PIPE-1-SCH80", 3, st="R"), # G _mat(w["G"], 10, 1, "PIPE-8-SCH40", 6), # Unreleased split of A: not F or R, so not a material leg _mat(w["A2"], 20, 1, "PIPE-6-SCH40", 6, st="U"), ] return lines def _orders(today: date) -> tuple[dict[str, list[dict[str, Any]]], dict[str, list[dict[str, Any]]]]: acme = { "cust_order_id": ORDER_ACME, "lines": [1], "top_jobs": ["L900002"], "status": "R", "customer_id": "ACME01", "customer_name": "Acme Refining", "customer_po_ref": "PO-1001", "order_date": _d(today, -40), "desired_ship_date": _d(today, 60), "promise_date": _d(today, 55), } bayou = { "cust_order_id": ORDER_BAYOU, "lines": [1, 2], "top_jobs": ["L900001"], "status": "F", "customer_id": "BAYOU01", "customer_name": "Bayou Chemical", "customer_po_ref": "PO-2002", "order_date": _d(today, -45), "desired_ship_date": _d(today, 70), "promise_date": None, } closed = { "cust_order_id": ORDER_L901, "lines": [1], "top_jobs": ["L901001"], "status": "C", "customer_id": "DELTA01", "customer_name": "Delta Gas", "customer_po_ref": "PO-3003", "order_date": _d(today, -200), "desired_ship_date": _d(today, -100), "promise_date": _d(today, -105), } base_entry = {k: v for k, v in acme.items() if k not in ("top_jobs", "order_date")} by_base = {"L900-04": [base_entry]} by_project = {PROJECT: [acme, bayou], PROJECT_CLOSED_ORDER: [closed]} return by_base, by_project def _labor(today: date) -> list[dict[str, Any]]: return [ { "wo": "L900-01", "lot": "1", "split": "0", "sub": "0", "seq": 20, "hours": 1.25, "tickets": 2, "last_ticket": _d(today, -1), }, { "wo": "L900-02", "lot": "1", "split": "0", "sub": "0", "seq": 10, "hours": 0.5, "tickets": 1, "last_ticket": _d(today, -1), }, ] def _meta(today: date, state: int) -> dict[str, Any]: return { "source": "fixture", "version": "synthetic", "server": "fixture", "database": f"state {state}", "site": SITE, "since": _d(today, -14), "until": _d(today, 30), "statuses": ["F", "R", "U", "C"], "generated_at": f"{today.isoformat()}T06:00:00", } def feed_state(n: int, today: date | None = None) -> dict[str, Any]: """The feed file for state 1 or 2 as a dict (the contract shape plus `labor`).""" if n not in (1, 2): raise ValueError("state is 1 or 2") today = today or date.today() w = _work_orders() if n == 2: w["A"].rev = "2" # rev 1 -> 2 on the WELD-PART drawing (and A's LAYOUT draft) w["A2"].rev = "2" w["B"].qty = 3 # qty change (also scales tgt) w["G"].ops = [(10, "WELD-PART", 2.5)] # tgt-only change: minor del w["E"] # gone from VISUAL w["M"].dwg = "L900-1602" # moved to another drawing w["J"].st = "C" w["K"].st = "X" rows = [r for wo in w.values() for r in wo.rows(today)] mats = _materials(w) if n == 1 else _materials_state2(w) by_base, by_project = _orders(today) return { "meta": _meta(today, n), "rows": rows, "materials": mats, "orders_by_base": by_base, "orders_by_project": by_project, "labor": _labor(today), } def _materials_state2(w: dict[str, _Wo]) -> list[dict[str, Any]]: # Same lines as state 1 on the state 2 work orders (A's drawing moved with it, E is gone) full = _work_orders() full.update(w) lines = _materials(full) return [m for m in lines if m["wo"] != "L900-05"] # ------------------------------------------------------------------------------------------------ drawings def _dwg( rev: int | None, file: str | None, *, n: int, version: int = 1, draft_rev: int | None = None, draft_where: str = "", in_drafting: bool = False, base: str | None = None, ) -> dict[str, Any]: return { "rev": rev, "file": file, "name": file[:-4] if file else None, "draft_rev": draft_rev, "draft_where": draft_where, "in_drafting": in_drafting, "ok": True, "note": "", "base": base, "obj_id": n, "obj_version": version, "file_id": 100 + n, "file_version": version, } def drawings_state(n: int) -> dict[str, Any]: """drawings..json as a dict. A drawing missing from it is not in M-Files.""" if n not in (1, 2): raise ValueError("state is 1 or 2") d: dict[str, Any] = { "L900-0101": _dwg(1, "L900-0101 r1.pdf", n=1), "L900-0201": { # VISUAL calls rev 2: rev mismatch. Rev 2 is Retired in M-Files. **_dwg(3, "L900-0201 r3.pdf", n=2), "revisions": { "2": { "file": "L900-0201 r2.pdf", "status": "Retired", "state": "Superseded", "obj_id": 32, "file_id": 132, } }, }, "L900-0401": _dwg(1, "L900-0401 r1.pdf", n=4), # VISUAL calls rev A: bad rev "L900-0501": _dwg( None, None, n=5, draft_rev=1, draft_where="in Drafting review", in_drafting=True, base="L900-05" ), # not released # L900-0701 is missing: not found. L900-1601 is missing too (so its draft has no pages and can retire). "L900-0601": _dwg(1, "L900-0601 r1.pdf", n=6), "L900-0901": _dwg(1, "L900-0901 r1.pdf", n=9), "L900-1001": _dwg(1, "L900-1001 r1.pdf", n=10), "L900-1301": _dwg(1, "L900-1301 r1.pdf", n=13), "L900-1401": _dwg(1, "L900-1401 r1.pdf", n=14), "L901-0101": _dwg(1, "L901-0101 r1.pdf", n=21), } if n == 2: d["L900-0101"] = _dwg(2, "L900-0101 r2.pdf", n=1, version=2) # the CO gets staged d["L900-1602"] = _dwg(1, "L900-1602 r1.pdf", n=16) return d def pdf_bytes(label: str) -> bytes: """A one-page PDF. The label goes into the document info so each file has its own sha256.""" from pypdf import PdfWriter w = PdfWriter() w.add_blank_page(width=792, height=612) w.add_metadata({"/Title": label, "/Producer": "Packet Station fixtures"}) buf = io.BytesIO() w.write(buf) return buf.getvalue() def write_state(out_dir: Path, n: int) -> None: """Point the fixture sources at state n (what `packetstation fixtures use N` does).""" (Path(out_dir) / "state").write_text(str(n), encoding="utf-8") def generate(out_dir: Path | str, today: date | None = None) -> Path: """Write every fixture file into out_dir and set the state to 1. Returns out_dir.""" out = Path(out_dir) out.mkdir(parents=True, exist_ok=True) files: set[str] = set() for n in (1, 2): feed = feed_state(n, today) (out / f"feed.{n}.json").write_text(json.dumps(feed, indent=1, ensure_ascii=False), encoding="utf-8") dwgs = drawings_state(n) (out / f"drawings.{n}.json").write_text(json.dumps(dwgs, indent=1, ensure_ascii=False), encoding="utf-8") files.update(v["file"] for v in dwgs.values() if v.get("file")) files.update(r["file"] for v in dwgs.values() for r in (v.get("revisions") or {}).values() if r.get("file")) for name in sorted(files): (out / name).write_bytes(pdf_bytes(name)) write_state(out, 1) return out def main(argv: list[str] | None = None) -> int: ap = argparse.ArgumentParser(description="Write Packet Station's synthetic VISUAL and M-Files fixtures") ap.add_argument("--out", type=Path, required=True, help="directory to write into") ap.add_argument("--today", type=date.fromisoformat, default=None, help="date the offsets count from") args = ap.parse_args(argv) out = generate(args.out, args.today) print(f"Wrote fixtures to {out} (state 1)", file=sys.stderr) return 0 __all__ = [ "RESOURCE_MAP", "drawings_state", "feed_state", "generate", "pdf_bytes", "write_state", ] if __name__ == "__main__": sys.exit(main())