821 lines
29 KiB
Python
821 lines
29 KiB
Python
"""Deterministic headless domain service for direct roll-diameter calculations."""
|
||
|
||
from __future__ import annotations
|
||
|
||
from dataclasses import asdict, dataclass
|
||
from decimal import Decimal, ROUND_HALF_UP
|
||
from difflib import SequenceMatcher
|
||
import json
|
||
import math
|
||
from pathlib import Path
|
||
import re
|
||
from typing import Any
|
||
|
||
from core_presets import CORE_PRESETS, core_family, matching_core_presets
|
||
|
||
|
||
ARTICLE_DATA_FILE = Path(__file__).parent / "static" / "article-data.json"
|
||
FUZZY_CANDIDATE_THRESHOLD = 0.72
|
||
FUZZY_STRONG_THRESHOLD = 0.92
|
||
FUZZY_STRONG_MARGIN = 0.08
|
||
PRODUCT_FAMILY_ALIASES = {
|
||
"stex": "stex",
|
||
"secutex": "stex",
|
||
"bfix": "bfix",
|
||
"bentofix": "bfix",
|
||
"sgrid": "sgrid",
|
||
"secugrid": "sgrid",
|
||
}
|
||
|
||
REQUEST_FIELDS = {
|
||
"intent",
|
||
"article_number",
|
||
"article_name_hint",
|
||
"roll_length_m",
|
||
"width_m",
|
||
"thickness_mm",
|
||
"thickness_stddev_mm",
|
||
"area_weight_g_m2",
|
||
"core_diameter_mm",
|
||
"core_type",
|
||
"category",
|
||
"production_site",
|
||
"include_roll_weight",
|
||
}
|
||
|
||
FORBIDDEN_RESULT_FIELDS = {
|
||
"diameter_mm",
|
||
"diameters",
|
||
"minimum_diameter_mm",
|
||
"average_diameter_mm",
|
||
"maximum_diameter_mm",
|
||
"roll_weight_kg",
|
||
"warnings",
|
||
"calculation",
|
||
}
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class CalculationRequest:
|
||
"""Raw decision inputs. It deliberately has no engineering result fields."""
|
||
|
||
article_number: str | None = None
|
||
article_name_hint: str | None = None
|
||
roll_length_m: float | None = None
|
||
width_m: float | None = None
|
||
thickness_mm: float | None = None
|
||
thickness_stddev_mm: float | None = None
|
||
area_weight_g_m2: float | None = None
|
||
core_diameter_mm: float | None = None
|
||
core_type: str | None = None
|
||
category: str | None = None
|
||
production_site: str | None = None
|
||
include_roll_weight: bool = False
|
||
intent: str = "calculate_roll_diameter"
|
||
|
||
@classmethod
|
||
def from_dict(cls, payload: Any) -> "CalculationRequest":
|
||
if not isinstance(payload, dict):
|
||
raise ValueError("request must be a JSON object")
|
||
forbidden = sorted(FORBIDDEN_RESULT_FIELDS.intersection(payload))
|
||
if forbidden:
|
||
raise ValueError(
|
||
"calculated result fields are not accepted: " + ", ".join(forbidden)
|
||
)
|
||
unknown = sorted(set(payload) - REQUEST_FIELDS)
|
||
if unknown:
|
||
raise ValueError("unknown request fields: " + ", ".join(unknown))
|
||
if payload.get("intent", "calculate_roll_diameter") != "calculate_roll_diameter":
|
||
raise ValueError("unsupported intent")
|
||
|
||
values = dict(payload)
|
||
if "article_number" in values and values["article_number"] is not None:
|
||
number = values["article_number"]
|
||
if not isinstance(number, str):
|
||
raise ValueError("article_number must be text")
|
||
values["article_number"] = number.strip() or None
|
||
|
||
for field in ("article_name_hint", "core_type", "category", "production_site"):
|
||
if field in values and values[field] is not None:
|
||
if not isinstance(values[field], str):
|
||
raise ValueError(f"{field} must be text")
|
||
values[field] = values[field].strip() or None
|
||
|
||
numeric_fields = (
|
||
"roll_length_m",
|
||
"width_m",
|
||
"thickness_mm",
|
||
"thickness_stddev_mm",
|
||
"area_weight_g_m2",
|
||
"core_diameter_mm",
|
||
)
|
||
for field in numeric_fields:
|
||
if field in values and values[field] is not None:
|
||
value = values[field]
|
||
if isinstance(value, bool) or not isinstance(value, (int, float)):
|
||
raise ValueError(f"{field} must be a number")
|
||
values[field] = float(value)
|
||
|
||
if "include_roll_weight" in values and not isinstance(
|
||
values["include_roll_weight"], bool
|
||
):
|
||
raise ValueError("include_roll_weight must be boolean")
|
||
return cls(**values)
|
||
|
||
def to_dict(self) -> dict[str, Any]:
|
||
return asdict(self)
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class CalculationState:
|
||
"""Structured conversational state that can be modified field by field."""
|
||
|
||
request: CalculationRequest
|
||
|
||
@classmethod
|
||
def from_dict(cls, payload: Any) -> "CalculationState":
|
||
if not isinstance(payload, dict) or "request" not in payload:
|
||
raise ValueError("state.request is required")
|
||
unknown = sorted(set(payload) - {"request"})
|
||
if unknown:
|
||
raise ValueError("unknown state fields: " + ", ".join(unknown))
|
||
return cls(request=CalculationRequest.from_dict(payload["request"]))
|
||
|
||
def to_dict(self) -> dict[str, Any]:
|
||
return {"request": self.request.to_dict()}
|
||
|
||
def apply(self, changes: dict[str, Any]) -> "CalculationState":
|
||
if not isinstance(changes, dict) or not changes:
|
||
raise ValueError("changes must be a non-empty object")
|
||
merged = self.request.to_dict()
|
||
if "article_number" in changes and "article_name_hint" not in changes:
|
||
merged["article_name_hint"] = None
|
||
if "core_type" in changes and "core_diameter_mm" not in changes:
|
||
core_type = changes["core_type"]
|
||
presets = (
|
||
matching_core_presets(core_type)
|
||
if isinstance(core_type, str)
|
||
else ()
|
||
)
|
||
if len(presets) == 1:
|
||
changes = dict(changes)
|
||
changes["core_type"] = presets[0].label
|
||
changes["core_diameter_mm"] = presets[0].diameter_mm
|
||
else:
|
||
merged["core_diameter_mm"] = None
|
||
if "core_diameter_mm" in changes and "core_type" not in changes:
|
||
merged["core_type"] = None
|
||
merged.update(changes)
|
||
return CalculationState(CalculationRequest.from_dict(merged))
|
||
|
||
|
||
class ArticleRepository:
|
||
def __init__(self, articles: list[dict[str, Any]]):
|
||
self.articles = articles
|
||
self.by_number: dict[str, list[dict[str, Any]]] = {}
|
||
for article in articles:
|
||
number = str(article.get("nr", ""))
|
||
self.by_number.setdefault(number, []).append(article)
|
||
|
||
@classmethod
|
||
def load(cls, path: Path = ARTICLE_DATA_FILE) -> "ArticleRepository":
|
||
with path.open(encoding="utf-8") as handle:
|
||
articles = json.load(handle)
|
||
if not isinstance(articles, list):
|
||
raise ValueError("article data must contain a JSON array")
|
||
return cls(articles)
|
||
|
||
def resolve(
|
||
self, article_number: str | None, article_name_hint: str | None
|
||
) -> dict[str, Any]:
|
||
if article_number:
|
||
matches = self.by_number.get(article_number, [])
|
||
if not matches:
|
||
return {
|
||
"status": "article_not_found",
|
||
"article_number": article_number,
|
||
}
|
||
if len(matches) > 1:
|
||
candidates = [_article_candidate(item) for item in matches]
|
||
return {
|
||
"status": "article_ambiguous",
|
||
"field": "article",
|
||
"reason": "ambiguous_article",
|
||
"article_number": article_number,
|
||
"candidates": candidates,
|
||
"matches": [
|
||
{
|
||
"number": candidate["article_number"],
|
||
"name": candidate["name"],
|
||
}
|
||
for candidate in candidates
|
||
],
|
||
}
|
||
article = matches[0]
|
||
if article_name_hint and not _name_hint_matches(
|
||
article_name_hint, str(article.get("name", ""))
|
||
):
|
||
return {
|
||
"status": "article_conflict",
|
||
"article_number": article_number,
|
||
"article_name_hint": article_name_hint,
|
||
"resolved_name": article.get("name", ""),
|
||
}
|
||
return {"status": "resolved", "article": _public_article(article)}
|
||
|
||
if article_name_hint:
|
||
matches = self._matching_names(article_name_hint)
|
||
if len(matches) == 1:
|
||
return {"status": "resolved", "article": _public_article(matches[0])}
|
||
if len(matches) > 1:
|
||
return self._ambiguous_name_resolution(article_name_hint, matches)
|
||
|
||
fuzzy_matches = self._fuzzy_candidates(article_name_hint)
|
||
if fuzzy_matches:
|
||
best_score, best_article = fuzzy_matches[0]
|
||
second_score = fuzzy_matches[1][0] if len(fuzzy_matches) > 1 else 0.0
|
||
if (
|
||
best_score >= FUZZY_STRONG_THRESHOLD
|
||
and best_score - second_score >= FUZZY_STRONG_MARGIN
|
||
):
|
||
return {"status": "resolved", "article": _public_article(best_article)}
|
||
return self._ambiguous_name_resolution(
|
||
article_name_hint,
|
||
[article for _, article in fuzzy_matches],
|
||
reason="uncertain_article_match",
|
||
)
|
||
return {
|
||
"status": "article_not_found",
|
||
"article_name_hint": article_name_hint,
|
||
}
|
||
|
||
return {"status": "not_requested", "article": None}
|
||
|
||
def _matching_names(self, article_name_hint: str) -> list[dict[str, Any]]:
|
||
hint_keys = _article_name_keys(article_name_hint)
|
||
if not hint_keys:
|
||
return []
|
||
return [
|
||
article
|
||
for article in self.articles
|
||
if hint_keys & _article_name_keys(str(article.get("name", "")))
|
||
]
|
||
|
||
def _fuzzy_candidates(
|
||
self, article_name_hint: str
|
||
) -> list[tuple[float, dict[str, Any]]]:
|
||
hint = _base_article_name(article_name_hint)
|
||
hint_tokens = set(hint.split())
|
||
if len(hint_tokens) < 2:
|
||
return []
|
||
scored = []
|
||
for article in self.articles:
|
||
candidate = _base_article_name(str(article.get("name", "")))
|
||
candidate_tokens = set(candidate.split())
|
||
if not _model_designators_match(hint_tokens, candidate_tokens):
|
||
continue
|
||
overlap = len(hint_tokens & candidate_tokens) / len(hint_tokens)
|
||
similarity = SequenceMatcher(
|
||
None,
|
||
hint.replace(" ", ""),
|
||
candidate.replace(" ", ""),
|
||
).ratio()
|
||
score = (0.45 * overlap) + (0.55 * similarity)
|
||
if score >= FUZZY_CANDIDATE_THRESHOLD:
|
||
scored.append((score, article))
|
||
return sorted(scored, key=lambda item: (-item[0], str(item[1].get("nr", ""))))[:5]
|
||
|
||
@staticmethod
|
||
def _ambiguous_name_resolution(
|
||
article_name_hint: str,
|
||
matches: list[dict[str, Any]],
|
||
*,
|
||
reason: str = "ambiguous_article",
|
||
) -> dict[str, Any]:
|
||
return {
|
||
"status": "article_ambiguous",
|
||
"field": "article",
|
||
"reason": reason,
|
||
"article_name_hint": article_name_hint,
|
||
"candidates": [_article_candidate(item) for item in matches],
|
||
"matches": [str(item.get("nr", "")) for item in matches],
|
||
}
|
||
|
||
|
||
def get_article(
|
||
article_number: str | None = None,
|
||
article_name_hint: str | None = None,
|
||
*,
|
||
article_repository: ArticleRepository | None = None,
|
||
) -> dict[str, Any]:
|
||
"""Resolve one article through the canonical article repository."""
|
||
for field, value in (
|
||
("article_number", article_number),
|
||
("article_name_hint", article_name_hint),
|
||
):
|
||
if value is not None and not isinstance(value, str):
|
||
return {
|
||
"status": "invalid_parameter",
|
||
"invalid": [{"field": field, "message": f"{field} must be text"}],
|
||
}
|
||
article_number = article_number.strip() if article_number else None
|
||
article_name_hint = article_name_hint.strip() if article_name_hint else None
|
||
repository = article_repository or ArticleRepository.load()
|
||
return repository.resolve(article_number, article_name_hint)
|
||
|
||
|
||
def _canonical_name(value: str) -> str:
|
||
value = value.casefold().replace("×", " x ")
|
||
tokens = re.findall(r"\d+(?:[.,]\d+)?|[^\W\d_]+", value)
|
||
normalized = [
|
||
PRODUCT_FAMILY_ALIASES.get(
|
||
token.replace(",", "."), token.replace(",", ".")
|
||
)
|
||
for token in tokens
|
||
]
|
||
compacted = []
|
||
index = 0
|
||
while index < len(normalized):
|
||
token = normalized[index]
|
||
if (
|
||
len(token) == 1
|
||
and token.isalpha()
|
||
and index + 1 < len(normalized)
|
||
and normalized[index + 1].isdigit()
|
||
):
|
||
compacted.append(token + normalized[index + 1])
|
||
index += 2
|
||
continue
|
||
compacted.append(token)
|
||
index += 1
|
||
return " ".join(compacted)
|
||
|
||
|
||
def _article_name_keys(value: str) -> set[str]:
|
||
keys = {_canonical_name(value)}
|
||
without_dimensions = re.sub(
|
||
r"\s*,?\s*\d{1,2}[,.]\d+\s*[x×]\s*\d+(?:[,.]\d+)?\s*m\s*$",
|
||
"",
|
||
value,
|
||
flags=re.IGNORECASE,
|
||
)
|
||
keys.add(_canonical_name(without_dimensions))
|
||
without_descriptors = re.sub(r"\([^)]*\)", " ", without_dimensions)
|
||
keys.add(_canonical_name(without_descriptors))
|
||
return {key for key in keys if key}
|
||
|
||
|
||
def _base_article_name(value: str) -> str:
|
||
return min(_article_name_keys(value), key=len, default="")
|
||
|
||
|
||
def _model_designators_match(
|
||
hint_tokens: set[str], candidate_tokens: set[str]
|
||
) -> bool:
|
||
models = {token for token in hint_tokens if re.fullmatch(r"[a-z]\d+", token)}
|
||
return not models or models <= candidate_tokens
|
||
|
||
|
||
def _name_hint_matches(hint: str, resolved_name: str) -> bool:
|
||
hint_tokens = _canonical_name(hint).split()
|
||
resolved_tokens = _canonical_name(resolved_name).split()
|
||
if not hint_tokens:
|
||
return True
|
||
return resolved_tokens[: len(hint_tokens)] == hint_tokens
|
||
|
||
|
||
def _article_candidate(article: dict[str, Any]) -> dict[str, str]:
|
||
return {
|
||
"article_number": str(article.get("nr", "")),
|
||
"name": str(article.get("name", "")),
|
||
}
|
||
|
||
|
||
def _article_width(name: str) -> float | None:
|
||
patterns = (
|
||
r"(?<!\d)(\d{1,2}[,.]\d{1,3})\s*[x×]\s*<?\s*\d+(?:[,.]\d+)?\s*m\b",
|
||
r"(?<!\d)(\d{1,2}[,.]\d{1,3})\s*m\s*breite\b",
|
||
)
|
||
widths = {
|
||
float(match.replace(",", "."))
|
||
for pattern in patterns
|
||
for match in re.findall(pattern, name, flags=re.IGNORECASE)
|
||
}
|
||
valid = {width for width in widths if 0 < width < 100}
|
||
return next(iter(valid)) if len(valid) == 1 else None
|
||
|
||
|
||
def _public_article(article: dict[str, Any]) -> dict[str, Any]:
|
||
return {
|
||
"number": str(article.get("nr", "")),
|
||
"name": str(article.get("name", "")),
|
||
"thickness_mm": article.get("thickness"),
|
||
"thickness_stddev_mm": article.get("thickness_stddev"),
|
||
"area_weight_g_m2": article.get("area_weight"),
|
||
"core_diameter_mm": article.get("core_type"),
|
||
"width_m": _article_width(str(article.get("name", ""))),
|
||
}
|
||
|
||
|
||
def _issue_result(
|
||
status: str,
|
||
request: CalculationRequest,
|
||
*,
|
||
missing: list[str] | None = None,
|
||
ambiguous: list[str] | None = None,
|
||
invalid: list[dict[str, str]] | None = None,
|
||
article_resolution: dict[str, Any] | None = None,
|
||
) -> dict[str, Any]:
|
||
result = {
|
||
"status": status,
|
||
"request": request.to_dict(),
|
||
"state": {"request": request.to_dict()},
|
||
"missing": missing or [],
|
||
"ambiguous": ambiguous or [],
|
||
"invalid": invalid or [],
|
||
}
|
||
if article_resolution is not None:
|
||
result["article_resolution"] = article_resolution
|
||
return result
|
||
|
||
|
||
def _resolve_core(
|
||
diameter: float | None, core_type: str | None
|
||
) -> tuple[float | None, str | None, dict[str, Any] | None]:
|
||
if core_type:
|
||
material = core_family(core_type)
|
||
candidates = matching_core_presets(core_type)
|
||
if diameter is not None:
|
||
matching_diameter = [
|
||
preset
|
||
for preset in CORE_PRESETS
|
||
if math.isclose(preset.diameter_mm, diameter)
|
||
]
|
||
matching_candidates = [
|
||
preset
|
||
for preset in candidates
|
||
if math.isclose(preset.diameter_mm, diameter)
|
||
]
|
||
if len(matching_candidates) == 1:
|
||
preset = matching_candidates[0]
|
||
return preset.diameter_mm, preset.label, None
|
||
if candidates:
|
||
return None, None, {
|
||
"kind": "invalid",
|
||
"field": "core_type",
|
||
"message": "core type conflicts with core diameter",
|
||
}
|
||
if material and matching_diameter:
|
||
return None, None, {
|
||
"kind": "invalid",
|
||
"field": "core_type",
|
||
"message": "core type conflicts with core diameter",
|
||
}
|
||
normalized = re.sub(r"\s+", " ", core_type.casefold()).strip()
|
||
if normalized not in {"custom", "custom / not specified"}:
|
||
return None, None, {
|
||
"kind": "clarification",
|
||
"field": "core_type",
|
||
"message": "unknown core type; provide a known core or diameter",
|
||
"options": [preset.label for preset in CORE_PRESETS],
|
||
}
|
||
label = matching_diameter[0].label if matching_diameter else core_type
|
||
return diameter, label, None
|
||
if len(candidates) == 1:
|
||
return candidates[0].diameter_mm, candidates[0].label, None
|
||
return None, None, {
|
||
"kind": "clarification",
|
||
"field": "core_type",
|
||
"message": "core type does not identify one core diameter",
|
||
"options": [preset.label for preset in candidates or CORE_PRESETS],
|
||
}
|
||
|
||
if diameter is None:
|
||
return None, None, None
|
||
matching = [
|
||
preset
|
||
for preset in CORE_PRESETS
|
||
if math.isclose(preset.diameter_mm, diameter)
|
||
]
|
||
label = matching[0].label if matching else "Custom / not specified"
|
||
return diameter, label, None
|
||
|
||
|
||
def calculate_roll(
|
||
request: CalculationRequest | dict[str, Any],
|
||
*,
|
||
article_repository: ArticleRepository | None = None,
|
||
build_info: dict[str, str] | None = None,
|
||
) -> dict[str, Any]:
|
||
"""Resolve, validate, and calculate one direct roll-diameter request."""
|
||
if not isinstance(request, CalculationRequest):
|
||
try:
|
||
request = CalculationRequest.from_dict(request)
|
||
except ValueError as error:
|
||
return {
|
||
"status": "invalid_parameter",
|
||
"missing": [],
|
||
"ambiguous": [],
|
||
"invalid": [{"field": "request", "message": str(error)}],
|
||
}
|
||
if not isinstance(request, CalculationRequest):
|
||
raise TypeError("request must be CalculationRequest or dict")
|
||
|
||
repository = article_repository or ArticleRepository.load()
|
||
resolution = repository.resolve(request.article_number, request.article_name_hint)
|
||
if resolution["status"] == "article_not_found":
|
||
return _issue_result(
|
||
"article_not_found", request, article_resolution=resolution
|
||
)
|
||
if resolution["status"] == "article_conflict":
|
||
return _issue_result(
|
||
"article_conflict", request, article_resolution=resolution
|
||
)
|
||
if resolution["status"] == "article_ambiguous":
|
||
ambiguous_field = (
|
||
"article_number"
|
||
if resolution.get("article_number")
|
||
else "article_name_hint"
|
||
)
|
||
result = _issue_result(
|
||
"needs_clarification",
|
||
request,
|
||
ambiguous=[ambiguous_field],
|
||
article_resolution=resolution,
|
||
)
|
||
result.update({
|
||
"field": "article",
|
||
"reason": "ambiguous_article",
|
||
"candidates": resolution.get("candidates", []),
|
||
})
|
||
return result
|
||
|
||
article = resolution.get("article")
|
||
effective: dict[str, dict[str, Any]] = {}
|
||
|
||
def choose(
|
||
field: str,
|
||
article_field: str | None = None,
|
||
*,
|
||
zero_is_missing: bool = True,
|
||
) -> Any:
|
||
value = getattr(request, field)
|
||
source = "user"
|
||
if value is None and article and article_field:
|
||
value = article.get(article_field)
|
||
if value == "" or (zero_is_missing and value in (0, 0.0)):
|
||
value = None
|
||
source = "article_master_data"
|
||
if value is not None:
|
||
effective[field] = {"value": value, "source": source}
|
||
return value
|
||
|
||
length = choose("roll_length_m")
|
||
thickness = choose("thickness_mm", "thickness_mm")
|
||
thickness_stddev = choose(
|
||
"thickness_stddev_mm",
|
||
"thickness_stddev_mm",
|
||
zero_is_missing=False,
|
||
)
|
||
if thickness_stddev is None:
|
||
thickness_stddev = 0.0
|
||
effective["thickness_stddev_mm"] = {"value": 0.0, "source": "not_available"}
|
||
width = choose("width_m", "width_m")
|
||
area_weight = choose("area_weight_g_m2", "area_weight_g_m2")
|
||
core_diameter = choose("core_diameter_mm", "core_diameter_mm")
|
||
core_diameter, core_label, core_issue = _resolve_core(
|
||
core_diameter, request.core_type
|
||
)
|
||
if core_issue and core_issue["kind"] == "clarification":
|
||
result = _issue_result(
|
||
"needs_clarification", request, ambiguous=[core_issue["field"]]
|
||
)
|
||
result["clarifications"] = [core_issue]
|
||
return result
|
||
if core_issue:
|
||
return _issue_result("invalid_parameter", request, invalid=[core_issue])
|
||
if core_diameter is not None:
|
||
source = effective.get("core_diameter_mm", {}).get("source", "core_type")
|
||
effective["core_diameter_mm"] = {"value": core_diameter, "source": source}
|
||
effective["core_type"] = {"value": core_label, "source": source}
|
||
|
||
missing = []
|
||
for field, value in (
|
||
("roll_length_m", length),
|
||
("thickness_mm", thickness),
|
||
("core_diameter_mm", core_diameter),
|
||
):
|
||
if value is None:
|
||
missing.append(field)
|
||
if request.include_roll_weight:
|
||
if width is None:
|
||
missing.append("width_m")
|
||
if area_weight is None:
|
||
missing.append("area_weight_g_m2")
|
||
if missing:
|
||
return _issue_result("needs_clarification", request, missing=missing)
|
||
|
||
invalid = []
|
||
positive_values = (
|
||
("roll_length_m", length),
|
||
("thickness_mm", thickness),
|
||
("core_diameter_mm", core_diameter),
|
||
)
|
||
if request.include_roll_weight:
|
||
positive_values += (("width_m", width), ("area_weight_g_m2", area_weight))
|
||
for field, value in positive_values:
|
||
if not math.isfinite(value) or value <= 0:
|
||
invalid.append({"field": field, "message": "must be greater than zero"})
|
||
if not math.isfinite(thickness_stddev) or thickness_stddev < 0:
|
||
invalid.append(
|
||
{"field": "thickness_stddev_mm", "message": "must be zero or greater"}
|
||
)
|
||
minimum_thickness = thickness - (2 * thickness_stddev)
|
||
if minimum_thickness <= 0:
|
||
invalid.append(
|
||
{
|
||
"field": "thickness_stddev_mm",
|
||
"message": "minus two standard deviations must remain positive",
|
||
}
|
||
)
|
||
if invalid:
|
||
return _issue_result("invalid_parameter", request, invalid=invalid)
|
||
|
||
try:
|
||
average = math.sqrt(
|
||
core_diameter**2 + (4 * length * 1000 * thickness) / math.pi
|
||
)
|
||
lower = math.sqrt(
|
||
core_diameter**2 + (4 * length * 1000 * minimum_thickness) / math.pi
|
||
)
|
||
upper = math.sqrt(
|
||
core_diameter**2
|
||
+ (4 * length * 1000 * (thickness + 2 * thickness_stddev))
|
||
/ math.pi
|
||
)
|
||
roll_weight = None
|
||
if request.include_roll_weight:
|
||
roll_weight = area_weight * length * width / 1000
|
||
except OverflowError:
|
||
return _issue_result(
|
||
"invalid_parameter",
|
||
request,
|
||
invalid=[
|
||
{
|
||
"field": "request",
|
||
"message": "values are outside the supported calculation range",
|
||
}
|
||
],
|
||
)
|
||
|
||
calculated_values = (average, lower, upper)
|
||
if roll_weight is not None:
|
||
calculated_values += (roll_weight,)
|
||
if not all(math.isfinite(value) and value > 0 for value in calculated_values):
|
||
return _issue_result(
|
||
"invalid_parameter",
|
||
request,
|
||
invalid=[
|
||
{
|
||
"field": "request",
|
||
"message": "values are outside the supported calculation range",
|
||
}
|
||
],
|
||
)
|
||
|
||
warnings: list[dict[str, str]] = []
|
||
notes: list[str] = []
|
||
if roll_weight is not None and (request.category or "").casefold() == "bentofix":
|
||
if roll_weight > 2750:
|
||
warnings.append(
|
||
{
|
||
"severity": "critical",
|
||
"title": "Weight Limit Exceeded",
|
||
"message": "Roll weight exceeds 2750 kg maximum",
|
||
}
|
||
)
|
||
elif roll_weight >= 1700:
|
||
core_message = (
|
||
"Core diameter OK"
|
||
if core_diameter >= 170
|
||
else f"Core diameter {core_diameter:g} mm (min 170 mm required)"
|
||
)
|
||
warnings.append(
|
||
{
|
||
"severity": "warning",
|
||
"title": "Heavy Roll – Special Equipment Required",
|
||
"message": (
|
||
"Special equipment required for rolls ≥ 1700 kg. "
|
||
f"{core_message}."
|
||
),
|
||
}
|
||
)
|
||
else:
|
||
notes.append(
|
||
f"Roll weight {roll_weight:.1f} kg is within standard limits. "
|
||
"No special equipment required."
|
||
)
|
||
|
||
calculation = {
|
||
"effective_roll_length_m": length,
|
||
"minimum_diameter_mm": min(lower, upper),
|
||
"average_diameter_mm": average,
|
||
"maximum_diameter_mm": max(lower, upper),
|
||
"roll_weight_kg": roll_weight,
|
||
}
|
||
return {
|
||
"status": "success",
|
||
"request": request.to_dict(),
|
||
"state": {"request": request.to_dict()},
|
||
"resolved_article": article,
|
||
"effective_inputs": effective,
|
||
"calculation": calculation,
|
||
"warnings": warnings,
|
||
"notes": notes,
|
||
"provenance": {
|
||
"calculator": "roll_calculation.calculate_roll",
|
||
"build": build_info or {},
|
||
},
|
||
}
|
||
|
||
|
||
def transport_roll_inputs_from_roll_calculation(
|
||
result: Any,
|
||
) -> dict[str, float]:
|
||
"""Derive browser-equivalent transport inputs from a roll calculation."""
|
||
if not isinstance(result, dict) or result.get("status") != "success":
|
||
raise ValueError("roll calculation result must have status success")
|
||
|
||
calculation = result.get("calculation")
|
||
effective_inputs = result.get("effective_inputs")
|
||
if not isinstance(calculation, dict) or not isinstance(effective_inputs, dict):
|
||
raise ValueError("successful roll calculation result is incomplete")
|
||
|
||
def calculation_number(field: str) -> float:
|
||
return _transport_roll_input_number(calculation.get(field), field)
|
||
|
||
def effective_number(field: str) -> float:
|
||
entry = effective_inputs.get(field)
|
||
if not isinstance(entry, dict):
|
||
raise ValueError(f"successful roll calculation result is missing {field}")
|
||
return _transport_roll_input_number(entry.get("value"), field)
|
||
|
||
return {
|
||
"roll_diameter_mm": _browser_one_decimal(
|
||
calculation_number("average_diameter_mm")
|
||
),
|
||
"core_diameter_mm": effective_number("core_diameter_mm"),
|
||
"roll_width_m": effective_number("width_m"),
|
||
"roll_weight_kg": _browser_one_decimal(
|
||
calculation_number("roll_weight_kg")
|
||
),
|
||
"product_length_m": calculation_number("effective_roll_length_m"),
|
||
}
|
||
|
||
|
||
def _transport_roll_input_number(value: Any, field: str) -> float:
|
||
if isinstance(value, bool) or not isinstance(value, (int, float)):
|
||
raise ValueError(f"successful roll calculation result is missing {field}")
|
||
number = float(value)
|
||
if not math.isfinite(number) or number <= 0:
|
||
raise ValueError(f"successful roll calculation result has invalid {field}")
|
||
return number
|
||
|
||
|
||
def _browser_one_decimal(value: float) -> float:
|
||
"""Match the browser's positive-number ``toFixed(1)`` transport values."""
|
||
return float(
|
||
Decimal.from_float(value).quantize(Decimal("0.1"), rounding=ROUND_HALF_UP)
|
||
)
|
||
|
||
|
||
def modify_calculation(
|
||
state: CalculationState | dict[str, Any],
|
||
changes: dict[str, Any],
|
||
*,
|
||
article_repository: ArticleRepository | None = None,
|
||
build_info: dict[str, str] | None = None,
|
||
) -> dict[str, Any]:
|
||
"""Apply controlled changes to structured state and recalculate."""
|
||
try:
|
||
current = (
|
||
state
|
||
if isinstance(state, CalculationState)
|
||
else CalculationState.from_dict(state)
|
||
)
|
||
modified = current.apply(changes)
|
||
except ValueError as error:
|
||
return {
|
||
"status": "invalid_parameter",
|
||
"missing": [],
|
||
"ambiguous": [],
|
||
"invalid": [{"field": "changes", "message": str(error)}],
|
||
}
|
||
result = calculate_roll(
|
||
modified.request,
|
||
article_repository=article_repository,
|
||
build_info=build_info,
|
||
)
|
||
result["state"] = modified.to_dict()
|
||
return result
|