""" qc_store.py — SQLite Quality Control store (DVI-1326 P1, board-approved plan). Backs the Togen QC tool's Batch Tickets capture: an administrable Materials list, versioned Mixture recipes with per-yd3 targets, and per-batch capture rows (one batch = one handwritten batch ticket page today; PLC/SCADA and spreadsheet imports arrive in later phases via the same `source` field). Flask-independent (mirrors assets_store.py / scada_store.py) so app.py, scripts, and tests can all import it. qc.db lives alongside the app's other state files. Design decisions (from the approved DVI-1326 plan + board answers 2026-08-08): - Quantities are stored in each material's CAPTURE unit (lb / oz / gal — what the batch ticket shows); consumption reporting converts to the material's CONSUMPTION unit (ton / lb / gal) with a tunable per-material decimal rounding, so "capture in oz, order by the lb" and "capture in lb, order by the ton" are both display-time conversions, never lossy storage. - Mixture targets are per-yd3 dosing amounts in the capture unit with -/+ tolerances (the scanned ticket's -/Target/+ columns); the batch form scales them by the batch's yards. Out-of-range NEVER blocks a save — it is flagged. - Mixtures are versioned: a targets change appends a mixture_versions row and bumps current_version; a name/active-only change does not. Batch lines snapshot the material name/unit and the scaled target/min/max at capture time, so recipe edits never rewrite captured history. - Materials are administrable (board answer #2): hard delete only when a material is unreferenced by any mixture target or batch line; otherwise deactivate (kept for history, hidden from new capture forms). Schema ------ materials(name UNIQUE, capture_unit, consumption_unit, consumption_decimals, sort, active) — the trackable materials/chemicals/VMAs. mixtures(name, segment, active, current_version) — named recipes per production segment (wet_cast / pipe / block). mixture_versions(mixture_id, version, created_at, created_by, note) — append-only recipe history. mixture_targets(version_id, material_id, target, minus, plus) — per-yd3 dosing in the material's capture unit; minus/plus are tolerance amounts below/above target. batches(batch_date, segment, batch_number, mixture_id, mixture_version, mixture_name, yards, rock_moisture, sand_moisture, initials, notes, source manual|import|plc|ocr, created_by/at, updated_by/at, deleted) — one captured batch ticket (soft-deleted so ids stay stable). batch_lines(batch_id, material_id, material_name, unit, target, min_ok, max_ok, actual, out_of_range) — snapshot per material per batch. batch_custom_values(batch_id, field_id, value) — flexible per-batch custom field values (DVI-1452 P2), id-keyed so a label rename never orphans. meta(key, value) — bookkeeping (incl. the admin-managed 'segments', 'custom_fields', 'batch_ticket_fields', … registries). """ import json import os import re import sqlite3 import threading import uuid from datetime import date as _date_cls from datetime import datetime from pathlib import Path QC_DB_FILE = Path( os.environ.get("QC_DB_FILE", Path(__file__).resolve().parent / "qc.db")) # Serialize writes within one process; SQLite handles cross-process locking. _write_lock = threading.Lock() SEGMENTS = ("wet_cast", "pipe", "block") SEGMENT_LABELS = {"wet_cast": "Wet Cast", "pipe": "Pipe Plant", "block": "Block Plant"} BATCH_SOURCES = ("manual", "import", "plc", "ocr", "daily_production") CAPTURE_UNITS = ("lb", "oz", "gal") # DVI-1423 P6: report-only aggregated materials. A material with a non-empty # aggregate_children list is an aggregator whose reported value is the SUM of # its children's converted (consumption-unit) totals — e.g. "Total Water" = # "205 Water" + "805 Water". aggregate_report_mode controls whether the # children are still itemized beside the aggregate ('both') or folded out of # the report ('aggregate_only', their capture still exists, just not itemized). # Aggregators are REPORT-ONLY: they never appear in batch entry, mixture target # grids, imports, or Daily Production — their value is always computed. Children # must share a compatible consumption unit and one nesting level only (a child # can't itself be an aggregator); both enforced at save. AGGREGATE_REPORT_MODES = ("both", "aggregate_only") # DVI-1433: an aggregate material may also carry a per-yd3 recipe target (stored # as an ordinary mixture_targets row keyed on the aggregate's material_id). A # conflict exists when the aggregate's own target and the summed child targets # disagree once both are converted to the shared consumption unit. Flag-only — # a conflicting recipe still saves (the QC-wide "flags never block" rule). CONFLICT_REL_TOL = 0.005 # 0.5% of the larger of {whole, summed-parts} CONFLICT_ABS_TOL = 1e-6 # Allowed capture->consumption unit pairs and their conversion factors. # (Capture is what the batch ticket shows; consumption is what ordering / # total-usage reports show.) Identity pairs are always allowed. _UNIT_FACTORS = { ("lb", "lb"): 1.0, ("lb", "ton"): 1.0 / 2000.0, ("oz", "oz"): 1.0, ("oz", "lb"): 1.0 / 16.0, ("oz", "gal"): 1.0 / 128.0, # fluid oz -> US gallon (board reject on 8beadb04) ("gal", "gal"): 1.0, } _DATE_RE = re.compile(r"^\d{4}-\d{2}-\d{2}$") # Seed materials from the reviewed source documents (batch ticket scan + # Master Sheet Batch Records): (name, capture_unit, consumption_unit). # Board answer #2: the three chemicals and both waters are separate materials. _SEED_MATERIALS = ( ("Fly Ash", "lb", "ton"), ("Cement", "lb", "ton"), ("Sand", "lb", "ton"), ("Rock", "lb", "ton"), ("Air", "oz", "lb"), ("2000 SCC", "oz", "lb"), ("Retarder", "oz", "lb"), ("Accelerator", "gal", "gal"), ("Water 80%", "gal", "gal"), ("Water 20%", "gal", "gal"), ) # Board answer #1: confirmed mix-code vocabulary. GREY first (most batches are # plain grey concrete; the paper form preprints it). Seeded with NO targets — # the QC manager enters real dosing per material in the Mixtures tab, and a # missing/zero target simply skips the range check on capture. _SEED_MIXTURES = ("GREY", "REG", "HE", "HHE", "BRI", "XIPEX") def _connect(): conn = sqlite3.connect(str(QC_DB_FILE), timeout=30) conn.execute("PRAGMA journal_mode=WAL") conn.execute("PRAGMA busy_timeout=10000") conn.row_factory = sqlite3.Row return conn def _now(): return datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S") def init_qc_db(): """Create tables/indexes if missing and seed defaults. Idempotent.""" with _write_lock, _connect() as conn: conn.execute( """CREATE TABLE IF NOT EXISTS materials ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL UNIQUE COLLATE NOCASE, capture_unit TEXT NOT NULL, consumption_unit TEXT NOT NULL, consumption_decimals INTEGER NOT NULL DEFAULT 0, sort INTEGER NOT NULL DEFAULT 0, active INTEGER NOT NULL DEFAULT 1, created_at TEXT, updated_at TEXT)""") conn.execute( """CREATE TABLE IF NOT EXISTS mixtures ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, segment TEXT NOT NULL, active INTEGER NOT NULL DEFAULT 1, current_version INTEGER NOT NULL DEFAULT 1, created_at TEXT, updated_at TEXT)""") conn.execute( """CREATE UNIQUE INDEX IF NOT EXISTS idx_mixtures_name ON mixtures(segment, name COLLATE NOCASE)""") conn.execute( """CREATE TABLE IF NOT EXISTS mixture_versions ( id INTEGER PRIMARY KEY AUTOINCREMENT, mixture_id INTEGER NOT NULL, version INTEGER NOT NULL, created_at TEXT, created_by TEXT, note TEXT, UNIQUE(mixture_id, version))""") conn.execute( """CREATE TABLE IF NOT EXISTS mixture_targets ( id INTEGER PRIMARY KEY AUTOINCREMENT, version_id INTEGER NOT NULL, material_id INTEGER NOT NULL, target REAL NOT NULL DEFAULT 0, minus REAL NOT NULL DEFAULT 0, plus REAL NOT NULL DEFAULT 0, UNIQUE(version_id, material_id))""") conn.execute( """CREATE TABLE IF NOT EXISTS batches ( id INTEGER PRIMARY KEY AUTOINCREMENT, batch_date TEXT NOT NULL, segment TEXT NOT NULL, batch_number TEXT NOT NULL DEFAULT '', mixture_id INTEGER, mixture_version INTEGER, mixture_name TEXT NOT NULL DEFAULT '', yards REAL NOT NULL DEFAULT 0, rock_moisture REAL, sand_moisture REAL, initials TEXT NOT NULL DEFAULT '', notes TEXT NOT NULL DEFAULT '', source TEXT NOT NULL DEFAULT 'manual', created_by TEXT, created_at TEXT, updated_by TEXT, updated_at TEXT, deleted INTEGER NOT NULL DEFAULT 0)""") conn.execute( """CREATE INDEX IF NOT EXISTS idx_batches_date ON batches(batch_date, segment, deleted)""") conn.execute( """CREATE TABLE IF NOT EXISTS batch_lines ( id INTEGER PRIMARY KEY AUTOINCREMENT, batch_id INTEGER NOT NULL, material_id INTEGER NOT NULL, material_name TEXT NOT NULL DEFAULT '', unit TEXT NOT NULL DEFAULT '', target REAL, min_ok REAL, max_ok REAL, actual REAL, out_of_range INTEGER NOT NULL DEFAULT 0, UNIQUE(batch_id, material_id))""") conn.execute( """CREATE INDEX IF NOT EXISTS idx_batch_lines_batch ON batch_lines(batch_id)""") # DVI-1452 P2: flexible per-batch custom field values. Values are keyed # by the custom field's immutable id (survives a label rename), one row # per (batch, field). Mirrors batch_lines — no schema churn per field. conn.execute( """CREATE TABLE IF NOT EXISTS batch_custom_values ( batch_id INTEGER NOT NULL, field_id TEXT NOT NULL, value TEXT NOT NULL DEFAULT '', UNIQUE(batch_id, field_id))""") conn.execute( """CREATE INDEX IF NOT EXISTS idx_batch_custom_values_batch ON batch_custom_values(batch_id)""") conn.execute( """CREATE TABLE IF NOT EXISTS meta ( key TEXT PRIMARY KEY, value TEXT)""") # DVI-1331 P3: idempotency key for imported/PLC batches. A daily Mix # Log re-upload updates the batch with a matching (date, segment, # source, import_key) instead of duplicating. Additive migration — # existing rows keep '' and behave exactly as before. cols = {r["name"] for r in conn.execute( "PRAGMA table_info(batches)").fetchall()} if "import_key" not in cols: conn.execute( "ALTER TABLE batches ADD COLUMN import_key TEXT NOT NULL " "DEFAULT ''") conn.execute( """CREATE INDEX IF NOT EXISTS idx_batches_import_key ON batches(batch_date, segment, source, import_key)""") # DVI-1423 P6: report-only aggregated materials. Additive migration — # existing materials keep '' / 'both' and behave exactly as before (a # material is an aggregator only once it has children). mcols = {r["name"] for r in conn.execute( "PRAGMA table_info(materials)").fetchall()} if "aggregate_children" not in mcols: conn.execute( "ALTER TABLE materials ADD COLUMN aggregate_children TEXT " "NOT NULL DEFAULT ''") if "aggregate_report_mode" not in mcols: conn.execute( "ALTER TABLE materials ADD COLUMN aggregate_report_mode TEXT " "NOT NULL DEFAULT 'both'") # DVI-1421 P3: mix code on mixtures — the Pipe import resolves a # segment='pipe' mixture by the Mix Log's recipe code. Additive # migration; existing mixtures keep '' and behave exactly as before. mxcols = {r["name"] for r in conn.execute( "PRAGMA table_info(mixtures)").fetchall()} if "mix_code" not in mxcols: conn.execute( "ALTER TABLE mixtures ADD COLUMN mix_code TEXT NOT NULL " "DEFAULT ''") # DVI-1440 P2 (D3-A): per-mixture material enablement. A non-versioned # JSON list of leaf material ids the mixture uses — capture forms filter # to it. Additive migration; existing mixtures keep '' = unset, which # falls back to the D4-A legacy default ("materials that currently have # targets"). Dosing targets keep their versioned history unchanged. if "included_materials" not in mxcols: conn.execute( "ALTER TABLE mixtures ADD COLUMN included_materials TEXT " "NOT NULL DEFAULT ''") # DVI-1440 review #2: JSON list of AGGREGATE material ids that this # mixture captures DIRECTLY (the operator keys the aggregate total on the # capture form; its constituent materials are NOT captured). Absent/'' # = every enabled aggregate is summed from its captured constituents # (the DVI-1423 report-only default). Additive, non-versioned — same # idempotent pattern as included_materials. if "aggregate_capture" not in mxcols: conn.execute( "ALTER TABLE mixtures ADD COLUMN aggregate_capture TEXT " "NOT NULL DEFAULT ''") # DVI-1452 P1: production segments are now an admin-managed registry in # meta, replacing the hardcoded SEGMENTS/SEGMENT_LABELS constants. # Idempotent seed: only written when absent, from the legacy constants, # so a live DB gets wet_cast/pipe/block with zero behavior change. if _get_meta(conn, _SEGMENTS_META_KEY) is None: _set_meta(conn, _SEGMENTS_META_KEY, json.dumps(_seed_segments())) _seed_defaults(conn) conn.commit() def _seed_defaults(conn): """Seed materials + mixture names only into EMPTY tables (never re-adds an admin-deleted row on restart).""" now = _now() if conn.execute("SELECT COUNT(*) FROM materials").fetchone()[0] == 0: for i, (name, cap, cons) in enumerate(_SEED_MATERIALS): conn.execute( """INSERT INTO materials(name, capture_unit, consumption_unit, consumption_decimals, sort, active, created_at, updated_at) VALUES(?,?,?,0,?,1,?,?)""", (name, cap, cons, (i + 1) * 10, now, now)) if conn.execute("SELECT COUNT(*) FROM mixtures").fetchone()[0] == 0: for name in _SEED_MIXTURES: cur = conn.execute( """INSERT INTO mixtures(name, segment, active, current_version, created_at, updated_at) VALUES(?,?,1,1,?,?)""", (name, "wet_cast", now, now)) conn.execute( """INSERT INTO mixture_versions(mixture_id, version, created_at, created_by, note) VALUES(?,1,?,?,?)""", (cur.lastrowid, now, "seed", "Seeded mixture — targets pending")) # --------------------------------------------------------------------------- # Units def convert_amount(value, capture_unit, consumption_unit): """Capture-unit value -> consumption-unit value (None on unknown pair).""" factor = _UNIT_FACTORS.get((capture_unit, consumption_unit)) if factor is None or value is None: return None return value * factor def valid_unit_pair(capture_unit, consumption_unit): return (capture_unit, consumption_unit) in _UNIT_FACTORS # --------------------------------------------------------------------------- # Materials def _parse_children(raw): """Parse a stored aggregate_children value (JSON list of material ids) into a de-duplicated list of ints. Tolerant: '' / bad JSON / non-list -> [].""" if not raw: return [] try: vals = json.loads(raw) except (TypeError, ValueError): return [] if not isinstance(vals, list): return [] out, seen = [], set() for v in vals: try: i = int(v) except (TypeError, ValueError): continue if i not in seen: seen.add(i) out.append(i) return out def _parse_included(raw): """Parse a stored included_materials value (DVI-1440 P2). '' / None -> None, meaning UNSET — the D4-A legacy default applies (membership = materials that currently have targets). A valid JSON list -> de-duplicated list of ints (possibly []); bad JSON / non-list -> None (treated as unset).""" if raw is None or raw == "": return None try: vals = json.loads(raw) except (TypeError, ValueError): return None if not isinstance(vals, list): return None out, seen = [], set() for v in vals: try: i = int(v) except (TypeError, ValueError): continue if i not in seen: seen.add(i) out.append(i) return out def _effective_included_ids(included, targets): """The mixture's effective capture membership as a set of material ids. Explicit list wins (D3-A); an unset list (None) falls back to the D4-A legacy default = materials that currently have targets. A legacy mixture with NO targets resolves to the empty set — callers that must still capture (batch/scan/daily) treat that empty legacy set as 'all materials'.""" if included is not None: return set(included) return {t["material_id"] for t in (targets or [])} def _parse_id_list(raw): """Parse a stored JSON id list where '' / None / bad JSON all mean the empty list (unlike _parse_included, whose None means 'unset / legacy default'). Used for aggregate_capture (DVI-1440 review #2).""" return _parse_included(raw) or [] def _capture_plan(direct_agg_ids, mats_by_id): """Resolve a mixture's aggregate capture modes into a (direct, suppressed) pair used by every capture surface (batch / scan / daily). ``direct`` = aggregate ids captured DIRECTLY (they become a capture line; the operator keys the aggregate total). ``suppressed`` = the child leaf ids of those direct aggregates — they are NOT captured on this mixture (the aggregate stands in for them). An aggregate not in ``direct_agg_ids`` keeps the DVI-1423 report-only default (summed from its captured constituents), so its children stay capturable. ``mats_by_id`` maps id -> a _material_row-shaped dict (needs is_aggregate + aggregate_children).""" direct, suppressed = set(), set() for aid in (direct_agg_ids or []): m = mats_by_id.get(aid) if m and m.get("is_aggregate"): direct.add(aid) for cid in (m.get("aggregate_children") or []): suppressed.add(cid) return direct, suppressed def _row_get(row, key, default=None): """sqlite3.Row has no .get(); read a possibly-absent column safely (rows read before the DVI-1423 migration lack the aggregate columns in tests).""" return row[key] if key in row.keys() else default def _material_row(row): children = _parse_children(_row_get(row, "aggregate_children", "")) mode = (_row_get(row, "aggregate_report_mode", "both") or "both") if mode not in AGGREGATE_REPORT_MODES: mode = "both" return { "id": row["id"], "name": row["name"], "capture_unit": row["capture_unit"], "consumption_unit": row["consumption_unit"], "consumption_decimals": row["consumption_decimals"], "sort": row["sort"], "active": bool(row["active"]), # DVI-1423 P6 aggregate config (leaf materials carry [] / 'both'). "aggregate_children": children, "aggregate_report_mode": mode, "is_aggregate": bool(children), } def list_materials(include_inactive=False, include_aggregates=True): """List materials. include_aggregates=False drops report-only aggregators (DVI-1423) — used by the capture-facing surfaces (batch entry, mixture target grids, imports) where aggregators must never appear.""" with _connect() as conn: sql = "SELECT * FROM materials" if not include_inactive: sql += " WHERE active=1" sql += " ORDER BY sort, name" rows = [_material_row(r) for r in conn.execute(sql).fetchall()] if not include_aggregates: rows = [r for r in rows if not r["is_aggregate"]] return rows def get_material(material_id): with _connect() as conn: row = conn.execute("SELECT * FROM materials WHERE id=?", (material_id,)).fetchone() return _material_row(row) if row else None def list_aggregates(include_inactive=False): """DVI-1433: report-only aggregators, each with its leaf children resolved to {id, name, capture_unit, consumption_unit}. Feeds the Mixtures editor's 'Aggregated totals' section (which materials each aggregate totals + the aggregate's own target). Distinct from list_materials(include_aggregates=False), which the capture surfaces read and which never includes aggregators.""" with _connect() as conn: mats = {r["id"]: _material_row(r) for r in conn.execute( "SELECT * FROM materials ORDER BY sort, name").fetchall()} out = [] for m in mats.values(): if not m["is_aggregate"]: continue if not include_inactive and not m["active"]: continue children = [] for cid in m["aggregate_children"]: c = mats.get(cid) if c is None: continue children.append({"id": c["id"], "name": c["name"], "capture_unit": c["capture_unit"], "consumption_unit": c["consumption_unit"]}) out.append({ "id": m["id"], "name": m["name"], "capture_unit": m["capture_unit"], "consumption_unit": m["consumption_unit"], "consumption_decimals": m["consumption_decimals"], "aggregate_report_mode": m["aggregate_report_mode"], "children": children, }) return out def mixture_target_conflicts(targets, mats_by_id): """Pure: given a mixture's target rows ({material_id, target, ...}, values in each material's CAPTURE unit) and a {id: material} map, return the list of aggregate-vs-parts conflicts. A conflict exists for aggregate A when A has a target > 0 AND >=1 child has a target > 0 AND the aggregate target and the summed child targets disagree once converted to the shared consumption unit: |conv(target_A) - sum conv(target_Ci)| > max(1e-6, 0.5% * larger). Converting to the consumption unit is exact because children share A's consumption unit (validated at material save). Setting only the whole, or only the children, is never a conflict. Each conflict dict carries the converted whole/parts values + delta + unit for display.""" tget = {} for t in (targets or []): try: mid = int(t.get("material_id")) except (TypeError, ValueError): continue try: tget[mid] = max(0.0, float(t.get("target") or 0)) except (TypeError, ValueError): tget[mid] = 0.0 out = [] for mid, whole_cap in tget.items(): mat = mats_by_id.get(mid) if not mat or not mat.get("aggregate_children") or whole_cap <= 0: continue whole = convert_amount(whole_cap, mat["capture_unit"], mat["consumption_unit"]) if whole is None: continue parts = 0.0 n_children = 0 for cid in mat["aggregate_children"]: child_cap = tget.get(cid, 0.0) if child_cap <= 0: continue child = mats_by_id.get(cid) if not child: continue conv = convert_amount(child_cap, child["capture_unit"], child["consumption_unit"]) if conv is None: continue parts += conv n_children += 1 if n_children == 0: continue delta = abs(whole - parts) tol = max(CONFLICT_ABS_TOL, CONFLICT_REL_TOL * max(whole, parts)) if delta > tol: dec = mat["consumption_decimals"] out.append({ "aggregate_id": mid, "aggregate_name": mat["name"], "unit": mat["consumption_unit"], "whole": round(whole, dec), "parts": round(parts, dec), "delta": round(delta, dec), }) return out def save_material(fields, material_id=None): """Create or update a material. Returns (material, error).""" name = str(fields.get("name") or "").strip()[:80] if not name: return None, "Material name is required." cap = str(fields.get("capture_unit") or "").strip().lower() cons = str(fields.get("consumption_unit") or "").strip().lower() if cap not in CAPTURE_UNITS: return None, "Capture unit must be one of: %s." % ", ".join(CAPTURE_UNITS) if not valid_unit_pair(cap, cons): allowed = sorted(c for (k, c) in _UNIT_FACTORS if k == cap) return None, "Consumption unit for %s must be one of: %s." % ( cap, ", ".join(allowed)) try: decimals = max(0, min(3, int(fields.get("consumption_decimals") or 0))) except (TypeError, ValueError): decimals = 0 try: sort = int(fields.get("sort") or 0) except (TypeError, ValueError): sort = 0 active = 1 if fields.get("active", True) else 0 # DVI-1423 P6: aggregate config. Children come in as a list of material ids. children = [] seen = set() for v in (fields.get("aggregate_children") or []): try: i = int(v) except (TypeError, ValueError): continue if i not in seen: seen.add(i) children.append(i) mode = str(fields.get("aggregate_report_mode") or "both").strip().lower() if mode not in AGGREGATE_REPORT_MODES: mode = "both" now = _now() with _write_lock, _connect() as conn: dup = conn.execute( "SELECT id FROM materials WHERE name=? COLLATE NOCASE", (name,)).fetchone() if dup and (material_id is None or dup["id"] != material_id): return None, "A material named '%s' already exists." % name if material_id is not None and not conn.execute( "SELECT id FROM materials WHERE id=?", (material_id,)).fetchone(): return None, "Material not found." # Validate the aggregate config (Q4 constraints) before writing. if children: if material_id is not None: children = [c for c in children if c != material_id] if not children: return None, "An aggregate must reference at least one other material." all_rows = {r["id"]: r for r in conn.execute("SELECT * FROM materials").fetchall()} for cid in children: crow = all_rows.get(cid) if crow is None: return None, "An aggregate child material no longer exists." if _parse_children(_row_get(crow, "aggregate_children", "")): return None, ("Aggregates can only contain leaf materials (one " "level of nesting) — '%s' is itself an aggregate." % crow["name"]) if crow["consumption_unit"] != cons: return None, ("Every aggregate child must consume in %s to match " "this material — '%s' consumes in %s." % ( cons, crow["name"], crow["consumption_unit"])) if material_id is not None: refs = conn.execute( "SELECT COUNT(*) FROM batch_lines WHERE material_id=?", (material_id,)).fetchone()[0] refs += conn.execute( "SELECT COUNT(*) FROM mixture_targets WHERE material_id=?", (material_id,)).fetchone()[0] if refs: return None, ("This material has captured batches or recipe " "targets and can't become a report-only aggregate " "— create a new material instead.") for r in all_rows.values(): if r["id"] == material_id: continue if material_id in _parse_children( _row_get(r, "aggregate_children", "")): return None, ("This material is a child of aggregate '%s' and " "can't also be an aggregate (one level of " "nesting)." % r["name"]) children_json = json.dumps(children) if children else "" report_mode = mode if children else "both" if material_id is None: cur = conn.execute( """INSERT INTO materials(name, capture_unit, consumption_unit, consumption_decimals, sort, active, aggregate_children, aggregate_report_mode, created_at, updated_at) VALUES(?,?,?,?,?,?,?,?,?,?)""", (name, cap, cons, decimals, sort, active, children_json, report_mode, now, now)) material_id = cur.lastrowid else: conn.execute( """UPDATE materials SET name=?, capture_unit=?, consumption_unit=?, consumption_decimals=?, sort=?, active=?, aggregate_children=?, aggregate_report_mode=?, updated_at=? WHERE id=?""", (name, cap, cons, decimals, sort, active, children_json, report_mode, now, material_id)) conn.commit() return get_material(material_id), None def delete_material(material_id): """Hard delete only when unreferenced (board answer #2); returns error text otherwise so the admin deactivates instead.""" with _write_lock, _connect() as conn: if not conn.execute("SELECT id FROM materials WHERE id=?", (material_id,)).fetchone(): return "Material not found." refs = conn.execute( "SELECT COUNT(*) FROM batch_lines WHERE material_id=?", (material_id,)).fetchone()[0] refs += conn.execute( "SELECT COUNT(*) FROM mixture_targets WHERE material_id=?", (material_id,)).fetchone()[0] if refs: return ("Material is referenced by captured batches or mixture " "recipes — deactivate it instead of deleting.") # DVI-1423: block deleting a material still folded into an aggregate. for r in conn.execute( "SELECT name, aggregate_children FROM materials WHERE id!=?", (material_id,)).fetchall(): if material_id in _parse_children(r["aggregate_children"]): return ("Material is a child of aggregate '%s' — remove it from " "the aggregate before deleting." % r["name"]) conn.execute("DELETE FROM materials WHERE id=?", (material_id,)) conn.commit() return None # --------------------------------------------------------------------------- # Mixtures def _targets_for_version(conn, version_id): rows = conn.execute( """SELECT t.material_id, t.target, t.minus, t.plus FROM mixture_targets t WHERE t.version_id=? ORDER BY t.material_id""", (version_id,)).fetchall() return [{"material_id": r["material_id"], "target": r["target"], "minus": r["minus"], "plus": r["plus"]} for r in rows] def _version_row_id(conn, mixture_id, version): row = conn.execute( "SELECT id FROM mixture_versions WHERE mixture_id=? AND version=?", (mixture_id, version)).fetchone() return row["id"] if row else None def _norm_mix_code(value): """Normalize a mix code for matching (Mix Log recipe cell <-> mixture mix_code). Case-insensitive, whitespace-trimmed; '' when blank.""" return str(value if value is not None else "").strip().lower() def _mats_by_id(conn): """{material_id: material} over ALL materials — needed for the DVI-1433 conflict math which spans an aggregate and its leaf children.""" return {r["id"]: _material_row(r) for r in conn.execute("SELECT * FROM materials").fetchall()} def _mixture_row(conn, row, with_targets=True, mats_by_id=None): out = { "id": row["id"], "name": row["name"], "segment": row["segment"], "active": bool(row["active"]), "version": row["current_version"], "mix_code": _row_get(row, "mix_code", "") or "", # DVI-1440 P2: raw membership list, or None when unset (D4-A default). "included_materials": _parse_included( _row_get(row, "included_materials", "")), # DVI-1440 review #2: aggregate ids captured directly on this mixture # (a capture line, constituents suppressed). '' -> []. "aggregate_capture": _parse_id_list( _row_get(row, "aggregate_capture", "")), "updated_at": row["updated_at"], } if with_targets: vid = _version_row_id(conn, row["id"], row["current_version"]) targets = _targets_for_version(conn, vid) if vid else [] out["targets"] = targets # DVI-1433: server-authoritative aggregate target conflicts. if mats_by_id is None: mats_by_id = _mats_by_id(conn) out["conflicts"] = mixture_target_conflicts(targets, mats_by_id) return out def list_mixtures(segment=None, include_inactive=False, with_targets=True): with _connect() as conn: sql, args = "SELECT * FROM mixtures", [] conds = [] if segment: conds.append("segment=?") args.append(segment) if not include_inactive: conds.append("active=1") if conds: sql += " WHERE " + " AND ".join(conds) sql += " ORDER BY name COLLATE NOCASE" mats_by_id = _mats_by_id(conn) if with_targets else None return [_mixture_row(conn, r, with_targets, mats_by_id) for r in conn.execute(sql, args).fetchall()] def get_mixture(mixture_id): with _connect() as conn: row = conn.execute("SELECT * FROM mixtures WHERE id=?", (mixture_id,)).fetchone() return _mixture_row(conn, row) if row else None def _sanitize_targets(targets): """Normalize a targets list; drops all-zero/unknown rows. Returns (rows, error).""" out, seen = [], set() for raw in (targets or []): if not isinstance(raw, dict): continue try: mid = int(raw.get("material_id")) except (TypeError, ValueError): continue vals = {} for key in ("target", "minus", "plus"): try: v = float(raw.get(key) or 0) except (TypeError, ValueError): v = 0.0 vals[key] = max(0.0, v) if mid in seen: continue seen.add(mid) if vals["target"] == 0 and vals["minus"] == 0 and vals["plus"] == 0: continue out.append({"material_id": mid, "target": vals["target"], "minus": vals["minus"], "plus": vals["plus"]}) return out, None def save_mixture(fields, mixture_id=None, actor=""): """Create or update a mixture. A targets change appends a new version; a name/active-only change does not. Returns (mixture, error).""" name = str(fields.get("name") or "").strip()[:80] if not name: return None, "Mixture name is required." segment = str(fields.get("segment") or "wet_cast").strip() if not is_segment(segment): return None, "Unknown segment." active = 1 if fields.get("active", True) else 0 # DVI-1421 P3: mix code (used by the Pipe import to resolve a mixture by the # Mix Log recipe code). Optional; validated unique per segment when set. mix_code = str(fields.get("mix_code") or "").strip()[:40] targets, err = _sanitize_targets(fields.get("targets")) if err: return None, err # DVI-1440 P2: per-mixture material membership. Absent key -> None (leave the # stored value untouched on update; '' on create = unset/legacy). A provided # list -> normalized ints (filtered to known leaf materials below). raw_incl = fields.get("included_materials") included_ids = None if raw_incl is not None: included_ids, seen_i = [], set() for v in (raw_incl if isinstance(raw_incl, list) else []): try: i = int(v) except (TypeError, ValueError): continue if i not in seen_i: seen_i.add(i) included_ids.append(i) # DVI-1440 review #2: aggregate ids captured directly (absent key -> leave # the stored value untouched on update / '' on create, like membership). raw_agg = fields.get("aggregate_capture") agg_capture_ids = None if raw_agg is not None: agg_capture_ids, seen_a = [], set() for v in (raw_agg if isinstance(raw_agg, list) else []): try: i = int(v) except (TypeError, ValueError): continue if i not in seen_a: seen_a.add(i) agg_capture_ids.append(i) now = _now() with _write_lock, _connect() as conn: mat_rows = conn.execute("SELECT id FROM materials").fetchall() known = {r["id"] for r in mat_rows} bad = [t for t in targets if t["material_id"] not in known] if bad: return None, "Unknown material in targets." included_json = None incl_set = None if included_ids is not None: # DVI-1440 review: membership covers leaf AND report-only aggregate # materials — an aggregate total is enable-able per mixture just like # a leaf. Filter to known ids (drop stale/deleted). Capture surfaces # iterate leaves only, so aggregate ids here never leak onto them. included_ids = [i for i in included_ids if i in known] included_json = json.dumps(included_ids) # Enforce targets subset of membership: a target for an unchecked # material — leaf OR aggregate — is dropped (unchecking removes it, # normal version bump). incl_set = set(included_ids) targets = [t for t in targets if t["material_id"] in incl_set] # DVI-1440 review #2: an aggregate is captured directly only when it is a # known aggregate AND enabled (in membership when membership is provided # this save; otherwise the editor's membership stands). Filter accordingly. agg_capture_json = None if agg_capture_ids is not None: mats_by_id = _mats_by_id(conn) agg_capture_ids = [ i for i in agg_capture_ids if mats_by_id.get(i) and mats_by_id[i]["is_aggregate"] and (incl_set is None or i in incl_set)] agg_capture_json = json.dumps(agg_capture_ids) dup = conn.execute( "SELECT id FROM mixtures WHERE segment=? AND name=? COLLATE NOCASE", (segment, name)).fetchone() if dup and (mixture_id is None or dup["id"] != mixture_id): return None, "A mixture named '%s' already exists for %s." % ( name, _segment_label_conn(conn, segment)) if mix_code: code_dup = conn.execute( "SELECT id, name FROM mixtures WHERE segment=? " "AND mix_code=? COLLATE NOCASE", (segment, mix_code)).fetchone() if code_dup and (mixture_id is None or code_dup["id"] != mixture_id): return None, ("Mix code '%s' is already used by '%s' for %s." % ( mix_code, code_dup["name"], _segment_label_conn(conn, segment))) if mixture_id is None: cur = conn.execute( """INSERT INTO mixtures(name, segment, active, current_version, mix_code, included_materials, aggregate_capture, created_at, updated_at) VALUES(?,?,?,1,?,?,?,?,?)""", (name, segment, active, mix_code, included_json if included_json is not None else "", agg_capture_json if agg_capture_json is not None else "", now, now)) mixture_id = cur.lastrowid cur = conn.execute( """INSERT INTO mixture_versions(mixture_id, version, created_at, created_by, note) VALUES(?,1,?,?,?)""", (mixture_id, now, actor, "Created")) vid = cur.lastrowid for t in targets: conn.execute( """INSERT INTO mixture_targets(version_id, material_id, target, minus, plus) VALUES(?,?,?,?,?)""", (vid, t["material_id"], t["target"], t["minus"], t["plus"])) else: row = conn.execute("SELECT * FROM mixtures WHERE id=?", (mixture_id,)).fetchone() if not row: return None, "Mixture not found." cur_vid = _version_row_id(conn, mixture_id, row["current_version"]) existing = _targets_for_version(conn, cur_vid) if cur_vid else [] changed = ( sorted((t["material_id"], t["target"], t["minus"], t["plus"]) for t in targets) != sorted((t["material_id"], t["target"], t["minus"], t["plus"]) for t in existing)) new_version = row["current_version"] if changed: new_version = row["current_version"] + 1 cur = conn.execute( """INSERT INTO mixture_versions(mixture_id, version, created_at, created_by, note) VALUES(?,?,?,?,?)""", (mixture_id, new_version, now, actor, "Targets updated")) vid = cur.lastrowid for t in targets: conn.execute( """INSERT INTO mixture_targets(version_id, material_id, target, minus, plus) VALUES(?,?,?,?,?)""", (vid, t["material_id"], t["target"], t["minus"], t["plus"])) # DVI-1439 P1 (D2-A): back-fill the new targets onto existing # target-less Daily Production entries of this mixture. _apply_targets_to_daily_production(conn, mixture_id, new_version) # Absent membership / aggregate-capture keys leave those columns # untouched (non-versioned, edited independently of targets). set_cols = ["name=?", "active=?", "current_version=?", "mix_code=?", "updated_at=?"] set_args = [name, active, new_version, mix_code, now] if included_json is not None: set_cols.insert(4, "included_materials=?") set_args.insert(4, included_json) if agg_capture_json is not None: set_cols.insert(len(set_cols) - 1, "aggregate_capture=?") set_args.insert(len(set_args) - 1, agg_capture_json) set_args.append(mixture_id) conn.execute( "UPDATE mixtures SET " + ", ".join(set_cols) + " WHERE id=?", set_args) conn.commit() return get_mixture(mixture_id), None def mixture_versions(mixture_id): with _connect() as conn: rows = conn.execute( """SELECT * FROM mixture_versions WHERE mixture_id=? ORDER BY version DESC""", (mixture_id,)).fetchall() return [{"version": r["version"], "created_at": r["created_at"], "created_by": r["created_by"], "note": r["note"], "targets": _targets_for_version(conn, r["id"])} for r in rows] # --------------------------------------------------------------------------- # Batches def _scaled_targets(conn, mixture_id, version, yards): """Per-material {target,min_ok,max_ok} scaled by yards for a mixture version (empty map when the mixture has no targets).""" vid = _version_row_id(conn, mixture_id, version) if mixture_id else None out = {} # DVI-1421 P3: a yards-less batch (e.g. an imported row with no yardage) # stays target-less even when a mixture is attached — scaling by 0 would # otherwise collapse the range to [0,0] and flag every line out of range. if not vid or not yards or yards <= 0: return out for t in _targets_for_version(conn, vid): tgt = t["target"] * yards if t["minus"] == 0 and t["plus"] == 0: # Target-only line (no tolerances, DVI-1420): keep the scaled # target for reporting/entry hints, but leave the acceptable range # NULL so the line is never flagged out of range. out[t["material_id"]] = { "target": tgt, "min_ok": None, "max_ok": None} else: out[t["material_id"]] = { "target": tgt, "min_ok": max(0.0, (t["target"] - t["minus"]) * yards), "max_ok": (t["target"] + t["plus"]) * yards, } return out def _apply_targets_to_daily_production(conn, mixture_id, version): """DVI-1439 P1 (D2-A) retroactive targets: when a mixture gains/changes its per-yd3 targets, back-fill the newly scaled target/min/max onto EXISTING Daily Production entries of that mixture — but ONLY on lines captured target-less (``target IS NULL``); an existing snapshot or captured actual is never overwritten. Out-of-range is recomputed on the back-filled lines. A batch whose lines are back-filled has its ``mixture_version`` advanced so its header matches the snapshots. Runs inside the caller's write transaction (the new version + targets are already inserted, so ``_scaled_targets`` sees them). (In-place edit/resubmit already re-snapshots current targets — unchanged.)""" batches = conn.execute( "SELECT id, yards FROM batches WHERE mixture_id=? " "AND source='daily_production' AND deleted=0", (mixture_id,)).fetchall() for b in batches: scaled = _scaled_targets(conn, mixture_id, version, b["yards"]) if not scaled: # no targets, or yards<=0 (stays bare) continue touched = False rows = conn.execute( "SELECT id, material_id, actual FROM batch_lines " "WHERE batch_id=? AND target IS NULL", (b["id"],)).fetchall() for ln in rows: snap = scaled.get(ln["material_id"]) if not snap: continue actual = ln["actual"] oor = 0 if (actual is not None and snap["min_ok"] is not None and snap["max_ok"] is not None and not (snap["min_ok"] <= actual <= snap["max_ok"])): oor = 1 conn.execute( "UPDATE batch_lines SET target=?, min_ok=?, max_ok=?, " "out_of_range=? WHERE id=?", (snap["target"], snap["min_ok"], snap["max_ok"], oor, ln["id"])) touched = True if touched: conn.execute("UPDATE batches SET mixture_version=? WHERE id=?", (version, b["id"])) def _sanitize_batch_fields(fields): """Validate/coerce batch header fields. Returns (clean, error).""" clean = {} date = str(fields.get("batch_date") or "").strip() if not _DATE_RE.match(date): return None, "batch_date must be YYYY-MM-DD." clean["batch_date"] = date segment = str(fields.get("segment") or "wet_cast").strip() if not is_segment(segment): return None, "Unknown segment." clean["segment"] = segment clean["batch_number"] = str(fields.get("batch_number") or "").strip()[:40] try: clean["yards"] = max(0.0, float(fields.get("yards") or 0)) except (TypeError, ValueError): return None, "yards must be a number." for key in ("rock_moisture", "sand_moisture"): raw = fields.get(key) if raw in (None, ""): clean[key] = None else: try: clean[key] = float(raw) except (TypeError, ValueError): return None, "%s must be a number." % key clean["initials"] = str(fields.get("initials") or "").strip()[:20] clean["notes"] = str(fields.get("notes") or "").strip()[:2000] source = str(fields.get("source") or "manual").strip() clean["source"] = source if source in BATCH_SOURCES else "manual" mid = fields.get("mixture_id") try: clean["mixture_id"] = int(mid) if mid not in (None, "") else None except (TypeError, ValueError): clean["mixture_id"] = None return clean, None def _build_lines(conn, clean, lines): """Snapshot batch lines from the mixture's CURRENT version targets scaled by yards. Returns (line_rows, mixture_name, mixture_version, error).""" mix_name, mix_version = "", None mrow = None if clean["mixture_id"] is not None: mrow = conn.execute("SELECT * FROM mixtures WHERE id=?", (clean["mixture_id"],)).fetchone() if not mrow: return None, None, None, "Mixture not found." mix_name, mix_version = mrow["name"], mrow["current_version"] scaled = _scaled_targets(conn, clean["mixture_id"], mix_version, clean["yards"]) if mix_version else {} materials = {r["id"]: r for r in conn.execute("SELECT * FROM materials")} # DVI-1440 review #2: which aggregates this mixture captures directly (they # become a capture line; their child leaves are suppressed). mats_meta = {i: _material_row(r) for i, r in materials.items()} direct, suppressed = _capture_plan( _parse_id_list(_row_get(mrow, "aggregate_capture", "")) if mrow else [], mats_meta) out, seen = [], set() for raw in (lines or []): if not isinstance(raw, dict): continue try: mid = int(raw.get("material_id")) except (TypeError, ValueError): continue mat = materials.get(mid) if mat is None or mid in seen: continue if mats_meta[mid]["is_aggregate"]: # An aggregator is report-only unless this mixture captures it # directly (DVI-1440 review #2). if mid not in direct: continue elif mid in suppressed: # A constituent of a direct-capture aggregate is not captured here. continue seen.add(mid) actual_raw = raw.get("actual") if actual_raw in (None, ""): actual = None else: try: actual = max(0.0, float(actual_raw)) except (TypeError, ValueError): return None, None, None, "Actual for %s must be a number." % mat["name"] snap = scaled.get(mid) oor = 0 # A target-only line (DVI-1420) has min_ok/max_ok NULL → never flagged. if (snap and actual is not None and snap["min_ok"] is not None and snap["max_ok"] is not None and not (snap["min_ok"] <= actual <= snap["max_ok"])): oor = 1 out.append({ "material_id": mid, "material_name": mat["name"], "unit": mat["capture_unit"], "target": snap["target"] if snap else None, "min_ok": snap["min_ok"] if snap else None, "max_ok": snap["max_ok"] if snap else None, "actual": actual, "out_of_range": oor, }) return out, mix_name, mix_version, None def create_batch(fields, lines, actor="", custom_values=None): """Capture one batch ticket. Returns (batch, error). ``custom_values`` is an optional {field_id: value} map for admin-defined custom fields (DVI-1452 P2); values are sanitized to the field type and never block the save.""" clean, err = _sanitize_batch_fields(fields) if err: return None, err if custom_values is None: custom_values = fields.get("custom_values") if isinstance(fields, dict) else None now = _now() with _write_lock, _connect() as conn: line_rows, mix_name, mix_version, err = _build_lines(conn, clean, lines) if err: return None, err cvals = _sanitize_custom_values(conn, custom_values) cur = conn.execute( """INSERT INTO batches(batch_date, segment, batch_number, mixture_id, mixture_version, mixture_name, yards, rock_moisture, sand_moisture, initials, notes, source, created_by, created_at, updated_by, updated_at, deleted) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,0)""", (clean["batch_date"], clean["segment"], clean["batch_number"], clean["mixture_id"], mix_version, mix_name, clean["yards"], clean["rock_moisture"], clean["sand_moisture"], clean["initials"], clean["notes"], clean["source"], actor, now, actor, now)) batch_id = cur.lastrowid for ln in line_rows: conn.execute( """INSERT INTO batch_lines(batch_id, material_id, material_name, unit, target, min_ok, max_ok, actual, out_of_range) VALUES(?,?,?,?,?,?,?,?,?)""", (batch_id, ln["material_id"], ln["material_name"], ln["unit"], ln["target"], ln["min_ok"], ln["max_ok"], ln["actual"], ln["out_of_range"])) _write_custom_values(conn, batch_id, cvals) conn.commit() return get_batch(batch_id), None def update_batch(batch_id, fields, lines, actor="", custom_values=None): """Edit a captured batch (always allowed — the issue requires editable submissions). Lines are re-snapshotted from the mixture's current version (mixture/yards may have been corrected). Returns (batch, error).""" clean, err = _sanitize_batch_fields(fields) if err: return None, err if custom_values is None: custom_values = fields.get("custom_values") if isinstance(fields, dict) else None now = _now() with _write_lock, _connect() as conn: row = conn.execute("SELECT * FROM batches WHERE id=? AND deleted=0", (batch_id,)).fetchone() if not row: return None, "Batch not found." line_rows, mix_name, mix_version, err = _build_lines(conn, clean, lines) if err: return None, err cvals = _sanitize_custom_values(conn, custom_values) conn.execute( """UPDATE batches SET batch_date=?, segment=?, batch_number=?, mixture_id=?, mixture_version=?, mixture_name=?, yards=?, rock_moisture=?, sand_moisture=?, initials=?, notes=?, updated_by=?, updated_at=? WHERE id=?""", (clean["batch_date"], clean["segment"], clean["batch_number"], clean["mixture_id"], mix_version, mix_name, clean["yards"], clean["rock_moisture"], clean["sand_moisture"], clean["initials"], clean["notes"], actor, now, batch_id)) conn.execute("DELETE FROM batch_lines WHERE batch_id=?", (batch_id,)) for ln in line_rows: conn.execute( """INSERT INTO batch_lines(batch_id, material_id, material_name, unit, target, min_ok, max_ok, actual, out_of_range) VALUES(?,?,?,?,?,?,?,?,?)""", (batch_id, ln["material_id"], ln["material_name"], ln["unit"], ln["target"], ln["min_ok"], ln["max_ok"], ln["actual"], ln["out_of_range"])) _write_custom_values(conn, batch_id, cvals) conn.commit() return get_batch(batch_id), None def delete_batch(batch_id, actor=""): """Soft delete (ids stay stable; excluded from lists/totals).""" with _write_lock, _connect() as conn: row = conn.execute("SELECT id FROM batches WHERE id=? AND deleted=0", (batch_id,)).fetchone() if not row: return "Batch not found." conn.execute( "UPDATE batches SET deleted=1, updated_by=?, updated_at=? WHERE id=?", (actor, _now(), batch_id)) conn.commit() return None def _batch_header(row): return { "id": row["id"], "batch_date": row["batch_date"], "segment": row["segment"], "batch_number": row["batch_number"], "mixture_id": row["mixture_id"], "mixture_version": row["mixture_version"], "mixture_name": row["mixture_name"], "yards": row["yards"], "rock_moisture": row["rock_moisture"], "sand_moisture": row["sand_moisture"], "initials": row["initials"], "notes": row["notes"], "source": row["source"], "import_key": (row["import_key"] if "import_key" in row.keys() else ""), "created_by": row["created_by"], "created_at": row["created_at"], "updated_by": row["updated_by"], "updated_at": row["updated_at"], } def get_batch(batch_id): with _connect() as conn: row = conn.execute("SELECT * FROM batches WHERE id=? AND deleted=0", (batch_id,)).fetchone() if not row: return None out = _batch_header(row) out["lines"] = [ {"material_id": r["material_id"], "material_name": r["material_name"], "unit": r["unit"], "target": r["target"], "min_ok": r["min_ok"], "max_ok": r["max_ok"], "actual": r["actual"], "out_of_range": bool(r["out_of_range"])} for r in conn.execute( """SELECT * FROM batch_lines WHERE batch_id=? ORDER BY id""", (batch_id,)).fetchall()] out["custom_values"] = _load_custom_values(conn, batch_id) return out def list_batches(batch_date, segment=None): """Batches for one day (newest first) with line summaries.""" with _connect() as conn: sql = "SELECT * FROM batches WHERE batch_date=? AND deleted=0" args = [batch_date] if segment: sql += " AND segment=?" args.append(segment) sql += " ORDER BY id DESC" out = [] for row in conn.execute(sql, args).fetchall(): item = _batch_header(row) counts = conn.execute( """SELECT COUNT(*) AS n, SUM(out_of_range) AS oor FROM batch_lines WHERE batch_id=? AND actual IS NOT NULL""", (row["id"],)).fetchone() item["n_lines"] = counts["n"] or 0 item["n_out_of_range"] = counts["oor"] or 0 item["custom_values"] = _load_custom_values(conn, row["id"]) out.append(item) return out # --------------------------------------------------------------------------- # Reporting def _totals_for_window(conn, date_start, date_end, segment): """Per-material captured totals over an inclusive [date_start, date_end] batch-date window (date_start None => from the earliest captured batch), optionally scoped to one segment, plus converted+rounded consumption values (the ordering/usage view — conversion uses each material's CONSUMPTION unit with its administered rounding decimals). Includes every active material (zero rows included so the report shape is stable) plus any inactive material the window's batches actually used. Shared by daily_totals (start == end) and range_totals. Returns {batch_count, total_yards, materials}.""" conds = ["b.deleted=0", "l.actual IS NOT NULL"] args = [] if date_start is not None: conds.append("b.batch_date>=?") args.append(date_start) conds.append("b.batch_date<=?") args.append(date_end) if segment: conds.append("b.segment=?") args.append(segment) sql = ("SELECT l.material_id, SUM(l.actual) AS total, " "COUNT(l.actual) AS n, SUM(l.out_of_range) AS oor " "FROM batch_lines l JOIN batches b ON b.id = l.batch_id " "WHERE " + " AND ".join(conds) + " GROUP BY l.material_id") sums = {r["material_id"]: r for r in conn.execute(sql, args).fetchall()} bconds = ["deleted=0"] bargs = [] if date_start is not None: bconds.append("batch_date>=?") bargs.append(date_start) bconds.append("batch_date<=?") bargs.append(date_end) if segment: bconds.append("segment=?") bargs.append(segment) brow = conn.execute( "SELECT COUNT(*) AS n, SUM(yards) AS yd FROM batches WHERE " + " AND ".join(bconds), bargs).fetchone() mats = [_material_row(r) for r in conn.execute( "SELECT * FROM materials ORDER BY sort, name").fetchall()] # First pass: leaf (non-aggregate) rows. An aggregator is report-only and # has no batch lines, so it never contributes captured data here. leaf_rows = {} for mat in mats: if mat["is_aggregate"]: continue mid = mat["id"] if not mat["active"] and mid not in sums: continue srow = sums.get(mid) total = (srow["total"] or 0.0) if srow else 0.0 cons = convert_amount(total, mat["capture_unit"], mat["consumption_unit"]) leaf_rows[mid] = { "material_id": mid, "name": mat["name"], "capture_unit": mat["capture_unit"], "captured_total": total, "consumption_unit": mat["consumption_unit"], "consumption_total": (round(cons, mat["consumption_decimals"]) if cons is not None else None), "consumption_decimals": mat["consumption_decimals"], "n_captured": (srow["n"] or 0) if srow else 0, "n_out_of_range": (srow["oor"] or 0) if srow else 0, "is_aggregate": False, } # Second pass (DVI-1423): aggregator rows = sum of their children's # converted (consumption-unit) totals. Children share this aggregate's # consumption unit (validated at save), so summing their consumption_total # is exact. The aggregate row is DISPLAY-ONLY — the grand-total figures # (batch_count / total_yards) are batch-level and never sum material rows, # so a 'both'-mode aggregate can't double-count. agg_rows = {} hidden_children = set() for mat in mats: if not mat["is_aggregate"] or not mat["active"]: continue # DVI-1440 review #2: an aggregate captured DIRECTLY on some mixture has # its own batch lines (material_id == the aggregate). Fold those direct # captures in alongside the summed constituents. No double-count: a batch # captures EITHER the aggregate directly OR its constituents, never both. srow = sums.get(mat["id"]) own_capt = (srow["total"] or 0.0) if srow else 0.0 own_cons = convert_amount(own_capt, mat["capture_unit"], mat["consumption_unit"]) or 0.0 total = own_cons for cid in mat["aggregate_children"]: child = leaf_rows.get(cid) if child and child["consumption_total"] is not None: total += child["consumption_total"] agg_rows[mat["id"]] = { "material_id": mat["id"], "name": mat["name"], "capture_unit": mat["capture_unit"], # captured_total is the directly-captured amount (None when this # aggregate is only ever summed from its constituents). "captured_total": own_capt if srow else None, "consumption_unit": mat["consumption_unit"], "consumption_total": round(total, mat["consumption_decimals"]), "consumption_decimals": mat["consumption_decimals"], "n_captured": (srow["n"] or 0) if srow else 0, "n_out_of_range": (srow["oor"] or 0) if srow else 0, "is_aggregate": True, "aggregate_children": list(mat["aggregate_children"]), "aggregate_report_mode": mat["aggregate_report_mode"], } if mat["aggregate_report_mode"] == "aggregate_only": hidden_children.update(mat["aggregate_children"]) # Assemble in sort order: an aggregator sits at its own sort position; a # child folded out by an 'aggregate_only' aggregate is dropped from the # itemized list (its capture still exists — it just isn't shown). rows = [] for mat in mats: mid = mat["id"] if mat["is_aggregate"]: if mid in agg_rows: rows.append(agg_rows[mid]) elif mid not in hidden_children and mid in leaf_rows: rows.append(leaf_rows[mid]) return {"batch_count": brow["n"] or 0, "total_yards": brow["yd"] or 0.0, "materials": rows} def daily_totals(batch_date, segment=None): """Per-material captured + converted/rounded consumption totals for one day (see _totals_for_window). Stable-shape single-day grid.""" with _connect() as conn: t = _totals_for_window(conn, batch_date, batch_date, segment) return { "batch_date": batch_date, "segment": segment or "", "batch_count": t["batch_count"], "total_yards": t["total_yards"], "materials": t["materials"], } def range_totals(start_date, end_date, segment=None): """Per-material captured + converted/rounded consumption totals over an inclusive [start_date, end_date] window (start_date None => from the first captured batch) — the date-range ordering view (DVI-1326 P2). When no segment filter is given, also returns a per-plant (segment) breakdown so consumption can be split by production line (Wet Cast today; Pipe/Block appear here automatically once their sources land). Segments with no batches in the window are omitted.""" with _connect() as conn: overall = _totals_for_window(conn, start_date, end_date, segment) segments = [] if not segment: for seg in _read_segments(conn): s = seg["id"] st = _totals_for_window(conn, start_date, end_date, s) if st["batch_count"]: segments.append({ "segment": s, "label": seg["label"], "batch_count": st["batch_count"], "total_yards": st["total_yards"], "materials": st["materials"], }) return { "start_date": start_date or "", "end_date": end_date, "segment": segment or "", "batch_count": overall["batch_count"], "total_yards": overall["total_yards"], "materials": overall["materials"], "segments": segments, } # --------------------------------------------------------------------------- # Live tile-summary reads (DVI-1519 P4 / DVI-1523, D8) — cheap, best-effort # counts for the QC Tile Surface catalog summaries. Never raise for the caller; # each returns a plain value from existing tables. # --------------------------------------------------------------------------- def daily_production_submitted_count(batch_date): """Number of distinct segments with a Daily Production entry captured on ``batch_date`` (source='daily_production').""" with _connect() as conn: row = conn.execute( "SELECT COUNT(DISTINCT segment) AS n FROM batches " "WHERE batch_date=? AND source='daily_production' AND deleted=0", (batch_date,)).fetchone() return int(row["n"] or 0) if row else 0 def last_import(): """The most recent Pipe/PLC import (source='import') as {batch_date, count} over that latest date, or None when nothing imported.""" with _connect() as conn: row = conn.execute( "SELECT MAX(batch_date) AS d FROM batches " "WHERE source='import' AND deleted=0").fetchone() if not row or not row["d"]: return None d = row["d"] n = conn.execute( "SELECT COUNT(*) AS n FROM batches " "WHERE source='import' AND batch_date=? AND deleted=0", (d,)).fetchone()["n"] return {"batch_date": d, "count": int(n or 0)} def last_batch_date(): """The latest captured (non-deleted) batch date across all sources, or None — the data-span anchor for the Reports tile summary.""" with _connect() as conn: row = conn.execute( "SELECT MAX(batch_date) AS d FROM batches WHERE deleted=0").fetchone() return row["d"] if row and row["d"] else None # --------------------------------------------------------------------------- # meta (bookkeeping key/value) def _get_meta(conn, key, default=None): row = conn.execute("SELECT value FROM meta WHERE key=?", (key,)).fetchone() return row["value"] if row else default def _set_meta(conn, key, value): conn.execute( "INSERT INTO meta(key, value) VALUES(?,?) " "ON CONFLICT(key) DO UPDATE SET value=excluded.value", (key, value)) # --------------------------------------------------------------------------- # Production Segments (DVI-1452 P1) # # Segments were hardcoded (SEGMENTS/SEGMENT_LABELS = wet_cast/pipe/block). # They are now an admin-managed registry in the meta table so a new plant is # pure configuration (no code change / deploy). Shape: # [{id, label, active, sort}] (JSON, key "segments") # # Rules (board-approved D2): # - `id` is minted once from a slug of the label and is IMMUTABLE (batches / # mixtures store it); the `label` is editable. # - A segment is DEACTIVATED (hidden from new-capture pickers), never # hard-deleted, while referenced by any batch/mixture (mirrors the materials # rule). Hard delete is allowed only when unreferenced. # - Idempotent seed: init_qc_db seeds wet_cast/pipe/block into the meta key # only when absent, so a live DB gets the exact current segments with zero # behavior change. # The SEGMENTS / SEGMENT_LABELS module constants remain ONLY as the seed source # + a defensive fallback; all runtime code reads the registry via the accessors. # --------------------------------------------------------------------------- _SEGMENTS_META_KEY = "segments" def _slugify_segment(label): return re.sub(r"[^a-z0-9]+", "_", str(label or "").strip().lower()).strip("_")[:40] def _coerce_segments(data): """Normalize a raw segments list -> [{id,label,active,sort}] sorted by sort. Drops entries without a usable id; de-dupes ids (first wins); self-healing.""" out, seen = [], set() if isinstance(data, list): for i, ent in enumerate(data): if not isinstance(ent, dict): continue sid = re.sub(r"[^a-z0-9_]+", "", str(ent.get("id") or "").strip().lower()) if not sid or sid in seen: continue label = str(ent.get("label") or "").strip() or sid try: sort = int(ent.get("sort")) except (TypeError, ValueError): sort = (i + 1) * 10 seen.add(sid) out.append({"id": sid, "label": label, "active": bool(ent.get("active", True)), "sort": sort}) out.sort(key=lambda s: (s["sort"], s["id"])) return out def _seed_segments(): return [{"id": s, "label": SEGMENT_LABELS.get(s, s), "active": True, "sort": (i + 1) * 10} for i, s in enumerate(SEGMENTS)] def _read_segments(conn): """The current segment registry for an open connection (self-healing; falls back to the legacy constants if the meta key is somehow absent).""" raw = _get_meta(conn, _SEGMENTS_META_KEY) try: data = json.loads(raw) if raw else None except (TypeError, ValueError): data = None return _coerce_segments(data if data else _seed_segments()) def _segment_label_conn(conn, segment_id): for s in _read_segments(conn): if s["id"] == segment_id: return s["label"] return segment_id def list_segments(include_inactive=False): """Segment registry [{id,label,active,sort}], sorted. Active-only by default (the set new-capture pickers should offer).""" with _connect() as conn: segs = _read_segments(conn) return segs if include_inactive else [s for s in segs if s["active"]] def segment_ids(include_inactive=False): return [s["id"] for s in list_segments(include_inactive=include_inactive)] def segment_options(include_inactive=False): """[{id,label}] for UI pickers/report params.""" return [{"id": s["id"], "label": s["label"]} for s in list_segments(include_inactive=include_inactive)] def segment_label(segment_id, default=None): for s in list_segments(include_inactive=True): if s["id"] == segment_id: return s["label"] return segment_id if default is None else default def is_segment(segment_id, include_inactive=True): """Whether ``segment_id`` names a known segment. Permissive by default (accepts inactive) so read/report paths for historical data still validate; capture pickers scope to active via list_segments().""" return segment_id in segment_ids(include_inactive=include_inactive) def add_segment(label): """Create a segment from ``label`` (id = slug, minted once). Returns (segment_id, error).""" label = str(label or "").strip() if not label: return None, "Segment name is required." sid = _slugify_segment(label) if not sid: return None, "Segment name must contain letters or numbers." with _write_lock, _connect() as conn: segs = _read_segments(conn) if any(s["id"] == sid for s in segs): return None, "A segment with a matching id already exists." segs.append({"id": sid, "label": label, "active": True, "sort": max([s["sort"] for s in segs], default=0) + 10}) _set_meta(conn, _SEGMENTS_META_KEY, json.dumps(_coerce_segments(segs))) conn.commit() return sid, None def update_segment(segment_id, label=None, active=None): """Rename (label) and/or activate/deactivate a segment. The id is immutable. Returns (segment, error).""" with _write_lock, _connect() as conn: segs = _read_segments(conn) target = next((s for s in segs if s["id"] == segment_id), None) if not target: return None, "Unknown segment." if label is not None: lbl = str(label).strip() if not lbl: return None, "Segment name is required." target["label"] = lbl if active is not None: target["active"] = bool(active) segs = _coerce_segments(segs) _set_meta(conn, _SEGMENTS_META_KEY, json.dumps(segs)) conn.commit() return next((s for s in segs if s["id"] == segment_id), None), None def reorder_segments(order_ids): """Reorder segments to match ``order_ids`` (unlisted ids keep their relative order at the end). Returns the reordered registry.""" with _write_lock, _connect() as conn: segs = _read_segments(conn) by_id = {s["id"]: s for s in segs} ordered = [] for sid in (order_ids or []): if sid in by_id and by_id[sid] not in ordered: ordered.append(by_id[sid]) for s in segs: if s not in ordered: ordered.append(s) for n, s in enumerate(ordered, start=1): s["sort"] = n * 10 segs = _coerce_segments(ordered) _set_meta(conn, _SEGMENTS_META_KEY, json.dumps(segs)) conn.commit() return segs def segment_in_use(segment_id): """True if any batch or mixture references the segment.""" with _connect() as conn: if conn.execute("SELECT 1 FROM batches WHERE segment=? LIMIT 1", (segment_id,)).fetchone(): return True return bool(conn.execute( "SELECT 1 FROM mixtures WHERE segment=? LIMIT 1", (segment_id,)).fetchone()) def delete_segment(segment_id): """Hard-delete a segment only when unreferenced; otherwise the caller should deactivate it (mirrors the materials rule). Returns (ok, error).""" if segment_in_use(segment_id): return False, ("Segment is referenced by batches or mixtures — " "deactivate it instead.") with _write_lock, _connect() as conn: if not any(s["id"] == segment_id for s in _read_segments(conn)): return False, "Unknown segment." segs = [s for s in _read_segments(conn) if s["id"] != segment_id] _set_meta(conn, _SEGMENTS_META_KEY, json.dumps(_coerce_segments(segs))) conn.commit() return True, None # --------------------------------------------------------------------------- # Report options (DVI-1424) — report-content settings for the range/ordering # ("Consumption Totals") sheet, kept in one additive meta key. Additive and # self-healing: a fresh DB, or any option absent from the stored blob, falls # back to the current behavior, so an existing install reports exactly as # before until an admin changes something. # # exclude_material_ids : materials the admin switched OFF for reports — the # per-material "include in reports" toggle (hides a # never-ordered material from the ordering sheet). # Default: none excluded. # show_zero_capture : include zero-capture materials in the range report. # Default False = today's behavior (they're omitted). # # The P6 aggregate display mode (both / aggregate_only) is deliberately NOT # stored here — it lives on the material (aggregate_report_mode) so there's one # source of truth; the Reporting sub-tab surfaces + edits it via set_report_mode. _REPORT_OPTIONS_META_KEY = "report_options" def _default_report_options(): return {"exclude_material_ids": [], "show_zero_capture": False} def _coerce_report_options(data, valid_ids=None): """Normalize a raw report_options blob. Drops excluded ids that aren't ints (or, when valid_ids is given, that no longer name a material) so a deleted material never lingers in the exclude list.""" opts = _default_report_options() if not isinstance(data, dict): return opts ex, seen = [], set() for v in (data.get("exclude_material_ids") or []): try: i = int(v) except (TypeError, ValueError): continue if valid_ids is not None and i not in valid_ids: continue if i not in seen: seen.add(i) ex.append(i) opts["exclude_material_ids"] = ex opts["show_zero_capture"] = bool(data.get("show_zero_capture")) return opts def get_report_options(): """Current report-content options (self-healing — drops excluded ids that no longer name a material).""" with _connect() as conn: raw = _get_meta(conn, _REPORT_OPTIONS_META_KEY) try: data = json.loads(raw) if raw else {} except (TypeError, ValueError): data = {} valid = {r["id"] for r in conn.execute("SELECT id FROM materials").fetchall()} return _coerce_report_options(data, valid) def save_report_options(options): """Persist report-content options; returns the normalized stored options.""" with _connect() as conn: valid = {r["id"] for r in conn.execute("SELECT id FROM materials").fetchall()} clean = _coerce_report_options(options, valid) with _write_lock, _connect() as conn: _set_meta(conn, _REPORT_OPTIONS_META_KEY, json.dumps(clean)) conn.commit() return clean def set_report_mode(material_id, mode): """Update just an aggregate material's report display mode — the Reporting sub-tab (DVI-1424) surfaces the P6 aggregate_report_mode alongside the material editor. Returns (material, error); only aggregates carry a mode.""" mode = str(mode or "both").strip().lower() if mode not in AGGREGATE_REPORT_MODES: return None, ("Report mode must be one of: %s." % ", ".join(AGGREGATE_REPORT_MODES)) now = _now() with _write_lock, _connect() as conn: row = conn.execute("SELECT * FROM materials WHERE id=?", (material_id,)).fetchone() if not row: return None, "Material not found." if not _parse_children(_row_get(row, "aggregate_children", "")): return None, "Only an aggregate material has a report display mode." conn.execute( "UPDATE materials SET aggregate_report_mode=?, updated_at=? " "WHERE id=?", (mode, now, material_id)) conn.commit() return get_material(material_id), None def filter_range_report(rep, options=None): """Return a COPY of a range_totals result with the report-content options applied to its material rows (top-level + per-plant sections). Grand totals (batch_count / total_yards) are batch-level and left untouched. Default options reproduce the pre-DVI-1424 ordering-sheet behavior: drop zero-capture leaves and zero-sum aggregates.""" opts = _coerce_report_options(options if options is not None else {}) exclude = set(opts["exclude_material_ids"]) show_zero = opts["show_zero_capture"] def keep(m): if m.get("material_id") in exclude: return False if show_zero: return True if m.get("is_aggregate"): return bool(m.get("consumption_total")) return bool(m.get("n_captured")) def filt(materials): return [m for m in (materials or []) if keep(m)] out = dict(rep) out["materials"] = filt(rep.get("materials")) out["segments"] = [dict(s, materials=filt(s.get("materials"))) for s in (rep.get("segments") or [])] return out # --------------------------------------------------------------------------- # Batch Ticket field tunables (DVI-1441, DVI-1438 item 3) # # Per-segment control over which OPTIONAL Batch Ticket header fields appear on # the two capture surfaces (the Batch Ticket modal + the Scan Tickets New Batch # form). Stored in one additive meta key {segment: {field: bool}}; default all- # enabled, so a fresh DB — and any field/segment absent from the stored blob — # keeps today's behavior (every field shown) until an admin turns one off. # # Only the fields in BATCH_TICKET_TUNABLE_FIELDS are tunable. Date, Mixture, # Yards and the material grid are always on: yards drives target scaling and # date/mixture key the record. D6-A semantics are display-only — a disabled # field is hidden on the capture forms, but the server still accepts a value for # it, so Pipe import / PLC ingest and existing captured data are untouched. _BATCH_TICKET_FIELDS_META_KEY = "batch_ticket_fields" # (field key, label). The key names the batch column the capture forms bind to. BATCH_TICKET_TUNABLE_FIELDS = ( ("batch_number", "Batch #"), ("rock_moisture", "Rock moisture %"), ("sand_moisture", "Sand moisture %"), ("initials", "Initials"), ("notes", "Notes"), ) _BATCH_TICKET_FIELD_KEYS = tuple(k for k, _ in BATCH_TICKET_TUNABLE_FIELDS) def _coerce_batch_ticket_fields(data): """Normalize a raw batch_ticket_fields blob into a stable {segment: {field: bool}} map covering every SEGMENT x tunable field. A missing segment/field defaults to True (enabled) so the capture forms show it (zero behavior change); unknown segments/fields are dropped.""" if not isinstance(data, dict): data = {} out = {} for seg in segment_ids(include_inactive=True): ent = data.get(seg) ent = ent if isinstance(ent, dict) else {} out[seg] = {k: bool(ent.get(k, True)) for k in _BATCH_TICKET_FIELD_KEYS} return out def get_batch_ticket_fields(): """Current per-segment Batch Ticket field visibility (self-healing — every segment x tunable field present, default enabled).""" with _connect() as conn: raw = _get_meta(conn, _BATCH_TICKET_FIELDS_META_KEY) try: data = json.loads(raw) if raw else {} except (TypeError, ValueError): data = {} return _coerce_batch_ticket_fields(data) def save_batch_ticket_fields(config): """Persist per-segment Batch Ticket field visibility; returns the normalized stored config (a stable all-segment/all-field map).""" clean = _coerce_batch_ticket_fields(config) with _write_lock, _connect() as conn: _set_meta(conn, _BATCH_TICKET_FIELDS_META_KEY, json.dumps(clean)) conn.commit() return clean # --------------------------------------------------------------------------- # Scan Tickets settings (DVI-1460 P3) # # Global (not per-segment) knobs for the QC -> Scan Tickets thumbnail review # grid: the thumbnail tile size, and whether OCR pre-processing auto-runs when # the review grid first loads. One additive self-healing meta key; a fresh DB # reproduces the recommended defaults (thumb 150px, auto-OCR ON per the approved # plan) so existing behavior is unchanged until an admin edits it. # --------------------------------------------------------------------------- _SCAN_SETTINGS_META_KEY = "scan_settings" _SCAN_THUMB_MIN = 100 _SCAN_THUMB_MAX = 320 _SCAN_THUMB_DEFAULT = 150 def _coerce_scan_settings(data): """Normalize a raw scan_settings blob into {thumb_size:int, auto_ocr:bool}. thumb_size is clamped to [_SCAN_THUMB_MIN, _SCAN_THUMB_MAX]; a bad value falls back to the default. auto_ocr defaults True (recommended default).""" if not isinstance(data, dict): data = {} try: ts = int(round(float(data.get("thumb_size", _SCAN_THUMB_DEFAULT)))) except (TypeError, ValueError): ts = _SCAN_THUMB_DEFAULT ts = max(_SCAN_THUMB_MIN, min(_SCAN_THUMB_MAX, ts)) return {"thumb_size": ts, "auto_ocr": bool(data.get("auto_ocr", True))} def get_scan_settings(): """Current Scan Tickets settings (self-healing; defaults if unset).""" with _connect() as conn: raw = _get_meta(conn, _SCAN_SETTINGS_META_KEY) try: data = json.loads(raw) if raw else {} except (TypeError, ValueError): data = {} return _coerce_scan_settings(data) def save_scan_settings(config): """Persist Scan Tickets settings; returns the normalized stored config.""" clean = _coerce_scan_settings(config) with _write_lock, _connect() as conn: _set_meta(conn, _SCAN_SETTINGS_META_KEY, json.dumps(clean)) conn.commit() return clean # --------------------------------------------------------------------------- # Custom Batch Ticket fields (DVI-1452 P2, DVI-1438 configurable QC) # # An admin-defined, typed field registry so a new Batch Ticket field is pure # configuration (no code / deploy). A field definition lives in one additive # meta key ('custom_fields'); the per-batch VALUE lives in the batch_custom_ # values child table keyed on the field's immutable id — so a label rename # never orphans historical data (the reason for a child table over columns). # # Shape (JSON list under _CUSTOM_FIELDS_META_KEY): # [{id, label, type, choices, required, segments, sort, active}] # id — minted once ('cf'), immutable (batch values store it) # label — editable display name # type — text | number | date | dropdown | checkbox (D5) # choices — [str] (dropdown only; ignored/empty otherwise) # required — soft flag only; surfaced but NEVER blocks a save (D7 / the # QC-wide "flags never block" rule) # segments — [segment_id] the field shows on; EMPTY = all segments (D6) # sort — display order; active — hidden from capture when False # Rules mirror segments/materials: id immutable, a referenced field is # DEACTIVATED not hard-deleted (so its captured values still resolve a label). # --------------------------------------------------------------------------- _CUSTOM_FIELDS_META_KEY = "custom_fields" CUSTOM_FIELD_TYPES = ("text", "number", "date", "dropdown", "checkbox") def _coerce_custom_fields(data): """Normalize a raw custom_fields blob -> stable [{...}] sorted by sort. Drops entries without a usable id/label; de-dupes ids; scopes segments to known ids (an unknown/dropped segment reverts the field to all-segments); self-healing.""" out, seen = [], set() known = set(segment_ids(include_inactive=True)) if isinstance(data, list): for i, ent in enumerate(data): if not isinstance(ent, dict): continue fid = re.sub(r"[^a-z0-9_]+", "", str(ent.get("id") or "").strip().lower()) if not fid or fid in seen: continue label = str(ent.get("label") or "").strip() if not label: continue ftype = str(ent.get("type") or "text").strip().lower() if ftype not in CUSTOM_FIELD_TYPES: ftype = "text" choices = [] if ftype == "dropdown": raw_choices = ent.get("choices") if isinstance(raw_choices, (list, tuple)): for c in raw_choices: c = str(c).strip()[:120] if c and c not in choices: choices.append(c) segs = [] raw_segs = ent.get("segments") if isinstance(raw_segs, (list, tuple)): for s in raw_segs: s = str(s).strip() if s in known and s not in segs: segs.append(s) try: sort = int(ent.get("sort")) except (TypeError, ValueError): sort = (i + 1) * 10 seen.add(fid) out.append({ "id": fid, "label": label[:120], "type": ftype, "choices": choices, "required": bool(ent.get("required", False)), "segments": segs, "sort": sort, "active": bool(ent.get("active", True))}) out.sort(key=lambda f: (f["sort"], f["id"])) return out def _read_custom_fields(conn): raw = _get_meta(conn, _CUSTOM_FIELDS_META_KEY) try: data = json.loads(raw) if raw else [] except (TypeError, ValueError): data = [] return _coerce_custom_fields(data) def list_custom_fields(segment=None, include_inactive=False): """Custom field registry, sorted. Active-only by default (the set capture forms render). ``segment`` filters to fields scoped to that segment (a field with an empty ``segments`` list applies to every segment).""" with _connect() as conn: fields = _read_custom_fields(conn) if not include_inactive: fields = [f for f in fields if f["active"]] if segment is not None: fields = [f for f in fields if not f["segments"] or segment in f["segments"]] return fields def get_custom_field(field_id): for f in list_custom_fields(include_inactive=True): if f["id"] == field_id: return f return None def _mint_custom_field_id(existing_ids): while True: fid = "cf" + uuid.uuid4().hex[:10] if fid not in existing_ids: return fid def add_custom_field(fields): """Create a custom field. ``fields`` = {label, type, choices?, required?, segments?}. The id is minted once and immutable. Returns (field, error).""" label = str((fields or {}).get("label") or "").strip() if not label: return None, "Field name is required." ftype = str((fields or {}).get("type") or "text").strip().lower() if ftype not in CUSTOM_FIELD_TYPES: return None, "Unknown field type." with _write_lock, _connect() as conn: defs = _read_custom_fields(conn) fid = _mint_custom_field_id({f["id"] for f in defs}) ent = {"id": fid, "label": label, "type": ftype, "choices": (fields or {}).get("choices") or [], "required": bool((fields or {}).get("required", False)), "segments": (fields or {}).get("segments") or [], "sort": max([f["sort"] for f in defs], default=0) + 10, "active": True} defs.append(ent) clean = _coerce_custom_fields(defs) _set_meta(conn, _CUSTOM_FIELDS_META_KEY, json.dumps(clean)) conn.commit() return get_custom_field(fid), None def update_custom_field(field_id, fields): """Edit a custom field (label / type / choices / required / segments / active). The id is immutable. Returns (field, error).""" fields = fields or {} with _write_lock, _connect() as conn: defs = _read_custom_fields(conn) target = next((f for f in defs if f["id"] == field_id), None) if not target: return None, "Unknown field." if "label" in fields: label = str(fields.get("label") or "").strip() if not label: return None, "Field name is required." target["label"] = label if "type" in fields: ftype = str(fields.get("type") or "").strip().lower() if ftype not in CUSTOM_FIELD_TYPES: return None, "Unknown field type." target["type"] = ftype if "choices" in fields: target["choices"] = fields.get("choices") or [] if "required" in fields: target["required"] = bool(fields.get("required")) if "segments" in fields: target["segments"] = fields.get("segments") or [] if "active" in fields: target["active"] = bool(fields.get("active")) clean = _coerce_custom_fields(defs) _set_meta(conn, _CUSTOM_FIELDS_META_KEY, json.dumps(clean)) conn.commit() return get_custom_field(field_id), None def reorder_custom_fields(order_ids): """Reorder fields to match ``order_ids`` (unlisted keep their relative order at the end). Returns the reordered registry.""" with _write_lock, _connect() as conn: defs = _read_custom_fields(conn) by_id = {f["id"]: f for f in defs} ordered = [] for fid in (order_ids or []): if fid in by_id and by_id[fid] not in ordered: ordered.append(by_id[fid]) for f in defs: if f not in ordered: ordered.append(f) for n, f in enumerate(ordered, start=1): f["sort"] = n * 10 clean = _coerce_custom_fields(ordered) _set_meta(conn, _CUSTOM_FIELDS_META_KEY, json.dumps(clean)) conn.commit() return clean def custom_field_in_use(field_id): """True if any batch has captured a value for this field.""" with _connect() as conn: return bool(conn.execute( "SELECT 1 FROM batch_custom_values WHERE field_id=? LIMIT 1", (field_id,)).fetchone()) def delete_custom_field(field_id): """Hard-delete a field only when unreferenced; otherwise the caller should deactivate it (a referenced field keeps its captured values, which still resolve a label). Returns (ok, error).""" if custom_field_in_use(field_id): return False, ("Field has captured values — deactivate it instead.") with _write_lock, _connect() as conn: defs = _read_custom_fields(conn) if not any(f["id"] == field_id for f in defs): return False, "Unknown field." defs = [f for f in defs if f["id"] != field_id] _set_meta(conn, _CUSTOM_FIELDS_META_KEY, json.dumps(_coerce_custom_fields(defs))) conn.commit() return True, None def _coerce_custom_value(field, raw): """Coerce a raw submitted value to its stored string form per field type. Never raises / never blocks (required is a soft flag) — a bad value stores as ''. Returns the normalized string.""" if raw is None: return "" ftype = field["type"] if ftype == "checkbox": return "1" if raw not in (False, "", "0", 0, "false", "False", None) else "" if ftype == "number": try: f = float(raw) except (TypeError, ValueError): return "" # Keep integers clean (no trailing .0) but preserve decimals. return str(int(f)) if f == int(f) else repr(f) if ftype == "date": s = str(raw).strip() return s if _DATE_RE.match(s) else "" if ftype == "dropdown": s = str(raw).strip() return s if (not field["choices"] or s in field["choices"]) else "" return str(raw).strip()[:500] def _sanitize_custom_values(conn, raw_values): """Map {field_id: value} -> {field_id: normalized_str} keeping only KNOWN, ACTIVE fields (a value for an unknown/inactive/absent field is dropped).""" if not isinstance(raw_values, dict): return {} by_id = {f["id"]: f for f in _read_custom_fields(conn) if f["active"]} out = {} for fid, val in raw_values.items(): field = by_id.get(str(fid)) if not field: continue out[field["id"]] = _coerce_custom_value(field, val) return out def _write_custom_values(conn, batch_id, values): """Replace a batch's custom values (values already sanitized). A '' value is still stored so an explicit clear round-trips.""" conn.execute("DELETE FROM batch_custom_values WHERE batch_id=?", (batch_id,)) for fid, val in (values or {}).items(): conn.execute( "INSERT INTO batch_custom_values(batch_id, field_id, value) " "VALUES(?,?,?)", (batch_id, fid, val)) def _load_custom_values(conn, batch_id): return {r["field_id"]: r["value"] for r in conn.execute( "SELECT field_id, value FROM batch_custom_values WHERE batch_id=?", (batch_id,)).fetchall()} # --------------------------------------------------------------------------- # OCR Region Templates (DVI-1442, DVI-1438 item 4) # # A template maps QC batch fields to normalized 0-1 rectangles on a scanned # ticket page, so a Production Segment with a consistent form layout can read # each field via a targeted per-rect OCR call instead of whole-page extraction. # Templates are reusable across segments; ocr_template_assignments picks one # template per segment. # # Stored in two additive meta keys (JSON): # ocr_region_templates : [{id, name, fields: {field_key: {x0,y0,x1,y1}}}] # ocr_template_assignments : {segment: template_id} # # field_key is a header field (OCR_TEMPLATE_HEADER_FIELD_KEYS) or a per-material # actual "mat:" (D7-A: header fields + a row per active material). # Rects are normalized 0-1 (the same shape POST /qc/scan/region takes). Self- # healing: unusable rects / unknown field keys are dropped, an assignment naming # a deleted template is dropped, and a "mat:" rect for a since-deleted # material is simply ignored at extract time. A fresh DB (or any segment with no # assigned template) extracts exactly as before (whole-page) until a template is # built and assigned (D8-A). _OCR_TEMPLATES_META_KEY = "ocr_region_templates" _OCR_ASSIGNMENTS_META_KEY = "ocr_template_assignments" # (field key, label). The key names the QC batch field the region fills. These # mirror the header fields the whole-page extractor produces, so template and # fallback extraction land in the same places. OCR_TEMPLATE_HEADER_FIELDS = ( ("batch_date", "Date"), ("batch_number", "Batch #"), ("mixture", "Mixture"), ("yards", "Yards"), ("rock_moisture", "Rock moisture %"), ("sand_moisture", "Sand moisture %"), ("initials", "Initials"), ("notes", "Notes"), ) OCR_TEMPLATE_HEADER_FIELD_KEYS = tuple(k for k, _ in OCR_TEMPLATE_HEADER_FIELDS) _OCR_MAT_FIELD_RE = re.compile(r"^mat:(\d+)$") # DVI-1452 P3: a custom Batch Ticket field (custom_fields registry) becomes an # assignable region key "custom:" alongside the header keys + mat:. # The id charset mirrors the minted custom-field id (_mint_custom_field_id / the # _coerce_custom_fields scrub -> [a-z0-9_]). A rect for a since-deleted/renamed # custom field is simply ignored at extract time (like a deleted material rect). _OCR_CUSTOM_FIELD_RE = re.compile(r"^custom:([a-z0-9_]+)$") def _coerce_rect(raw): """Normalize a raw rect into {x0,y0,x1,y1} clamped to [0,1] and ordered, or None if it isn't a usable rect (non-numeric or too small to OCR).""" if not isinstance(raw, dict): return None try: x0, y0 = float(raw.get("x0")), float(raw.get("y0")) x1, y1 = float(raw.get("x1")), float(raw.get("y1")) except (TypeError, ValueError): return None x0, x1 = sorted((max(0.0, min(1.0, x0)), max(0.0, min(1.0, x1)))) y0, y1 = sorted((max(0.0, min(1.0, y0)), max(0.0, min(1.0, y1)))) if (x1 - x0) < 0.005 or (y1 - y0) < 0.005: return None return {"x0": x0, "y0": y0, "x1": x1, "y1": y1} def _is_valid_ocr_field_key(key): return (key in OCR_TEMPLATE_HEADER_FIELD_KEYS or bool(_OCR_MAT_FIELD_RE.match(key or "")) or bool(_OCR_CUSTOM_FIELD_RE.match(key or ""))) def ocr_field_material_id(key): """Return the int material id for a "mat:" field key, else None.""" m = _OCR_MAT_FIELD_RE.match(key or "") return int(m.group(1)) if m else None def ocr_field_custom_id(key): """Return the custom-field id for a "custom:" field key, else None (DVI-1452 P3).""" m = _OCR_CUSTOM_FIELD_RE.match(key or "") return m.group(1) if m else None def _coerce_ocr_fields(raw): """Keep only valid field keys that carry a usable rect.""" out = {} if isinstance(raw, dict): for k, v in raw.items(): if _is_valid_ocr_field_key(k): rect = _coerce_rect(v) if rect is not None: out[k] = rect return out def _coerce_ocr_template(raw): """Normalize one stored template dict, or None if unusable (no id/name).""" if not isinstance(raw, dict): return None tid = str(raw.get("id") or "").strip() name = str(raw.get("name") or "").strip()[:120] if not tid or not name: return None return {"id": tid, "name": name, "fields": _coerce_ocr_fields(raw.get("fields"))} def _read_ocr_templates(conn): raw = _get_meta(conn, _OCR_TEMPLATES_META_KEY) try: data = json.loads(raw) if raw else [] except (TypeError, ValueError): data = [] out, seen = [], set() if isinstance(data, list): for ent in data: t = _coerce_ocr_template(ent) if t and t["id"] not in seen: seen.add(t["id"]) out.append(t) return out def list_ocr_templates(): """All OCR region templates (self-healing — unusable entries/rects dropped).""" with _connect() as conn: return _read_ocr_templates(conn) def get_ocr_template(template_id): for t in list_ocr_templates(): if t["id"] == template_id: return t return None def _mint_ocr_template_id(existing): ids = {t["id"] for t in existing} while True: tid = "tpl_" + os.urandom(5).hex() if tid not in ids: return tid def save_ocr_template(template): """Create or update a template. A blank/absent id mints a new one; an existing id updates in place. Returns (template, error).""" if not isinstance(template, dict): return None, "Invalid template." name = str(template.get("name") or "").strip()[:120] if not name: return None, "Template name is required." fields = _coerce_ocr_fields(template.get("fields")) with _write_lock, _connect() as conn: templates = _read_ocr_templates(conn) tid = str(template.get("id") or "").strip() if tid and any(t["id"] == tid for t in templates): for t in templates: if t["id"] == tid: t["name"], t["fields"] = name, fields break else: tid = tid or _mint_ocr_template_id(templates) templates.append({"id": tid, "name": name, "fields": fields}) _set_meta(conn, _OCR_TEMPLATES_META_KEY, json.dumps(templates)) conn.commit() result = next((t for t in templates if t["id"] == tid), None) return result, None def delete_ocr_template(template_id): """Delete a template and drop any segment assignment naming it.""" with _write_lock, _connect() as conn: templates = [t for t in _read_ocr_templates(conn) if t["id"] != template_id] _set_meta(conn, _OCR_TEMPLATES_META_KEY, json.dumps(templates)) valid = {t["id"] for t in templates} assigns = _read_ocr_assignments(conn, valid) _set_meta(conn, _OCR_ASSIGNMENTS_META_KEY, json.dumps(assigns)) conn.commit() return True def _read_ocr_assignments(conn, valid_ids=None): raw = _get_meta(conn, _OCR_ASSIGNMENTS_META_KEY) try: data = json.loads(raw) if raw else {} except (TypeError, ValueError): data = {} if valid_ids is None: valid_ids = {t["id"] for t in _read_ocr_templates(conn)} out = {} if isinstance(data, dict): for seg in [s["id"] for s in _read_segments(conn)]: tid = data.get(seg) if tid and tid in valid_ids: out[seg] = tid return out def get_ocr_template_assignments(): """Per-segment template assignments (self-healing — unknown segments and assignments naming a missing template are dropped).""" with _connect() as conn: return _read_ocr_assignments(conn) def save_ocr_template_assignments(assignments): """Persist per-segment template assignments; returns the normalized map.""" with _write_lock, _connect() as conn: valid = {t["id"] for t in _read_ocr_templates(conn)} clean = {} if isinstance(assignments, dict): for seg in [s["id"] for s in _read_segments(conn)]: tid = str(assignments.get(seg) or "").strip() if tid and tid in valid: clean[seg] = tid _set_meta(conn, _OCR_ASSIGNMENTS_META_KEY, json.dumps(clean)) conn.commit() return clean def get_assigned_template(segment): """The OCR template assigned to a segment, or None.""" with _connect() as conn: tid = _read_ocr_assignments(conn).get(segment) if not tid: return None for t in _read_ocr_templates(conn): if t["id"] == tid: return t return None # --------------------------------------------------------------------------- # PLC / Mix Log import (DVI-1326 P3) # # The Pipe plant exports a daily "Mix Log" spreadsheet from its PLC — one row # per batch, with per-material dispensed quantities plus moisture / recipe / # mix-time columns. Today those numbers are keyed into the master sheet's Pipe # columns by hand; this imports them straight into batches/batch_lines with # source='import' so they flow through the same capture/report machinery as # manual Wet Cast tickets and stay editable in the existing batch-edit UI. # # The parser (parse_mix_log) is deliberately format-generic — it normalizes any # header/row grid — while the Pipe-specific semantics (which column is a # material vs. moisture vs. recipe, and how each column maps onto a Material) # live in the import layer. That keeps parse_mix_log reusable for the future # Wet Cast SCADA/PLC feed (P5 POST /qc/ingest), which will hand it an already- # parsed row grid instead of an .xlsx. PIPE_SEGMENT = "pipe" # meta key holding the admin's column->material overrides for the Pipe import. _PIPE_MAP_META_KEY = "pipe_material_map" _PIPE_IGNORE = "__ignore__" # DVI-1421 P3: a column may be mapped to the per-batch yardage instead of a # material (covers header drift when the auto-detect below misses the yards # column). Stored in pipe_material_map alongside material ids / __ignore__. _PIPE_YARDS = "__yards__" # Default column->material-NAME map for the known Pipe Mix Log columns, matched # on a normalized header fragment. Resolved to a material id at import time so a # rename/reseed can't leave a stale id behind. Columns not covered here (e.g. # HyxD_TotalWater — waters are split 80/20 in the recipe, so the single total # has no unambiguous target) are surfaced to the admin as UNMAPPED rather than # silently dropped; the admin maps or ignores them in the Import panel. _PIPE_DEFAULT_MATERIAL_BY_NORM = { "mixlogflyash": "Fly Ash", "mixlogcement": "Cement", "mixlogsand": "Sand", "mixlogrock": "Rock", } def _pipe_norm(header): return re.sub(r"[^a-z0-9]", "", str(header or "").lower()) def _pipe_column_role(header): """Classify a Mix Log column by its (normalized) header. Everything that isn't an identified date/time/moisture/recipe/meta column is treated as a material candidate — so an unrecognized quantity column surfaces to the admin instead of vanishing.""" n = _pipe_norm(header) if n == "date": return "date" if n == "time": return "time" if "sandmoisture" in n or ("sand" in n and "moisture" in n): return "sand_moisture" if "rockmoisture" in n or ("rock" in n and "moisture" in n): return "rock_moisture" if "recipe" in n: return "recipe" # DVI-1421 P3: per-batch yardage column (e.g. "Yards"/"yd3"/"MixLog_Yds"). # Guarded against the moisture/mixtime meta check below so it wins. if "yard" in n or "yds" in n: return "yards" if "moisture" in n or "mixtime" in n: return "meta" return "material" def _coerce_import_date(value): """A Mix Log Date cell -> 'YYYY-MM-DD' (None when unparseable).""" if isinstance(value, datetime): return value.strftime("%Y-%m-%d") if isinstance(value, _date_cls): return value.strftime("%Y-%m-%d") s = str(value or "").strip() if not s: return None if _DATE_RE.match(s): return s for fmt in ("%m/%d/%Y", "%m/%d/%y", "%Y/%m/%d", "%m.%d.%Y", "%m.%d.%y", "%m-%d-%Y", "%m-%d-%y"): try: return datetime.strptime(s, fmt).strftime("%Y-%m-%d") except ValueError: continue return None def _coerce_import_num(value): if value in (None, ""): return None try: return float(value) except (TypeError, ValueError): return None def parse_mix_log(source): """Parse a PLC Mix Log .xlsx into a normalized row grid. `source` is a path or a file-like object (an uploaded file's stream). Reusable for any single- sheet header+rows export; Pipe-specific meaning is applied by the caller. Returns {ok, error, sheet, columns:[stripped headers in order], rows:[{header: cell_value}]}. Rows that are entirely blank are dropped.""" try: import openpyxl # lazy: keep qc_store importable without openpyxl except Exception as exc: # pragma: no cover - env without openpyxl return {"ok": False, "error": "openpyxl is required to read the " "spreadsheet (%s)." % exc, "sheet": "", "columns": [], "rows": []} try: wb = openpyxl.load_workbook(source, data_only=True, read_only=True) except Exception as exc: return {"ok": False, "error": "Could not open the spreadsheet: %s" % exc, "sheet": "", "columns": [], "rows": []} try: ws = wb.worksheets[0] if wb.worksheets else None if ws is None: return {"ok": False, "error": "The workbook has no worksheets.", "sheet": "", "columns": [], "rows": []} headers, seen, rows = [], {}, [] it = ws.iter_rows(values_only=True) try: header_row = next(it) except StopIteration: header_row = () for idx, raw in enumerate(header_row): name = str(raw).strip() if raw is not None else "" if not name: name = "col%d" % (idx + 1) # de-duplicate collided headers so the row dict never loses a column if name in seen: seen[name] += 1 name = "%s_%d" % (name, seen[name]) else: seen[name] = 0 headers.append(name) for raw in it: if raw is None: continue values = list(raw) if all(v is None or (isinstance(v, str) and not v.strip()) for v in values): continue row = {} for i, header in enumerate(headers): row[header] = values[i] if i < len(values) else None rows.append(row) return {"ok": True, "error": None, "sheet": ws.title, "columns": headers, "rows": rows} finally: wb.close() def get_pipe_material_map(): """Admin column->material overrides for the Pipe import: {header: material_id | '__ignore__'}.""" with _connect() as conn: raw = _get_meta(conn, _PIPE_MAP_META_KEY) if not raw: return {} try: data = json.loads(raw) except (TypeError, ValueError): return {} out = {} if isinstance(data, dict): for k, v in data.items(): if v == _PIPE_IGNORE: out[str(k)] = _PIPE_IGNORE elif v == _PIPE_YARDS: out[str(k)] = _PIPE_YARDS else: try: out[str(k)] = int(v) except (TypeError, ValueError): continue return out def save_pipe_material_map(mapping): """Persist the admin column->material overrides. Values are a material id, the string '__ignore__', or falsy (which clears that column back to the default/auto behavior). Returns the normalized stored map.""" clean = {} if isinstance(mapping, dict): for k, v in mapping.items(): key = str(k or "").strip() if not key: continue if v in (None, "", "auto", "default"): continue # cleared -> fall back to default/auto resolution if v == _PIPE_IGNORE: clean[key] = _PIPE_IGNORE continue if v == _PIPE_YARDS: clean[key] = _PIPE_YARDS continue try: clean[key] = int(v) except (TypeError, ValueError): continue with _write_lock, _connect() as conn: _set_meta(conn, _PIPE_MAP_META_KEY, json.dumps(clean)) conn.commit() return clean def _resolve_pipe_columns(conn, material_cols, admin_map): """Resolve each material-candidate column to a material (or ignored / unmapped). Returns {header: {status, material_id?, material_name?}} where status in mapped_admin | mapped_default | ignored | unmapped.""" mats = [_material_row(r) for r in conn.execute( "SELECT * FROM materials").fetchall()] by_id = {m["id"]: m for m in mats} by_name = {m["name"].lower(): m for m in mats} out = {} for col in material_cols: if col in admin_map: val = admin_map[col] if val == _PIPE_IGNORE: out[col] = {"status": "ignored"} continue mat = by_id.get(val) if mat: out[col] = {"status": "mapped_admin", "material_id": mat["id"], "material_name": mat["name"]} else: out[col] = {"status": "unmapped"} # mapped to a deleted material continue name = _PIPE_DEFAULT_MATERIAL_BY_NORM.get(_pipe_norm(col)) mat = by_name.get(name.lower()) if name else None if mat: out[col] = {"status": "mapped_default", "material_id": mat["id"], "material_name": mat["name"]} else: out[col] = {"status": "unmapped"} return out def _first_col(columns, roles, want): for c in columns: if roles.get(c) == want: return c return None def import_pipe_mix_log(source, actor="", dry_run=False): """Import a daily Pipe Mix Log .xlsx: one row -> one `pipe` batch with source='import'. Idempotent per (batch_date, import_key=batch time), so a same-day re-upload UPDATES matching batches (re-snapshotting from the file) instead of duplicating them. Material columns are mapped onto the administrable Materials list; unmapped columns are surfaced in the report, never silently dropped. dry_run analyzes + reports without writing. Returns a report dict: {ok, error?, batch_date, total_rows, imported, updated, skipped_no_date, material_columns, mapped[], unmapped[], ignored[], line_count}.""" parsed = parse_mix_log(source) if not parsed["ok"]: return {"ok": False, "error": parsed["error"]} columns = parsed["columns"] rows = parsed["rows"] roles = {c: _pipe_column_role(c) for c in columns} admin_map = get_pipe_material_map() # DVI-1421 P3: resolve the per-batch yardage column. An explicit __yards__ # admin mapping wins (covers header drift); otherwise auto-detect a # yard/yds header. The yards column is never treated as a material. yards_col = next((c for c in columns if admin_map.get(c) == _PIPE_YARDS), None) if yards_col is None: yards_col = _first_col(columns, roles, "yards") material_cols = [c for c in columns if roles[c] == "material" and c != yards_col and admin_map.get(c) != _PIPE_YARDS] date_col = _first_col(columns, roles, "date") time_col = _first_col(columns, roles, "time") sand_col = _first_col(columns, roles, "sand_moisture") rock_col = _first_col(columns, roles, "rock_moisture") recipe_col = _first_col(columns, roles, "recipe") meta_cols = [c for c in columns if roles[c] == "meta"] if date_col is None: return {"ok": False, "error": "No Date column found in the Mix Log — " "expected a 'Date' column."} if not material_cols: return {"ok": False, "error": "No material quantity columns found in " "the Mix Log."} now = _now() report = { "ok": True, "batch_date": "", "total_rows": len(rows), "imported": 0, "updated": 0, "skipped_no_date": 0, "material_columns": material_cols, "mapped": [], "unmapped": [], "ignored": [], "line_count": 0, # DVI-1421 P3 yardage + mix-code surfacing (never silent). "yards_column": yards_col, "yards_blank": 0, "yards_total": 0.0, "mix_code_column": recipe_col, "mixtures_matched": 0, "unmatched_mix_codes": [], "dry_run": bool(dry_run), } dates_seen = [] unmatched_codes = {} with _write_lock, _connect() as conn: resolution = _resolve_pipe_columns(conn, material_cols, admin_map) # DVI-1421 P3: resolve each row's Mix Log recipe code to a pipe mixture # (active mixtures with a non-blank mix code). An unmatched code never # blocks the row — the batch still imports without a mixture. pipe_mix_by_code = {} for mrow in conn.execute( "SELECT id, name, mix_code FROM mixtures " "WHERE segment=? AND active=1", (PIPE_SEGMENT,)).fetchall(): code = _norm_mix_code(mrow["mix_code"]) if code: pipe_mix_by_code.setdefault(code, mrow) # Build the mapping report (samples help the admin recognize a column). sample = {} for r in rows: for c in material_cols: if c not in sample and _coerce_import_num(r.get(c)) is not None: sample[c] = _coerce_import_num(r.get(c)) for col in material_cols: res = resolution[col] if res["status"] in ("mapped_admin", "mapped_default"): report["mapped"].append({ "column": col, "material_id": res["material_id"], "material_name": res["material_name"], "source": res["status"]}) elif res["status"] == "ignored": report["ignored"].append({"column": col}) else: report["unmapped"].append( {"column": col, "sample": sample.get(col)}) for idx, r in enumerate(rows): bdate = _coerce_import_date(r.get(date_col)) if not bdate: report["skipped_no_date"] += 1 continue if bdate not in dates_seen: dates_seen.append(bdate) time_val = str(r.get(time_col) or "").strip() if time_col else "" import_key = time_val or ("row%d" % (idx + 1)) # DVI-1421 P3: per-batch yardage. Blank/unparseable -> 0, counted # in the report (never silent). Fixes the Consumption Totals bug — # SUM(yards) was 0 because yards was hardcoded 0 on every row. yards = 0.0 if yards_col is not None: yval = _coerce_import_num(r.get(yards_col)) if yval is None or yval < 0: report["yards_blank"] += 1 else: yards = yval report["yards_total"] += yards # DVI-1421 P3: resolve the recipe code to a pipe mixture. Unmatched # codes are surfaced per-code; the batch still imports mixture-less. mixture_id = None recipe_raw = r.get(recipe_col) if recipe_col is not None else None code_norm = _norm_mix_code(recipe_raw) if code_norm: m = pipe_mix_by_code.get(code_norm) if m: mixture_id = m["id"] report["mixtures_matched"] += 1 else: label = str(recipe_raw).strip() unmatched_codes[label] = unmatched_codes.get(label, 0) + 1 # Header fields clean = { "batch_date": bdate, "segment": PIPE_SEGMENT, "batch_number": time_val or ("#%d" % (idx + 1)), "yards": yards, "rock_moisture": _coerce_import_num(r.get(rock_col)) if rock_col else None, "sand_moisture": _coerce_import_num(r.get(sand_col)) if sand_col else None, "initials": "", "notes": "", "source": "import", "mixture_id": mixture_id, } # Notes preserve the non-material metadata (recipe / mix time / # final moisture) so nothing from the sheet is lost. note_bits = [] if recipe_col is not None and r.get(recipe_col) not in (None, ""): note_bits.append("Recipe %s" % r.get(recipe_col)) for c in meta_cols: v = r.get(c) if v not in (None, ""): note_bits.append("%s: %s" % (c, v)) clean["notes"] = "; ".join(note_bits)[:2000] # Material lines from the mapped columns. lines = [] for col in material_cols: res = resolution[col] if res["status"] not in ("mapped_admin", "mapped_default"): continue val = _coerce_import_num(r.get(col)) if val is None: continue lines.append({"material_id": res["material_id"], "actual": val}) if dry_run: existing = conn.execute( "SELECT id FROM batches WHERE batch_date=? AND segment=? " "AND source='import' AND import_key=? AND deleted=0", (bdate, PIPE_SEGMENT, import_key)).fetchone() if existing: report["updated"] += 1 else: report["imported"] += 1 report["line_count"] += len(lines) continue action, err = _upsert_import_batch( conn, clean, lines, import_key, actor, now) if err: return {"ok": False, "error": err} report[action] += 1 report["line_count"] += len(lines) if not dry_run: conn.commit() report["batch_date"] = dates_seen[0] if dates_seen else "" report["dates"] = dates_seen report["yards_total"] = round(report["yards_total"], 4) report["unmatched_mix_codes"] = [ {"code": code, "count": count} for code, count in sorted(unmatched_codes.items())] return report def _upsert_import_batch(conn, clean, lines, import_key, actor, now): """Insert or (idempotently) update one keyed batch on (batch_date, segment, source, import_key) — the source comes from ``clean['source']`` so both the Pipe import ('import') and Daily Production ('daily_production', DVI-1410 P3) share this upsert. Returns ('imported'|'updated', None) or (None, error).""" line_rows, mix_name, mix_version, err = _build_lines(conn, clean, lines) if err: return None, err existing = conn.execute( "SELECT id FROM batches WHERE batch_date=? AND segment=? " "AND source=? AND import_key=? AND deleted=0", (clean["batch_date"], clean["segment"], clean["source"], import_key)).fetchone() if existing: batch_id = existing["id"] conn.execute( """UPDATE batches SET batch_number=?, mixture_id=?, mixture_version=?, mixture_name=?, yards=?, rock_moisture=?, sand_moisture=?, initials=?, notes=?, updated_by=?, updated_at=? WHERE id=?""", (clean["batch_number"], clean["mixture_id"], mix_version, mix_name, clean["yards"], clean["rock_moisture"], clean["sand_moisture"], clean["initials"], clean["notes"], actor, now, batch_id)) conn.execute("DELETE FROM batch_lines WHERE batch_id=?", (batch_id,)) action = "updated" else: cur = conn.execute( """INSERT INTO batches(batch_date, segment, batch_number, mixture_id, mixture_version, mixture_name, yards, rock_moisture, sand_moisture, initials, notes, source, import_key, created_by, created_at, updated_by, updated_at, deleted) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,0)""", (clean["batch_date"], clean["segment"], clean["batch_number"], clean["mixture_id"], mix_version, mix_name, clean["yards"], clean["rock_moisture"], clean["sand_moisture"], clean["initials"], clean["notes"], clean["source"], import_key, actor, now, actor, now)) batch_id = cur.lastrowid action = "imported" for ln in line_rows: conn.execute( """INSERT INTO batch_lines(batch_id, material_id, material_name, unit, target, min_ok, max_ok, actual, out_of_range) VALUES(?,?,?,?,?,?,?,?,?)""", (batch_id, ln["material_id"], ln["material_name"], ln["unit"], ln["target"], ln["min_ok"], ln["max_ok"], ln["actual"], ln["out_of_range"])) return action, None # --------------------------------------------------------------------------- # Daily Production (DVI-1410 P3) # # A generalized end-of-day capture: an admin assigns a Mixture to a Production # Segment (Settings); an operator keys in Total Yards + one value per ingredient # in that mixture's recipe, once per day. Stored as ONE source='daily_production' # batch per (segment, date) — keyed by import_key so a re-save corrects it in # place — so it flows into the existing reporting with no new reporting code. _DAILY_PROD_META_KEY = "daily_production_config" def _daily_key(segment, batch_date): return "daily:%s:%s" % (segment, batch_date) def get_daily_production_config(): """{segment: {enabled: bool, mixture_id: int|None}} — which segments capture Daily Production and each one's assigned Mixture. Self-heals: drops unknown segments and mixture ids that no longer exist / don't belong to the segment.""" with _connect() as conn: raw = _get_meta(conn, _DAILY_PROD_META_KEY) try: data = json.loads(raw) if raw else {} except (TypeError, ValueError): data = {} valid_mix = {r["id"]: r["segment"] for r in conn.execute("SELECT id, segment FROM mixtures").fetchall()} valid_segs = {s["id"] for s in _read_segments(conn)} out = {} if isinstance(data, dict): for seg, ent in data.items(): if seg not in valid_segs or not isinstance(ent, dict): continue mid = ent.get("mixture_id") try: mid = int(mid) if mid not in (None, "") else None except (TypeError, ValueError): mid = None # A mixture must exist and belong to this segment to stay assigned. if mid is not None and valid_mix.get(mid) != seg: mid = None out[seg] = {"enabled": bool(ent.get("enabled")), "mixture_id": mid} return out def save_daily_production_config(config): """Persist the per-segment Daily Production config. A segment is stored only when it has a real assignment (enabled and/or a valid same-segment mixture); an all-default segment is dropped. Returns the normalized stored config.""" clean = {} if isinstance(config, dict): with _connect() as conn: valid_mix = {r["id"]: r["segment"] for r in conn.execute("SELECT id, segment FROM mixtures").fetchall()} valid_segs = {s["id"] for s in _read_segments(conn)} for seg, ent in config.items(): if seg not in valid_segs or not isinstance(ent, dict): continue mid = ent.get("mixture_id") try: mid = int(mid) if mid not in (None, "") else None except (TypeError, ValueError): mid = None if mid is not None and valid_mix.get(mid) != seg: mid = None enabled = bool(ent.get("enabled")) if not enabled and mid is None: continue clean[seg] = {"enabled": enabled, "mixture_id": mid} with _write_lock, _connect() as conn: _set_meta(conn, _DAILY_PROD_META_KEY, json.dumps(clean)) conn.commit() return clean def daily_production_segments(): """Segments enabled for Daily Production with an assigned mixture, for the tab's segment picker: [{segment, label, mixture_id, mixture_name}].""" cfg = get_daily_production_config() out = [] with _connect() as conn: for sdef in [s for s in _read_segments(conn) if s["active"]]: seg = sdef["id"] ent = cfg.get(seg) or {} if not (ent.get("enabled") and ent.get("mixture_id")): continue mrow = conn.execute("SELECT name FROM mixtures WHERE id=?", (ent["mixture_id"],)).fetchone() out.append({"segment": seg, "label": sdef["label"], "mixture_id": ent["mixture_id"], "mixture_name": mrow["name"] if mrow else ""}) return out def _daily_ingredient_rows(included, direct_aggs, targets, mats): """DVI-1439 P1 (D1-A) / DVI-1440 P2 row source for the Daily Production form. ``included`` is the mixture's per-material membership (DVI-1440): an explicit list of leaf material ids, or None when unset (legacy). Precedence: * membership set -> exactly those leaf materials (in sort order), each with its per-yd3 target/-/+ when it has one, else target-less; * no membership + targets -> one row per target (DVI-1439 behavior); * no membership + no targets -> every ACTIVE leaf material target-less so the day can still be captured (DVI-1439 capture-without-targets). ``direct_aggs`` (DVI-1440 review #2) is the mixture's list of aggregate ids captured DIRECTLY: each becomes a capture line (in place of its constituent materials, which are suppressed). An aggregate not in ``direct_aggs`` stays report-only (summed from constituents) and never a capture line — matching ``_build_lines``. Target/range cells show '—' for a target-less row and nothing flags out of range.""" tmap = {t["material_id"]: t for t in (targets or [])} direct, suppressed = _capture_plan(direct_aggs, mats) def _mk(mid): m = mats.get(mid) if not m: return None if m["is_aggregate"]: if mid not in direct: return None elif mid in suppressed: return None t = tmap.get(mid) return {"material_id": mid, "name": m["name"], "unit": m["capture_unit"], "target": t["target"] if t else None, "minus": t["minus"] if t else None, "plus": t["plus"] if t else None} rows = [] if included is not None: incl = set(included) # A membership material shows if it's a captured leaf (or a direct- # capture aggregate) and is active (or still carries a target — so a # target on a now-inactive material isn't silently dropped from capture, # matching the legacy targets branch). wanted = [m for m in mats.values() if m["id"] in incl and (m["active"] or m["id"] in tmap)] wanted.sort(key=lambda m: (m["sort"], (m["name"] or "").lower())) for m in wanted: r = _mk(m["id"]) if r: rows.append(r) return rows if targets: for t in targets: r = _mk(t["material_id"]) if r: rows.append(r) return rows leaves = [m for m in mats.values() if m["active"] and (not m["is_aggregate"] or m["id"] in direct)] leaves.sort(key=lambda m: (m["sort"], (m["name"] or "").lower())) for m in leaves: r = _mk(m["id"]) if r: rows.append(r) return rows def daily_production_view(segment, batch_date): """Form context for one (segment, date): the segment's Daily Production config, the ingredient rows to capture, and any already-captured entry for that day (so re-opening edits it in place). ``ingredients`` drives the form — one input per ingredient + Total Yards. ``mixture`` is None only when no mixture is assigned to the segment (the SPA then shows the Settings admin-pointer). ``has_targets`` is False when a mixture IS assigned but has no per-yd3 targets yet — the form still captures (DVI-1439 P1 item 2), and the SPA shows a Mixtures-tab NOTICE, not a blocker. Empty ``ingredients`` means there are no active leaf materials.""" cfg = get_daily_production_config().get(segment) or {} out = {"segment": segment, "batch_date": batch_date, "enabled": bool(cfg.get("enabled")), "mixture": None, "has_targets": False, "ingredients": [], "entry": None} mid = cfg.get("mixture_id") entry_id = None with _connect() as conn: if cfg.get("enabled") and mid: mrow = conn.execute("SELECT * FROM mixtures WHERE id=?", (mid,)).fetchone() if mrow: vid = _version_row_id(conn, mid, mrow["current_version"]) targets = _targets_for_version(conn, vid) if vid else [] mats = {r["id"]: _material_row(r) for r in conn.execute("SELECT * FROM materials").fetchall()} included = _parse_included( _row_get(mrow, "included_materials", "")) direct_aggs = _parse_id_list( _row_get(mrow, "aggregate_capture", "")) out["mixture"] = {"id": mrow["id"], "name": mrow["name"], "version": mrow["current_version"]} out["has_targets"] = bool(targets) out["ingredients"] = _daily_ingredient_rows( included, direct_aggs, targets, mats) row = conn.execute( "SELECT id FROM batches WHERE batch_date=? AND segment=? " "AND source='daily_production' AND import_key=? AND deleted=0", (batch_date, segment, _daily_key(segment, batch_date))).fetchone() if row: entry_id = row["id"] if entry_id: out["entry"] = get_batch(entry_id) return out def upsert_daily_production(segment, batch_date, fields, actor=""): """Capture (or correct in place) the day's Daily Production entry for a segment. One entry per (segment, date), stored as a source='daily_production' batch keyed import_key='daily::' with batch_lines snapshotting the assigned mixture's recipe scaled by the entered yards. Out-of-range is flagged, never blocks. Returns (view, error).""" if not is_segment(segment): return None, "Unknown segment." if not _DATE_RE.match(str(batch_date or "")): return None, "date must be YYYY-MM-DD." cfg = get_daily_production_config().get(segment) or {} if not cfg.get("enabled") or not cfg.get("mixture_id"): return None, "Daily Production is not enabled for this segment." fields = dict(fields or {}) fields["batch_date"] = batch_date fields["segment"] = segment fields["source"] = "daily_production" fields["mixture_id"] = cfg["mixture_id"] clean, err = _sanitize_batch_fields(fields) if err: return None, err now = _now() key = _daily_key(segment, batch_date) with _write_lock, _connect() as conn: action, err = _upsert_import_batch(conn, clean, fields.get("lines"), key, actor, now) if err: return None, err conn.commit() return daily_production_view(segment, batch_date), None