""" scada_report.py — SCADA daily-log report engine for Togen (DVI-482 / DVI-486). Self-contained extraction of the SCADA reporting engine that ST1 (DVI-482) first authored inside app.py. It is split out here so both the Flask app and the standalone scheduler (``scada_report_scheduler.py``, DVI-486) can build reports from a single source of truth without importing app.py — app.py spins up background loops at import time and is unsuitable for a one-shot cron/timer job (mirrors how ``notification_scheduler.py`` imports ``notifications.py``, never app.py). The engine reads daily X-400 "Log" emails from the SCADA mailbox via Microsoft Graph, parses their multi-row tables (columns derived per-email from the header row), bins rows by their parsed Date Time, and renders an XLSX workbook with one sheet per day. It is Flask-independent and configured purely from env vars, so it can be imported and unit-tested in isolation. Required env vars (same as app.py / notifications.py): AZURE_CLIENT_ID, AZURE_CLIENT_SECRET, AZURE_TENANT_ID Optional env var: SCADA_MAILBOX (default scadalogs@icastinc.com) """ import html import io import json import os import re from datetime import datetime, timedelta from pathlib import Path from urllib.parse import quote import msal import requests as _http_requests # Azure / Graph config (env-derived, mirrors app.py). ``.get`` with a default # keeps the module importable when creds are absent (e.g. tests); the Graph # fetch then simply returns an error string instead of raising at import. _AZ_CLIENT_ID = os.environ.get("AZURE_CLIENT_ID", "") _AZ_SECRET = os.environ.get("AZURE_CLIENT_SECRET", "") _AZ_TENANT = os.environ.get("AZURE_TENANT_ID", "") _AZ_AUTHORITY_HOME = f"https://login.microsoftonline.com/{_AZ_TENANT}" SCADA_MAILBOX = os.environ.get("SCADA_MAILBOX", "scadalogs@icastinc.com") # --------------------------------------------------------------------------- # Daily-log emails (subject "X-400 - Log") carry a full-day, 15-min-interval # table in the body. The sensor column set is site-specific (e.g. "K1 Temp(F)" # vs "Probe 1 Temperature(F)"), so columns are always derived per-email from the # header row and never hardcoded. bodyPreview truncates the table, so we always # use the full body.content (HTML or text). # Matches the leading "MM/DD/YYYY HH:MM:SS" of a data row's Date Time cell. _SCADA_LOG_DT_RE = re.compile(r"^(\d{1,2})/(\d{1,2})/(\d{4})\s+(\d{1,2}):(\d{2}):(\d{2})") # The mailbox carries two subject types we care about (DVI-538): # "X-400 - Log" -> daily 24h table, one row per 15-min interval # "X-400 - Trigger: Logs..." -> point-in-time status snapshot # Trigger subjects end in varied text and commonly contain the word "Logs" # (e.g. "X-400 - Trigger: Logs to Scadalogs"), so a naive substring match on # "log" misclassifies them as log emails. Both are matched by prefix, and # Trigger is tested first so the "Logs" inside a Trigger subject never wins. _SCADA_SUBJECT_TRIGGER_RE = re.compile(r"^\s*x-400\s*-\s*trigger\b", re.I) _SCADA_SUBJECT_LOG_RE = re.compile(r"^\s*x-400\s*-\s*log\b", re.I) def scada_subject_kind(subject): """Classify a SCADA email by subject line: 'log', 'trigger', or None. Routing both parsers off the subject keeps daily-log tables out of the Trigger-format status parser and vice-versa. """ s = subject or "" if _SCADA_SUBJECT_TRIGGER_RE.match(s): return "trigger" if _SCADA_SUBJECT_LOG_RE.match(s): return "log" return None def _scada_log_body_text(msg): """Return a daily-log email body as tab-delimited plain text. Converts HTML table markup to tabs/newlines so the same tab-delimited parser handles both HTML and text emails. bodyPreview is intentionally ignored — it truncates the multi-row table. """ body_obj = msg.get("body", {}) or {} content = body_obj.get("content", "") or "" is_html = (body_obj.get("contentType", "").lower() == "html" or re.search(r"<(table|td|tr)\b", content, re.I) is not None) if is_html: content = re.sub(r"", "\t", content, flags=re.I) content = re.sub(r"", "\n", content, flags=re.I) content = re.sub(r"", "\n", content, flags=re.I) content = re.sub(r"<[^>]+>", "", content) content = html.unescape(content) return content def _scada_log_delimiter(body_text): """Pick the cell delimiter for a log body: tab (HTML-converted) or comma. Real X-400 plain-text log emails are CSV ("Date Time,Vin(V),K1 Temp(F),...") — the original tab-only split left every line as one cell, which is what collapsed the XLSX report into a single column (DVI-1095). Tab wins when present so HTML-table bodies (converted to tabs by _scada_log_body_text) keep working even though their values never contain tabs. """ return "\t" if "\t" in body_text else "," def _parse_scada_log_email(body_text): """Parse a daily X-400 log email body into a delimited table (tab or CSV). Returns {"columns": [...], "rows": [...]}. Columns come verbatim from the email's header row ("Date Time | | Trigger"). Each row is a dict {col: value, ..., "_dt": datetime|None}. Missing readings ("x.x") are kept verbatim. """ delim = _scada_log_delimiter(body_text) lines = body_text.splitlines() columns = [] for line in lines: cells = [c.strip() for c in line.split(delim)] if cells and cells[0].lower() == "date time": columns = [c for c in cells if c] break rows = [] for line in lines: cells = line.split(delim) m = _SCADA_LOG_DT_RE.match(cells[0].strip()) if not m: continue values = [c.strip() for c in cells] cols = list(columns) while len(cols) < len(values): cols.append(f"Column{len(cols) + 1}") row = {cols[k]: values[k] for k in range(len(values))} try: mo, da, yr, hh, mm, ss = (int(g) for g in m.groups()) row["_dt"] = datetime(yr, mo, da, hh, mm, ss) except (ValueError, TypeError): row["_dt"] = None rows.append(row) if not columns and rows: columns = [k for k in rows[0].keys() if k != "_dt"] return {"columns": columns, "rows": rows} def _scada_norm_sensor(name): """Strip the unit suffix from a column name: 'K1 Temp(F)' -> 'K1 Temp'.""" return re.sub(r"\s*\([^)]*\)\s*$", "", name).strip() def _graph_error_message(resp): """Short human-readable message from a Graph error response body. Surfaced so callers can distinguish e.g. a tenant ApplicationAccessPolicy (RAOP) block from a missing Mail.Read grant (DVI-1095 P0 — the two need different admins to fix). """ try: msg = (resp.json().get("error") or {}).get("message", "") except Exception: msg = "" return msg or resp.text[:300] def _scada_fetch_log_messages(received_ge=None, received_le=None, page_size=500, order="desc", max_messages=5000): """Fetch daily-log mailbox messages (subject contains 'Log') with full body. Filters receivedDateTime server-side when a window is given; the subject match is client-side. Returns (messages, error); error is None on success, "403:" when mailbox access is denied, else a short string. """ try: app_client = msal.ConfidentialClientApplication( _AZ_CLIENT_ID, authority=_AZ_AUTHORITY_HOME, client_credential=_AZ_SECRET) token_result = app_client.acquire_token_for_client( scopes=["https://graph.microsoft.com/.default"]) token = token_result.get("access_token") if not token: return [], token_result.get("error_description", "no access_token") filters = [] if received_ge: filters.append(f"receivedDateTime ge {received_ge}") if received_le: filters.append(f"receivedDateTime le {received_le}") url = (f"https://graph.microsoft.com/v1.0/users/{SCADA_MAILBOX}/messages" f"?$top={int(page_size)}&$orderby=receivedDateTime {order}" "&$select=subject,receivedDateTime,body") if filters: url += "&$filter=" + quote(" and ".join(filters), safe=" ") headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"} msgs = [] while url and len(msgs) < max_messages: resp = _http_requests.get(url, headers=headers, timeout=60) if resp.status_code == 403: # Keep the "403" prefix — callers match on it — but carry the # real Graph message (e.g. the RAOP AccessPolicy block). return [], "403:" + _graph_error_message(resp) if resp.status_code != 200: return [], f"Graph API {resp.status_code}: {_graph_error_message(resp)}" data = resp.json() msgs.extend(data.get("value", [])) url = data.get("@odata.nextLink") log_msgs = [m for m in msgs if scada_subject_kind(m.get("subject", "")) == "log"] return log_msgs, None except Exception as exc: # network / parse errors return [], str(exc) def _scada_earliest_log_date(): """Earliest log day available in the mailbox (capped by retention), or None. Surfaced so the report filter UI (ST2) can bound its date picker. """ msgs, err = _scada_fetch_log_messages(order="asc", page_size=100, max_messages=100) if err or not msgs: return None dts = [] for msg in msgs: parsed = _parse_scada_log_email(_scada_log_body_text(msg)) dts.extend(r["_dt"] for r in parsed["rows"] if r.get("_dt")) if dts: break if dts: return min(dts).date().isoformat() received = msgs[0].get("receivedDateTime", "") try: return datetime.fromisoformat(received.replace("Z", "+00:00")).date().isoformat() except (ValueError, AttributeError): return None def _build_scada_report(date_from=None, date_to=None, sensors=None): """Aggregate daily-log rows for [date_from, date_to] (inclusive) into per-day tables. date_from/date_to are date objects or None. Log emails arrive the morning after the log day, so the received window is widened by +2 days and rows are binned by their parsed Date Time. ``sensors`` (normalized or native names) filters which sensor columns appear; Date Time + Trigger are always kept. Returns {"days", "columns", "available_sensors", "earliest_date", "row_count", "error"}. """ received_ge = received_le = None if date_from: received_ge = datetime(date_from.year, date_from.month, date_from.day ).strftime("%Y-%m-%dT00:00:00Z") if date_to: widened = datetime(date_to.year, date_to.month, date_to.day) \ + timedelta(days=2) received_le = widened.strftime("%Y-%m-%dT23:59:59Z") msgs, err = _scada_fetch_log_messages(received_ge=received_ge, received_le=received_le) if err: return {"days": [], "columns": [], "available_sensors": [], "earliest_date": None, "row_count": 0, "error": err} # Parse + bin rows by their log-day date. native_columns = [] by_day = {} for msg in msgs: parsed = _parse_scada_log_email(_scada_log_body_text(msg)) if parsed["columns"] and not native_columns: native_columns = parsed["columns"] for row in parsed["rows"]: dt = row.get("_dt") if dt is None: continue day = dt.date() if date_from and day < date_from: continue if date_to and day > date_to: continue by_day.setdefault(day, []).append(row) if not native_columns and by_day: sample = next(iter(by_day.values()))[0] native_columns = [k for k in sample.keys() if k != "_dt"] # Resolve which columns to emit. First + last columns (Date Time, Trigger) # are structural and always retained; the middle columns are sensors. middle = native_columns[1:-1] if len(native_columns) >= 2 else [] available = [_scada_norm_sensor(c) for c in middle] if sensors: wanted = {s.strip() for s in sensors if s.strip()} keep_middle = [c for c in middle if c in wanted or _scada_norm_sensor(c) in wanted] else: keep_middle = middle out_columns = (native_columns[:1] + keep_middle + native_columns[-1:]) \ if len(native_columns) >= 2 else list(native_columns) days = [] row_count = 0 for day in sorted(by_day): day_rows = sorted(by_day[day], key=lambda r: r["_dt"]) table = [[r.get(col, "") for col in out_columns] for r in day_rows] row_count += len(table) days.append({"date": day.isoformat(), "columns": out_columns, "rows": table}) return { "days": days, "columns": out_columns, "available_sensors": available, "earliest_date": _scada_earliest_log_date(), "row_count": row_count, "error": None, } # --------------------------------------------------------------------------- # Report criteria (DVI-1095 P3) — the compliance questions the Kiln Temp report # must answer, e.g. "no kiln went under 45F for more than an hour, and never # above 160F". All four checks are evaluated per Temp sensor over the whole # report window; each can be toggled on/off and the thresholds/duration are # user-editable (persisted in scada_config.json under "report_criteria" so the # Flask UI and the standalone scheduler read the same settings). SCADA_CONFIG_PATH = Path(__file__).resolve().parent / "scada_config.json" SCADA_CRITERIA_DEFAULTS = { "low_limit": 45.0, "high_limit": 160.0, "sustain_minutes": 60, "enabled": { "lowest_overall": True, "lowest_sustained": True, "highest_sustained": True, "highest_overall": True, }, # Change-over-time (rate-of-change) rules (DVI-1148). Each rule flags when a # Temp sensor changes by more than ``max_delta`` F within any rolling # ``window_minutes`` window. Seeded with the priority rule (45F / 60 min); # the list is user-editable so future limits (e.g. 10F / 10 min) just add or # edit a rule. ``enabled`` includes it in the report Summary; ``notify`` opts # the rule into the Sensor-out-of-range notification component. "rate_rules": [ {"id": "roc1", "max_delta": 45.0, "window_minutes": 60, "enabled": True, "notify": True}, ], } _CRITERIA_KEYS = ("lowest_overall", "lowest_sustained", "highest_sustained", "highest_overall") def _normalize_rate_rules(raw_rules): """Validate/normalize a raw ``rate_rules`` list, dropping malformed entries. Each kept rule has a unique string ``id``, positive ``max_delta`` (F) and ``window_minutes`` (int), plus bool ``enabled``/``notify``. Bad rules are dropped rather than raising so a hand-edited config still builds a report. """ out = [] seen = set() for i, r in enumerate(raw_rules): if not isinstance(r, dict): continue try: max_delta = float(r.get("max_delta")) window = int(r.get("window_minutes")) except (TypeError, ValueError): continue if max_delta <= 0 or window <= 0: continue rid = str(r.get("id") or "").strip() or f"roc{i + 1}" base, n = rid, 2 while rid in seen: rid = f"{base}_{n}" n += 1 seen.add(rid) out.append({ "id": rid, "max_delta": max_delta, "window_minutes": window, "enabled": bool(r.get("enabled", True)), "notify": bool(r.get("notify", False)), }) return out def scada_report_criteria(raw=None): """Return merged criteria settings: ``raw`` (or scada_config.json's ``report_criteria``) over SCADA_CRITERIA_DEFAULTS. Bad values fall back to the default rather than raising — the report must still build when the config file is hand-edited.""" if raw is None: raw = {} if SCADA_CONFIG_PATH.is_file(): try: saved = json.loads(SCADA_CONFIG_PATH.read_text()) if isinstance(saved.get("report_criteria"), dict): raw = saved["report_criteria"] except (json.JSONDecodeError, OSError): pass out = {"enabled": dict(SCADA_CRITERIA_DEFAULTS["enabled"])} for key, cast in (("low_limit", float), ("high_limit", float), ("sustain_minutes", int)): try: out[key] = cast(raw.get(key, SCADA_CRITERIA_DEFAULTS[key])) except (TypeError, ValueError): out[key] = SCADA_CRITERIA_DEFAULTS[key] enabled = raw.get("enabled") if isinstance(enabled, dict): for k in _CRITERIA_KEYS: if k in enabled: out["enabled"][k] = bool(enabled[k]) # Rate-of-change rules: an absent key seeds the default rule (backward # compatible); an explicit list (even empty) is honored as given. if "rate_rules" in raw and isinstance(raw["rate_rules"], list): out["rate_rules"] = _normalize_rate_rules(raw["rate_rules"]) else: out["rate_rules"] = [dict(r) for r in SCADA_CRITERIA_DEFAULTS["rate_rules"]] return out def _scada_sensor_series(report): """Extract per-Temp-sensor time series from a built report. Returns {sensor: [(datetime, float), ...]} sorted by time. Non-numeric values ("x.x" gaps) are skipped; RH / Vin / Trigger columns are ignored — the criteria are temperature-compliance checks. """ series = {} for day in report["days"]: cols = day["columns"] temp_idx = [(i, _scada_norm_sensor(c)) for i, c in enumerate(cols) if i not in (0, len(cols) - 1) and "Temp" in c] for row in day["rows"]: m = _SCADA_LOG_DT_RE.match((row[0] or "").strip()) if not m: continue try: mo, da, yr, hh, mm, ss = (int(g) for g in m.groups()) dt = datetime(yr, mo, da, hh, mm, ss) except (ValueError, TypeError): continue for i, sensor in temp_idx: try: val = float(row[i]) except (TypeError, ValueError, IndexError): continue series.setdefault(sensor, []).append((dt, val)) for pts in series.values(): pts.sort(key=lambda p: p[0]) return series def _longest_run(points, predicate, extreme_fn, gap_break_minutes): """Longest consecutive stretch of ``points`` where predicate(value) holds. Returns {"minutes", "extreme", "start", "end"} for the longest run or None; ``extreme_fn`` (min/max) picks the run's reported extreme. A gap between samples longer than ``gap_break_minutes`` breaks the run — missing data must not silently count as a sustained excursion. Duration is last-sample minus first-sample, so a single out-of-range sample is 0 min (the 15-min sample interval bounds what "sustained" can resolve). """ best = None run = [] for dt, val in points: if predicate(val): if run and (dt - run[-1][0]).total_seconds() / 60 > gap_break_minutes: run = [] run.append((dt, val)) minutes = (run[-1][0] - run[0][0]).total_seconds() / 60 if best is None or minutes > best["minutes"]: best = {"minutes": round(minutes, 1), "extreme": extreme_fn(v for _, v in run), "start": run[0][0], "end": run[-1][0]} else: run = [] return best def _max_rate_of_change(points, window_minutes): """Largest absolute value change between any two readings that fall within a rolling ``window_minutes`` window (DVI-1148, change over time). Returns {"delta", "start", "end", "start_val", "end_val", "span_minutes"} for the worst change, or None when fewer than two readings share a window. Tracks absolute change (rise OR fall — both are thermal-integrity risks). Gap-aware: only pairs whose timestamps are within ``window_minutes`` of each other are compared, so a data gap wider than the window is never mistaken for an instantaneous jump. A trailing left pointer keeps this ~O(n · window). """ best = None lo = 0 for hi in range(len(points)): dt_hi, val_hi = points[hi] while lo < hi and (dt_hi - points[lo][0]).total_seconds() / 60 > window_minutes: lo += 1 for k in range(lo, hi): dt_k, val_k = points[k] delta = abs(val_hi - val_k) if best is None or delta > best["delta"]: best = { "delta": round(delta, 2), "start": dt_k, "end": dt_hi, "start_val": val_k, "end_val": val_hi, "span_minutes": round((dt_hi - dt_k).total_seconds() / 60, 1), } return best def scada_rate_rule_check(points, rule): """Evaluate one rate rule against a (datetime, value) series (DVI-1148). Shared by the report Summary and the live notification path. Returns ``(violated, info)`` where ``info`` is the :func:`_max_rate_of_change` result (or None when the window holds fewer than two readings, in which case ``violated`` is False — an unmeasurable window is never an alert). """ roc = _max_rate_of_change(points, int(rule["window_minutes"])) if roc is None: return False, None return roc["delta"] > float(rule["max_delta"]), roc def _fmt_dt(dt): return dt.strftime("%m/%d/%Y %H:%M") if dt else "" def _scada_criteria_summary(report, criteria=None): """Evaluate the enabled report criteria against a built report. Returns {"settings", "sensors": [{sensor, checks: [...], pass}], "pass"}. Each check: {key, label, threshold, observed, when, pass, detail}. ``pass`` is None (informational) when a check has no data to judge. """ crit = scada_report_criteria(criteria if isinstance(criteria, dict) else None) low, high = crit["low_limit"], crit["high_limit"] sustain = crit["sustain_minutes"] enabled = crit["enabled"] series = _scada_sensor_series(report) sensors_out = [] overall_pass = True for sensor in sorted(series): pts = series[sensor] vals = [v for _, v in pts] checks = [] if enabled.get("lowest_overall"): vmin = min(vals) when = next(dt for dt, v in pts if v == vmin) ok = vmin >= low checks.append({ "key": "lowest_overall", "label": f"Lowest temp (entire period) at or above {low:g}F", "threshold": low, "observed": vmin, "when": _fmt_dt(when), "pass": ok, "detail": f"Lowest reading {vmin:g}F at {_fmt_dt(when)}", }) if enabled.get("lowest_sustained"): run = _longest_run(pts, lambda v: v < low, min, max(sustain, 60)) ok = run is None or run["minutes"] <= sustain checks.append({ "key": "lowest_sustained", "label": f"Never below {low:g}F for more than {sustain} min", "threshold": low, "observed": run["extreme"] if run else None, "when": (f"{_fmt_dt(run['start'])} – {_fmt_dt(run['end'])}" f" ({run['minutes']:g} min)") if run else "", "pass": ok, "detail": (f"Longest stretch below {low:g}F lasted " f"{run['minutes']:g} min (low {run['extreme']:g}F)" if run else f"No readings below {low:g}F"), }) if enabled.get("highest_sustained"): run = _longest_run(pts, lambda v: v > high, max, max(sustain, 60)) ok = run is None or run["minutes"] <= sustain checks.append({ "key": "highest_sustained", "label": f"Never above {high:g}F for more than {sustain} min", "threshold": high, "observed": run["extreme"] if run else None, "when": (f"{_fmt_dt(run['start'])} – {_fmt_dt(run['end'])}" f" ({run['minutes']:g} min)") if run else "", "pass": ok, "detail": (f"Longest stretch above {high:g}F lasted " f"{run['minutes']:g} min (high {run['extreme']:g}F)" if run else f"No readings above {high:g}F"), }) if enabled.get("highest_overall"): vmax = max(vals) when = next(dt for dt, v in pts if v == vmax) ok = vmax <= high checks.append({ "key": "highest_overall", "label": f"Highest temp (entire period) at or below {high:g}F", "threshold": high, "observed": vmax, "when": _fmt_dt(when), "pass": ok, "detail": f"Highest reading {vmax:g}F at {_fmt_dt(when)}", }) for rule in crit.get("rate_rules", []): if not rule.get("enabled"): continue md, wm, rid = rule["max_delta"], rule["window_minutes"], rule["id"] roc = _max_rate_of_change(pts, wm) if roc is None: checks.append({ "key": f"rate_of_change_{rid}", "label": f"Change no more than {md:g}F within any {wm} min", "threshold": md, "observed": None, "when": "", "pass": None, "detail": (f"Not enough data within a {wm} min window to " "measure the rate of change"), }) continue ok = roc["delta"] <= md checks.append({ "key": f"rate_of_change_{rid}", "label": f"Change no more than {md:g}F within any {wm} min", "threshold": md, "observed": roc["delta"], "when": (f"{_fmt_dt(roc['start'])} – {_fmt_dt(roc['end'])} " f"({roc['span_minutes']:g} min)"), "pass": ok, "detail": (f"Largest change {roc['delta']:g}F " f"({roc['start_val']:g}F to {roc['end_val']:g}F) over " f"{roc['span_minutes']:g} min"), }) sensor_pass = all(c["pass"] is not False for c in checks) overall_pass = overall_pass and sensor_pass sensors_out.append({"sensor": sensor, "checks": checks, "pass": sensor_pass}) return {"settings": crit, "sensors": sensors_out, "pass": overall_pass if sensors_out else None} def _scada_report_workbook(report, criteria_summary=None): """Build an XLSX workbook from a report dict: an optional criteria Summary sheet first, then one sheet per day (title row, blank row, column header row, data rows).""" from openpyxl import Workbook from openpyxl.styles import Font wb = Workbook() wb.remove(wb.active) if criteria_summary and criteria_summary["sensors"]: ws = wb.create_sheet(title="Summary") bold = Font(bold=True) s = criteria_summary["settings"] ws.append(["Kiln Temperature Report — Criteria Summary"]) ws["A1"].font = Font(bold=True, size=14) days = report["days"] window = f"{days[0]['date']} to {days[-1]['date']}" if days else "no data" ws.append(["Period", window]) ws.append(["Limits", f"Low {s['low_limit']:g}F / High {s['high_limit']:g}F / " f"Sustained window {s['sustain_minutes']} min"]) overall = criteria_summary["pass"] ws.append(["Overall result", "PASS" if overall else "FAIL" if overall is False else "NO DATA"]) cell = ws.cell(row=4, column=2) cell.font = Font(bold=True, color="FF008000" if overall else "FFCC0000") ws.append([]) header = ["Sensor", "Criterion", "Observed", "When", "Result", "Detail"] ws.append(header) for c in range(1, len(header) + 1): ws.cell(row=6, column=c).font = bold for sensor in criteria_summary["sensors"]: for chk in sensor["checks"]: result = ("PASS" if chk["pass"] else "FAIL" if chk["pass"] is False else "N/A") ws.append([sensor["sensor"], chk["label"], chk["observed"], chk["when"], result, chk["detail"]]) rc = ws.cell(row=ws.max_row, column=5) rc.font = Font(bold=chk["pass"] is False, color="FFCC0000" if chk["pass"] is False else "FF008000") widths = {"A": 12, "B": 44, "C": 10, "D": 36, "E": 8, "F": 56} for col, w in widths.items(): ws.column_dimensions[col].width = w if not report["days"]: wb.create_sheet(title="No Data") for day in report["days"]: # Excel sheet titles cap at 31 chars and forbid : \ / ? * [ ] title = re.sub(r"[:\\/?*\[\]]", "-", day["date"])[:31] ws = wb.create_sheet(title=title) ws.append(["X-400 - Log", day["date"]]) ws.append([]) ws.append(day["columns"]) for row in day["rows"]: ws.append(row) buf = io.BytesIO() wb.save(buf) buf.seek(0) return buf # --------------------------------------------------------------------------- # View-model (DVI-1163) — the normalized, JSON-serializable shape the in-app # read-only Report Viewer renders from. It is built from an already-built report # dict + criteria summary (never by re-parsing the XLSX), so the on-demand path # builds it live from the report dict and the scheduler persists a sibling # ``.json`` next to each generated ``.xlsx`` at generation time. The JSON # is deliberately the future edit surface: ``days`` tables + criteria ``settings`` # are all plain data, so a later phase can wire inputs + a save route that # rewrites the JSON and regenerates the XLSX without touching the viewer shape. VIEW_MODEL_VERSION = 1 def build_report_view_model(report, criteria_summary=None, meta=None): """Normalize a built report + criteria summary into the viewer view-model. Returns ``{version, meta, summary, days:[{date, columns, rows}]}``. ``summary`` is the criteria pass/fail block (``{pass, settings, sensors}``) or None when the report type declares no criteria. ``meta`` merges derived facts (period, row count, sensors) with any caller-supplied metadata (schedule name, source, generated-at). All values are JSON-serializable — the criteria summary uses pre-formatted ``when`` strings, so no datetime leaks in. """ days = [{"date": d.get("date"), "columns": list(d.get("columns", [])), "rows": [list(r) for r in d.get("rows", [])]} for d in report.get("days", [])] summary = None if criteria_summary and criteria_summary.get("sensors"): summary = { "pass": criteria_summary.get("pass"), "settings": criteria_summary.get("settings"), "sensors": criteria_summary.get("sensors"), } vm_meta = { "columns": list(report.get("columns", [])), "row_count": report.get("row_count", 0), "available_sensors": list(report.get("available_sensors", [])), "earliest_date": report.get("earliest_date"), "day_count": len(days), "date_from": days[0]["date"] if days else None, "date_to": days[-1]["date"] if days else None, } if meta: vm_meta.update(meta) return { "version": VIEW_MODEL_VERSION, "meta": vm_meta, "summary": summary, "days": days, } def view_model_path_for(xlsx_path): """Sibling ``.json`` path for a generated ``.xlsx``.""" return Path(xlsx_path).with_suffix(".json") def write_report_view_model(xlsx_path, report, criteria_summary=None, meta=None): """Persist the sibling ``.json`` view-model beside a generated XLSX. Returns the JSON path. Raises OSError on write failure so the caller can decide whether that should degrade the (already-persisted) report/email. """ json_path = view_model_path_for(xlsx_path) vm = build_report_view_model(report, criteria_summary, meta) json_path.write_text(json.dumps(vm, indent=2)) return json_path # --------------------------------------------------------------------------- # Report email content (DVI-1165 / DVI-1155 P3) — shared by the Flask app and # the standalone scheduler so both render the same "Attention" summary block and # honor the same toggles. Settings live under the ``report_email`` key in # scada_config.json. The compact summary is exceptions-only by default (D5): it # lists the report's out-of-spec sensors and change-over-time violations rather # than the full pass/fail matrix (that lives in the XLSX Summary sheet + the # in-app viewer). The viewer link is SSO-gated and is added to the *internal* # report email only — external recipients get the XLSX attachment ONLY, no link # (D2, board-directed). The scheduler enforces that split at the call site. REPORT_EMAIL_DEFAULTS = { "include_viewer_link": True, "include_summary": True, "include_out_of_spec": True, "include_rate_violations": True, "summary_mode": "exceptions", # "exceptions" | "full" } def report_email_settings(raw=None): """Merged report-email settings: ``raw`` (or scada_config.json's ``report_email``) over REPORT_EMAIL_DEFAULTS. Bad/absent values fall back to the default rather than raising — the email must still send when the config is missing or hand-edited. """ if raw is None: raw = {} if SCADA_CONFIG_PATH.is_file(): try: saved = json.loads(SCADA_CONFIG_PATH.read_text()) if isinstance(saved.get("report_email"), dict): raw = saved["report_email"] except (json.JSONDecodeError, OSError): pass out = dict(REPORT_EMAIL_DEFAULTS) if isinstance(raw, dict): for key in ("include_viewer_link", "include_summary", "include_out_of_spec", "include_rate_violations"): if key in raw: out[key] = bool(raw[key]) mode = raw.get("summary_mode") if mode in ("exceptions", "full"): out["summary_mode"] = mode return out def _email_esc(value): return (str(value) .replace("&", "&").replace("<", "<").replace(">", ">")) def build_report_email_summary(report, criteria_summary, opts=None): """Compact HTML "Attention" block for a report email body (DVI-1165). Renders the overall PASS/FAIL result plus, per the toggles in ``opts``, the out-of-spec sensor checks (over-high / under-low overall+sustained) and the change-over-time (rate-rule) violations. ``summary_mode`` "exceptions" (default) shows only *failing* checks; "full" shows every enabled check. Returns "" when there is nothing to render — the summary is disabled, the report type has no criteria, or (exceptions-only) every criterion passed and there are no rows to show. Callers append the returned fragment into the email body; an empty string simply adds nothing. """ opts = opts or report_email_settings() if not opts.get("include_summary"): return "" if not criteria_summary or not criteria_summary.get("sensors"): return "" exceptions_only = opts.get("summary_mode", "exceptions") != "full" show_spec = opts.get("include_out_of_spec", True) show_rate = opts.get("include_rate_violations", True) rows = [] # (sensor, is_rate, detail, failed) for sensor in criteria_summary["sensors"]: for chk in sensor.get("checks", []): is_rate = str(chk.get("key", "")).startswith("rate_of_change_") if is_rate and not show_rate: continue if not is_rate and not show_spec: continue failed = chk.get("pass") is False if exceptions_only and not failed: continue rows.append((sensor.get("sensor", ""), is_rate, chk.get("detail", chk.get("label", "")), failed)) overall = criteria_summary.get("pass") if overall is False: status_txt, color, bg = "FAIL", "#b91c1c", "#fef2f2" elif overall is True: status_txt, color, bg = "PASS", "#166534", "#f0fdf4" else: status_txt, color, bg = "NO DATA", "#4b5563", "#f9fafb" parts = [ '
'.format(c=color), '
' '' 'Attention · {st}
'.format(bg=bg, c=color, st=status_txt), ] if rows: trs = [] for sensor, is_rate, detail, failed in rows: tag = "Rate of change" if is_rate else "Out of spec" dot = "#b91c1c" if failed else "#166534" trs.append( '' '{s}' '{tag}' '{d}'.format( s=_email_esc(sensor), dot=dot, tag=tag, d=_email_esc(detail))) parts.append( '' + "".join(trs) + '
') else: parts.append( '
' 'All monitored criteria are within limits.
') parts.append('
') return "".join(parts) # --------------------------------------------------------------------------- # Report type registry (DVI-1095 F3) # --------------------------------------------------------------------------- # A report type bundles the three things that vary between report kinds: # * source — how raw data is fetched/parsed into a report dict (build) # * criteria — the pass/fail check set (defaults + validator + evaluator), # or absent for a type with no compliance criteria # * workbook — how the report dict + criteria summary render to XLSX # "Kiln Temp Report" is the first (today, only) type; it wraps the existing # temperature engine verbatim, so existing schedules and on-demand downloads # stay byte-identical. A new report type is added by registering a descriptor # — the Flask routes and the standalone scheduler resolve types through # ``get_report_type`` and need no per-type changes. class ReportType: """Descriptor bundling a report's data source, criteria, and XLSX layout. ``build(date_from, date_to, sensors) -> report dict`` and ``workbook(report, criteria_summary) -> BytesIO`` are required. Criteria are optional: a type with ``has_criteria`` supplies ``criteria`` (a validator ``raw -> merged dict``) and ``criteria_summary`` (``(report, raw) -> summary``); ``criteria_config_key`` names where its settings live in scada_config.json and ``criteria_keys`` lists its per-check enable-flag keys (for the editor UI). """ def __init__(self, type_id, name, *, build, workbook, order=0, description="", criteria=None, criteria_defaults=None, criteria_summary=None, criteria_keys=(), criteria_config_key=None): self.id = type_id self.name = name self.order = order self.description = description self._build = build self._workbook = workbook self._criteria = criteria self._criteria_summary = criteria_summary self.criteria_defaults = criteria_defaults or {} self.criteria_keys = tuple(criteria_keys) self.criteria_config_key = criteria_config_key @property def has_criteria(self): return self._criteria is not None and self._criteria_summary is not None def build(self, date_from=None, date_to=None, sensors=None): return self._build(date_from, date_to, sensors) def criteria(self, raw=None): """Merged criteria settings for this type, or None when it has none.""" return self._criteria(raw) if self._criteria else None def criteria_summary(self, report, raw=None): """Evaluated pass/fail summary for this type, or None when it has none. ``raw`` is the persisted criteria block (or None to read the shared config file); the summary function re-merges it through the validator. """ return self._criteria_summary(report, raw) if self.has_criteria else None def workbook(self, report, criteria_summary=None): return self._workbook(report, criteria_summary) def info(self): """JSON-serializable descriptor for the report-type picker UI.""" return { "id": self.id, "name": self.name, "order": self.order, "description": self.description, "has_criteria": self.has_criteria, "criteria_keys": list(self.criteria_keys), "criteria_config_key": self.criteria_config_key, } _REPORT_TYPES = {} DEFAULT_REPORT_TYPE_ID = "kiln_temp" def register_report_type(report_type): _REPORT_TYPES[report_type.id] = report_type return report_type def get_report_type(type_id=None): """Resolve a report type by id, falling back to the default (kiln_temp). An unknown or empty id resolves to the default, so schedules created before the registry (which carry no ``report_type``) stay on the Kiln Temp engine — the source of the byte-identical guarantee. """ if type_id and type_id in _REPORT_TYPES: return _REPORT_TYPES[type_id] return _REPORT_TYPES[DEFAULT_REPORT_TYPE_ID] def list_report_types(): """All registered types, ordered for display (order, then name).""" return sorted(_REPORT_TYPES.values(), key=lambda t: (t.order, t.name)) register_report_type(ReportType( "kiln_temp", "Kiln Temp Report", order=0, description=("Daily X-400 kiln temperature logs with low/high/sustained " "compliance checks."), build=_build_scada_report, workbook=_scada_report_workbook, criteria=scada_report_criteria, criteria_defaults=SCADA_CRITERIA_DEFAULTS, criteria_summary=_scada_criteria_summary, criteria_keys=_CRITERIA_KEYS, criteria_config_key="report_criteria", ))