feat: add conversational RollCalc assistant

This commit is contained in:
2026-08-29 22:37:56 +02:00
parent 3b44048250
commit fa3000b325
25 changed files with 6820 additions and 48 deletions
+595
View File
@@ -0,0 +1,595 @@
"""Validated, dependency-free PDF rendering for roll calculation reports."""
from __future__ import annotations
from dataclasses import dataclass
from datetime import datetime
import math
import re
from typing import Any
A4_WIDTH = 595.28
A4_HEIGHT = 841.89
class ReportValidationError(ValueError):
"""Raised when report input is incomplete or unsafe to render."""
@dataclass(frozen=True)
class DiameterValue:
value: float
display: str
def _text(value: Any, field: str, *, required: bool = True, limit: int = 240) -> str:
if value is None:
value = ""
if not isinstance(value, str):
raise ReportValidationError(f"{field} must be text")
value = " ".join(value.split()).strip()
if required and not value:
raise ReportValidationError(f"{field} is required")
if len(value) > limit:
raise ReportValidationError(f"{field} is too long")
return value
def _positive_number(value: Any, field: str) -> float:
if isinstance(value, bool) or not isinstance(value, (int, float)):
raise ReportValidationError(f"{field} must be a number")
value = float(value)
if not math.isfinite(value) or value <= 0:
raise ReportValidationError(f"{field} must be greater than zero")
return value
def _measurement_display(
value: float, display: str, unit: str, field: str, *, decimals: int = 1
) -> str:
match = re.fullmatch(
rf"([0-9]+\.[0-9]{{{decimals}}}) {re.escape(unit)}", display
)
tolerance = (0.5 * (10 ** -decimals)) + 0.000001
if not match or abs(float(match.group(1)) - value) > tolerance:
raise ReportValidationError(f"{field} does not match its numeric value")
return display
def validate_report_payload(payload: Any) -> dict[str, Any]:
"""Validate the internal PDF layout model produced from a calculation result."""
if not isinstance(payload, dict):
raise ReportValidationError("JSON object required")
if payload.get("report_type") != "roll_diameter":
raise ReportValidationError("unsupported or incomplete calculation result")
raw_diameters = payload.get("diameters")
if not isinstance(raw_diameters, dict):
raise ReportValidationError("diameters are required")
diameters: dict[str, DiameterValue] = {}
for key in ("minimum", "average", "maximum"):
raw = raw_diameters.get(key)
if not isinstance(raw, dict):
raise ReportValidationError(f"diameters.{key} is required")
value = _positive_number(raw.get("value"), f"diameters.{key}.value")
display_field = f"diameters.{key}.display"
display = _text(raw.get("display"), display_field, limit=40)
display = _measurement_display(value, display, "mm", display_field)
diameters[key] = DiameterValue(value=value, display=display)
if not (
diameters["minimum"].value
<= diameters["average"].value
<= diameters["maximum"].value
):
raise ReportValidationError("diameter values must be ordered minimum to maximum")
raw_length = payload.get("roll_length")
if not isinstance(raw_length, dict):
raise ReportValidationError("roll_length is required")
length_value = _positive_number(raw_length.get("value"), "roll_length.value")
length_display = _text(
raw_length.get("display"), "roll_length.display", limit=40
)
roll_length = {
"value": length_value,
"display": _measurement_display(
length_value,
length_display,
"m",
"roll_length.display",
decimals=2,
),
}
raw_inputs = payload.get("inputs")
if not isinstance(raw_inputs, list) or not 3 <= len(raw_inputs) <= 16:
raise ReportValidationError("between 3 and 16 calculation inputs are required")
inputs = []
for index, item in enumerate(raw_inputs):
if not isinstance(item, dict):
raise ReportValidationError(f"inputs[{index}] must be an object")
inputs.append(
{
"label": _text(item.get("label"), f"inputs[{index}].label", limit=60),
"value": _text(item.get("value"), f"inputs[{index}].value", limit=100),
}
)
article = None
raw_article = payload.get("article")
if raw_article is not None:
if not isinstance(raw_article, dict):
raise ReportValidationError("article must be an object")
number = _text(raw_article.get("number"), "article.number", required=False, limit=60)
name = _text(raw_article.get("name"), "article.name", required=False, limit=240)
if number or name:
article = {"number": number, "name": name}
roll_weight = None
raw_weight = payload.get("roll_weight")
if raw_weight is not None:
if not isinstance(raw_weight, dict):
raise ReportValidationError("roll_weight must be an object")
weight_value = _positive_number(raw_weight.get("value"), "roll_weight.value")
weight_display = _text(
raw_weight.get("display"), "roll_weight.display", limit=40
)
roll_weight = {
"value": weight_value,
"display": _measurement_display(
weight_value, weight_display, "kg", "roll_weight.display"
),
}
warnings = []
raw_warnings = payload.get("warnings", [])
if not isinstance(raw_warnings, list) or len(raw_warnings) > 12:
raise ReportValidationError("warnings must be a list of at most 12 items")
for index, item in enumerate(raw_warnings):
if not isinstance(item, dict):
raise ReportValidationError(f"warnings[{index}] must be an object")
warnings.append(
{
"title": _text(item.get("title"), f"warnings[{index}].title", limit=120),
"message": _text(item.get("message"), f"warnings[{index}].message", limit=500),
"severity": (
item.get("severity")
if item.get("severity") in {"warning", "critical"}
else "warning"
),
}
)
notes = []
raw_notes = payload.get("notes", [])
if not isinstance(raw_notes, list) or len(raw_notes) > 12:
raise ReportValidationError("notes must be a list of at most 12 items")
for index, item in enumerate(raw_notes):
notes.append(_text(item, f"notes[{index}]", limit=500))
report_text_size = sum(
len(item["title"]) + len(item["message"]) for item in warnings
) + sum(len(note) for note in notes)
if report_text_size > 1800:
raise ReportValidationError("warnings and notes are too long for a one-page report")
return {
"article": article,
"inputs": inputs,
"diameters": diameters,
"roll_length": roll_length,
"roll_weight": roll_weight,
"warnings": warnings,
"notes": notes,
}
def safe_report_filename(article_number: str | None, generated_at: datetime) -> str:
reference = re.sub(r"[^A-Za-z0-9._-]+", "-", article_number or "calculation")
reference = reference.strip("-._")[:60] or "calculation"
return f"roll-calculation_{reference}_{generated_at:%Y-%m-%d_%H%M%S}.pdf"
def report_from_calculation_result(result: Any) -> dict[str, Any]:
"""Adapt an authoritative calculation result to the existing PDF model."""
if not isinstance(result, dict) or result.get("status") != "success":
raise ReportValidationError("a successful calculation result is required")
calculation = result.get("calculation")
effective = result.get("effective_inputs")
request = result.get("request")
provenance = result.get("provenance")
if not all(isinstance(item, dict) for item in (calculation, effective, request, provenance)):
raise ReportValidationError("incomplete authoritative calculation result")
if provenance.get("calculator") != "roll_calculation.calculate_roll":
raise ReportValidationError("unknown calculation provenance")
def effective_value(field: str) -> Any:
item = effective.get(field)
if not isinstance(item, dict) or "value" not in item:
raise ReportValidationError(f"effective input {field} is required")
return item["value"]
length = effective_value("roll_length_m")
core_diameter = effective_value("core_diameter_mm")
core_type = effective_value("core_type")
thickness = effective_value("thickness_mm")
thickness_stddev = effective_value("thickness_stddev_mm")
inputs = [
{"label": "Core diameter", "value": f"{core_diameter:.1f} mm"},
{"label": "Core type", "value": str(core_type)},
{"label": "Product thickness", "value": f"{thickness:.3f} mm"},
{"label": "Thickness stddev", "value": f"{thickness_stddev:.3f} mm"},
{"label": "Product length", "value": f"{length:.2f} m"},
]
if request.get("category"):
inputs.append({"label": "Product category", "value": request["category"]})
if request.get("production_site"):
inputs.append({"label": "Production site", "value": request["production_site"]})
weight = calculation.get("roll_weight_kg")
if weight is not None:
width = effective_value("width_m")
area_weight = effective_value("area_weight_g_m2")
inputs.extend(
(
{"label": "Roll width", "value": f"{width:.2f} m"},
{"label": "Area weight", "value": f"{area_weight:.1f} g/m²"},
)
)
article = result.get("resolved_article")
raw_report = {
"report_type": "roll_diameter",
"article": (
{"number": article.get("number", ""), "name": article.get("name", "")}
if isinstance(article, dict)
else None
),
"inputs": inputs,
"diameters": {
"minimum": {
"value": calculation.get("minimum_diameter_mm"),
"display": f"{calculation.get('minimum_diameter_mm'):.1f} mm",
},
"average": {
"value": calculation.get("average_diameter_mm"),
"display": f"{calculation.get('average_diameter_mm'):.1f} mm",
},
"maximum": {
"value": calculation.get("maximum_diameter_mm"),
"display": f"{calculation.get('maximum_diameter_mm'):.1f} mm",
},
},
"roll_length": {"value": length, "display": f"{length:.2f} m"},
"roll_weight": (
{"value": weight, "display": f"{weight:.1f} kg"}
if weight is not None
else None
),
"warnings": result.get("warnings", []),
"notes": result.get("notes", []),
}
return validate_report_payload(raw_report)
def _pdf_escape(value: str) -> bytes:
value = value.translate(
str.maketrans(
{
"≥": ">=",
"≤": "<=",
"−": "-",
"σ": "sigma",
"π": "pi",
"✓": "OK",
"✗": "X",
}
)
)
encoded = value.encode("cp1252", errors="replace")
return encoded.replace(b"\\", b"\\\\").replace(b"(", b"\\(").replace(b")", b"\\)")
def _wrap(value: str, width: float, font_size: float) -> list[str]:
max_chars = max(12, int(width / (font_size * 0.52)))
words = value.split()
lines: list[str] = []
current = ""
for word in words:
while len(word) > max_chars:
if current:
lines.append(current)
current = ""
lines.append(word[: max_chars - 1] + "-")
word = word[max_chars - 1 :]
candidate = f"{current} {word}".strip()
if len(candidate) <= max_chars:
current = candidate
else:
if current:
lines.append(current)
current = word
if current or not lines:
lines.append(current)
return lines
class _Canvas:
def __init__(self) -> None:
self.commands: list[bytes] = []
def raw(self, command: str) -> None:
self.commands.append(command.encode("ascii"))
def text(
self,
x: float,
y: float,
value: str,
*,
size: float = 9,
bold: bool = False,
color: tuple[float, float, float] = (0.16, 0.20, 0.18),
) -> None:
font = "F2" if bold else "F1"
prefix = (
f"BT /{font} {size:.2f} Tf "
f"{color[0]:.3f} {color[1]:.3f} {color[2]:.3f} rg "
f"1 0 0 1 {x:.2f} {y:.2f} Tm ("
).encode("ascii")
self.commands.append(prefix + _pdf_escape(value) + b") Tj ET")
def line(
self,
x1: float,
y1: float,
x2: float,
y2: float,
*,
color=(0.8, 0.8, 0.8),
width=0.7,
) -> None:
self.raw(
f"{color[0]:.3f} {color[1]:.3f} {color[2]:.3f} RG {width:.2f} w "
f"{x1:.2f} {y1:.2f} m {x2:.2f} {y2:.2f} l S"
)
def rect(
self, x: float, y: float, width: float, height: float, *, fill, stroke=None
) -> None:
command = f"{fill[0]:.3f} {fill[1]:.3f} {fill[2]:.3f} rg "
if stroke:
command += f"{stroke[0]:.3f} {stroke[1]:.3f} {stroke[2]:.3f} RG 0.8 w "
command += f"{x:.2f} {y:.2f} {width:.2f} {height:.2f} re {'B' if stroke else 'f'}"
self.raw(command)
def stream(self) -> bytes:
return b"\n".join(self.commands) + b"\n"
def _make_pdf(content: bytes) -> bytes:
objects = [
b"<< /Type /Catalog /Pages 2 0 R >>",
b"<< /Type /Pages /Kids [3 0 R] /Count 1 >>",
(
b"<< /Type /Page /Parent 2 0 R /MediaBox [0 0 595.28 841.89] "
b"/Resources << /Font << /F1 4 0 R /F2 5 0 R >> /ExtGState << /GS1 6 0 R >> >> "
b"/Contents 7 0 R >>"
),
b"<< /Type /Font /Subtype /Type1 /BaseFont /Helvetica /Encoding /WinAnsiEncoding >>",
b"<< /Type /Font /Subtype /Type1 /BaseFont /Helvetica-Bold /Encoding /WinAnsiEncoding >>",
b"<< /Type /ExtGState /ca 0.085 /CA 0.085 >>",
(
b"<< /Length "
+ str(len(content)).encode("ascii")
+ b" >>\nstream\n"
+ content
+ b"endstream"
),
]
pdf = bytearray(b"%PDF-1.4\n%\xe2\xe3\xcf\xd3\n")
offsets = [0]
for number, obj in enumerate(objects, start=1):
offsets.append(len(pdf))
pdf.extend(f"{number} 0 obj\n".encode("ascii"))
pdf.extend(obj)
pdf.extend(b"\nendobj\n")
xref = len(pdf)
pdf.extend(f"xref\n0 {len(objects) + 1}\n".encode("ascii"))
pdf.extend(b"0000000000 65535 f \n")
for offset in offsets[1:]:
pdf.extend(f"{offset:010d} 00000 n \n".encode("ascii"))
pdf.extend(
(
f"trailer\n<< /Size {len(objects) + 1} /Root 1 0 R >>\n"
f"startxref\n{xref}\n%%EOF\n"
).encode("ascii")
)
return bytes(pdf)
def render_roll_report(
report: dict[str, Any],
*,
generated_at: datetime,
build_info: dict[str, str],
) -> bytes:
"""Render a validated report model as a deliberately compact one-page PDF."""
canvas = _Canvas()
green = (0.10, 0.43, 0.28)
pale_green = (0.91, 0.97, 0.94)
dark = (0.12, 0.18, 0.15)
muted = (0.38, 0.43, 0.40)
margin = 40
usable_width = A4_WIDTH - (2 * margin)
canvas.rect(0, A4_HEIGHT - 92, A4_WIDTH, 92, fill=(0.08, 0.29, 0.20))
canvas.text(margin, 805, "ROLL CALCULATOR", size=20, bold=True, color=(1, 1, 1))
canvas.text(margin, 787, "Calculation report", size=10, color=(0.82, 0.93, 0.87))
generated_text = generated_at.astimezone().strftime("Generated %Y-%m-%d %H:%M %Z")
canvas.text(370, 805, generated_text, size=8, color=(0.88, 0.95, 0.91))
canvas.text(
370,
790,
(
f"Version {build_info.get('version', 'unknown')} | "
f"Build {build_info.get('commit', 'unknown')}"
),
size=8,
color=(0.88, 0.95, 0.91),
)
article = report["article"]
y = 726
if article:
article_heading = " | ".join(
part for part in (article["number"], article["name"]) if part
)
canvas.text(margin, y + 13, "ARTICLE", size=7.5, bold=True, color=green)
article_lines = _wrap(article_heading, usable_width, 11)
if len(article_lines) > 2:
article_lines[1] = article_lines[1].rstrip("- ") + "…"
for line in article_lines[:2]:
canvas.text(margin, y, line, size=11, bold=True, color=dark)
y -= 14
y -= 5
canvas.text(margin, y, "CALCULATION RESULTS", size=8, bold=True, color=green)
y -= 67
card_gap = 10
card_width = (usable_width - 3 * card_gap) / 4
result_cards = (
(("MINIMUM", "DIAMETER"), report["diameters"]["minimum"].display),
(("AVERAGE", "DIAMETER"), report["diameters"]["average"].display),
(("MAXIMUM", "DIAMETER"), report["diameters"]["maximum"].display),
(("ROLL", "LENGTH"), report["roll_length"]["display"]),
)
for index, (label_lines, display) in enumerate(result_cards):
x = margin + index * (card_width + card_gap)
canvas.rect(x, y, card_width, 54, fill=pale_green, stroke=(0.55, 0.77, 0.65))
canvas.text(x + 10, y + 39, label_lines[0], size=6.5, bold=True, color=green)
canvas.text(x + 10, y + 31, label_lines[1], size=6.5, bold=True, color=green)
canvas.text(
x + 10,
y + 10,
display,
size=13.5,
bold=True,
color=dark,
)
y -= 26
if report["roll_weight"]:
canvas.text(margin, y, "CALCULATED ROLL WEIGHT", size=7.5, bold=True, color=green)
canvas.text(
margin + 146,
y - 2,
report["roll_weight"]["display"],
size=12,
bold=True,
color=dark,
)
y -= 28
canvas.line(margin, y, A4_WIDTH - margin, y, color=(0.72, 0.78, 0.74))
y -= 19
canvas.text(margin, y, "INPUT DATA", size=8, bold=True, color=green)
y -= 18
column_width = (usable_width - 24) / 2
row_height = 23
inputs = report["inputs"]
rows = math.ceil(len(inputs) / 2)
for index, item in enumerate(inputs):
column = index % 2
row = index // 2
x = margin + column * (column_width + 24)
item_y = y - row * row_height
canvas.text(x, item_y, item["label"].upper(), size=6.8, bold=True, color=muted)
value = item["value"]
if len(value) > 48:
value = value[:47] + "…"
canvas.text(x, item_y - 11, value, size=9.2, color=dark)
y -= rows * row_height + 9
entries = [
(item["title"], item["message"], item["severity"])
for item in report["warnings"]
]
entries.extend(("Note", note, "note") for note in report["notes"])
if entries:
canvas.line(margin, y, A4_WIDTH - margin, y, color=(0.72, 0.78, 0.74))
y -= 19
canvas.text(margin, y, "WARNINGS & NOTES", size=8, bold=True, color=green)
y -= 17
available = max(72, y - 63)
font_size = 8.2
prepared = []
while True:
prepared = []
total_lines = 0
for title, message, severity in entries:
title_lines = _wrap(title, usable_width - 20, font_size)
message_lines = _wrap(message, usable_width - 20, font_size)
prepared.append((title_lines, message_lines, severity))
total_lines += len(title_lines) + len(message_lines)
line_height = font_size + 2.2
required = total_lines * line_height + len(entries) * 5
if required <= available or font_size <= 5.6:
break
font_size -= 0.4
if required > available:
raise ReportValidationError(
"warnings and notes do not fit on a one-page report"
)
line_height = font_size + 2.2
for title_lines, message_lines, severity in prepared:
if severity == "critical":
color = (0.69, 0.21, 0.17)
elif severity == "warning":
color = (0.61, 0.40, 0.08)
else:
color = green
for line in title_lines:
canvas.text(margin + 8, y, line, size=font_size, bold=True, color=color)
y -= line_height
for line in message_lines:
canvas.text(margin + 8, y, line, size=font_size, color=dark)
y -= line_height
y -= 5
footer_y = 34
canvas.line(margin, footer_y + 21, A4_WIDTH - margin, footer_y + 21, color=(0.72, 0.78, 0.74))
canvas.text(
margin,
footer_y + 7,
(
"Internal advisory calculation. Verify all inputs and outputs for "
"plausibility before use."
),
size=7.2,
color=muted,
)
canvas.text(
A4_WIDTH - 160,
footer_y + 7,
f"Version {build_info.get('version', 'unknown')}",
size=7.2,
color=muted,
)
# Repeated marks are painted into the page content stream, not attached as annotations/layers.
canvas.raw("q /GS1 gs 0.65 0.70 0.67 rg")
angle = math.radians(35)
a = math.cos(angle)
b = math.sin(angle)
for x, mark_y in ((-40, 155), (120, 300), (-20, 445), (145, 590), (-15, 735)):
canvas.raw(
f"BT /F2 29 Tf {a:.5f} {b:.5f} {-b:.5f} {a:.5f} {x:.2f} {mark_y:.2f} Tm "
"(INTERNAL USE ONLY) Tj ET"
)
canvas.raw("Q")
return _make_pdf(canvas.stream())