"""VISUAL rows to packet candidates: the old visBuild, typed. One candidate = unit + resource + drawing, holding every work order operation on that drawing. The key is `f"{unit}|{res}|{dwg or '(no drawing)'}"`, which is `packets.visual_key`. Project = the work order base before its first dash; unit = project + the lot zero-padded to 3. A non-numeric lot (7A) makes no candidate, as the old build: it is counted in `Built.skipped_lots`. Resources route through the resource map: a shop department code, `fold` (rides along on the packet of the same work order and lot), `ignore`, or no entry (counted as unmapped). `MAT_*` with no map entry folds. """ from __future__ import annotations from collections.abc import Collection, Iterable, Mapping, Sequence from dataclasses import dataclass, field from datetime import date from decimal import Decimal from ..text import RX_LOT, RX_PROJECT from . import materials as mats from .materials import BomLine from .types import Material, Row, WoKey FOLD = "fold" IGNORE = "ignore" NO_DRAWING = "(no drawing)" __all__ = [ "FOLD", "IGNORE", "NO_DRAWING", "BomLine", "Built", "Candidate", "Folded", "Leg", "build", "project_of", "route", "unit_number", ] @dataclass(slots=True) class Leg: """One work order operation on the packet.""" key: WoKey seq: int op: str qty: Decimal tgt: Decimal dwg: str rev: str partspecs: str st: str @dataclass(slots=True) class Folded: key: WoKey seq: int op: str tgt: Decimal @dataclass(slots=True) class Candidate: key: str # f"{unit}|{res}|{dwg or '(no drawing)'}" == packets.visual_key project: str unit: str lot: str dept: str op: str drawing: str rev: str title: str legs: list[Leg] folded: list[Folded] partspecs: list[str] rel: date | None want: date | None bom: list[BomLine] | None # None when materials are unknown for this candidate statuses: dict[WoKey, str] # header status per leg key base: str # the first leg's wo base, for the order lookup in_window: bool # at least one leg came from the window read @property def mat_known(self) -> bool: return self.bom is not None @property def leg_keys(self) -> set[WoKey]: return {lg.key for lg in self.legs} @dataclass(slots=True) class Built: candidates: dict[str, Candidate] unmapped: dict[str, int] ignored: list[Row] # rows below the go-live date (not tracked) skipped_lots: dict[str, int] = field(default_factory=dict) # f"{wo}/{lot}" -> rows (non-numeric lot) def route(res: str, resource_map: Mapping[str, str]) -> str | None: """'fold' | 'ignore' | a dept code | None (no map entry). MAT_* with no entry folds.""" if res in resource_map: return resource_map[res] if res.upper().startswith("MAT_"): return FOLD return None def project_of(wo: str) -> str: """'L489-05-5GA' -> 'L489' (upper-cased); a base with no dash is its own project.""" head = wo.partition("-")[0].strip() return (head or wo).upper() def unit_number(project: str, lot: str) -> str | None: """project + lot zero-padded to 3 when the lot is numeric; None for a non-numeric lot (the old unit_no).""" lot = lot.strip() if not lot.isdigit(): return None padded = lot.zfill(3) if not RX_LOT.match(padded): return None return project + padded def _rev_wins(new: str, cur: str) -> bool: """The old rule: a numeric rev higher than the current wins; anything wins over an empty current.""" if not new or new == cur: return False if not cur: return True try: return float(new) > float(cur) except ValueError: return False def build( rows: Sequence[Row], materials: Iterable[Material] | None, material_legs: Collection[WoKey] | None, *, resource_map: Mapping[str, str], window: tuple[date, date], golive: date | None, departments: Collection[str], tracked_keys: Collection[WoKey], today: date | None = None, ) -> Built: """Group rows into candidates. See the module docstring and phase3-design "Candidates".""" del today # kept for the contract signature; the window already carries it tracked = set(tracked_keys) depts = set(departments) by_wo_lot: dict[tuple[str, str], list[Row]] = {} for r in rows: by_wo_lot.setdefault((r.wo, r.lot), []).append(r) cands: dict[str, Candidate] = {} unmapped: dict[str, int] = {} ignored: list[Row] = [] skipped_lots: dict[str, int] = {} leg_seen: dict[str, set[tuple[WoKey, int]]] = {} fold_seen: dict[str, set[tuple[WoKey, int]]] = {} for r in rows: target = route(r.res, resource_map) if target in (FOLD, IGNORE): continue is_tracked = r.key in tracked in_window = r.st == "F" and r.rel is not None and window[0] <= r.rel <= window[1] if not is_tracked and not in_window: continue if target is None: unmapped[r.res] = unmapped.get(r.res, 0) + 1 continue if depts and target not in depts: continue if not is_tracked and golive is not None and r.rel is not None and r.rel < golive: ignored.append(r) continue project = project_of(r.wo) unit = unit_number(project, r.lot) if RX_PROJECT.match(project) else None if unit is None: k = f"{r.wo}/{r.lot}" skipped_lots[k] = skipped_lots.get(k, 0) + 1 continue key = f"{unit}|{r.res}|{r.dwg or NO_DRAWING}" c = cands.get(key) if c is None: c = Candidate( key=key, project=project, unit=unit, lot=unit[len(project) :], dept=target, op=r.res, drawing=r.dwg, rev=r.rev, title=r.desc, legs=[], folded=[], partspecs=[], rel=r.rel, want=r.want, bom=None, statuses={}, base=r.wo, in_window=False, ) cands[key] = c leg_seen[key] = set() fold_seen[key] = set() c.in_window = c.in_window or in_window if (r.key, r.seq) not in leg_seen[key]: leg_seen[key].add((r.key, r.seq)) c.legs.append( Leg(key=r.key, seq=r.seq, op=r.res, qty=r.qty, tgt=r.tgt, dwg=r.dwg, rev=r.rev, partspecs=r.ps, st=r.st) ) c.statuses[r.key] = r.st if r.rel is not None and (c.rel is None or r.rel < c.rel): c.rel = r.rel if r.want is not None and (c.want is None or r.want > c.want): c.want = r.want if _rev_wins(r.rev, c.rev): c.rev = r.rev for o in by_wo_lot.get((r.wo, r.lot), []): if route(o.res, resource_map) == FOLD and (o.key, o.seq) not in fold_seen[key]: fold_seen[key].add((o.key, o.seq)) c.folded.append(Folded(key=o.key, seq=o.seq, op=o.res, tgt=o.tgt)) if o.ps and o.ps not in c.partspecs: c.partspecs.append(o.ps) if material_legs is not None: idx = mats.index(materials) known = set(material_legs) for c in cands.values(): if any(lg.key in known for lg in c.legs): c.bom = mats.bom_for(c.legs, idx) return Built(candidates=cands, unmapped=unmapped, ignored=ignored, skipped_lots=skipped_lots)