"""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 search(self, query: str, *, limit: int = 20) -> dict[str, Any]: """Return deterministic article-discovery candidates without resolving one.""" if not isinstance(query, str): return { "status": "invalid_query", "query": None, "total_matches": 0, "candidates": [], } if isinstance(limit, bool) or not isinstance(limit, int) or limit <= 0: raise ValueError("limit must be a positive integer") query = query.strip() query_key = _canonical_name(query) if not query_key: return { "status": "search_not_found", "query": query, "total_matches": 0, "candidates": [], } query_tokens = _token_counts(query_key) matches: list[tuple[bool, int, dict[str, Any]]] = [] for article in self.articles: name_key = _canonical_name(str(article.get("name", ""))) name_tokens = _token_counts(name_key) if not _tokens_contained(query_tokens, name_tokens): continue unmatched_tokens = _unmatched_token_count(name_tokens, query_tokens) matches.append((name_key == query_key, unmatched_tokens, article)) matches.sort( key=lambda item: ( not item[0], item[1], str(item[2].get("nr", "")), str(item[2].get("name", "")), ) ) return { "status": "search_results" if matches else "search_not_found", "query": query, "total_matches": len(matches), "candidates": [ _article_search_candidate(article) for _, _, article in matches[:limit] ], } 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 calculate_material_weight( *, roll_length_m: Any, width_m: Any, area_weight_g_m2: Any, ) -> dict[str, Any]: """Calculate material-only roll weight without a core or diameter assumption.""" values = { "roll_length_m": roll_length_m, "width_m": width_m, "area_weight_g_m2": area_weight_g_m2, } invalid = [] normalized = {} for field, value in values.items(): if isinstance(value, bool) or not isinstance(value, (int, float)): invalid.append({"field": field, "message": "must be a number"}) continue number = float(value) if not math.isfinite(number) or number <= 0: invalid.append( {"field": field, "message": "must be greater than zero"} ) continue normalized[field] = number if invalid: return {"status": "invalid_parameter", "invalid": invalid} material_weight = _material_weight_kg( normalized["roll_length_m"], normalized["width_m"], normalized["area_weight_g_m2"], ) if not math.isfinite(material_weight) or material_weight <= 0: return { "status": "invalid_parameter", "invalid": [ { "field": "request", "message": "values are outside the supported calculation range", } ], } return { "status": "success", "weight_scope": "material_only", "inputs": normalized, "material_weight_kg": material_weight, } def calculate_material_length_for_weight( *, target_material_weight_kg: Any, width_m: Any, area_weight_g_m2: Any, ) -> dict[str, Any]: """Calculate material length for a target material-only roll weight.""" values = { "target_material_weight_kg": target_material_weight_kg, "width_m": width_m, "area_weight_g_m2": area_weight_g_m2, } invalid = [] normalized = {} for field, value in values.items(): if isinstance(value, bool) or not isinstance(value, (int, float)): invalid.append({"field": field, "message": "must be a number"}) continue number = float(value) if not math.isfinite(number) or number <= 0: invalid.append({"field": field, "message": "must be greater than zero"}) continue normalized[field] = number if invalid: return {"status": "invalid_parameter", "invalid": invalid} length_m = ( normalized["target_material_weight_kg"] * 1000 / normalized["width_m"] / normalized["area_weight_g_m2"] ) if not math.isfinite(length_m) or length_m <= 0: return { "status": "invalid_parameter", "invalid": [{ "field": "request", "message": "values are outside the supported calculation range", }], } return { "status": "success", "weight_scope": "material_only", "inputs": normalized, "material_length_m": length_m, } def _material_weight_kg( roll_length_m: float, width_m: float, area_weight_g_m2: float ) -> float: return area_weight_g_m2 * roll_length_m * width_m / 1000 def calculate_product_length( *, target_roll_diameter_mm: Any = None, core_diameter_mm: Any = None, thickness_mm: Any = None, thickness_stddev_mm: Any = None, width_m: Any = None, area_weight_g_m2: Any = None, ) -> dict[str, Any]: """Calculate product-length ranges for a target outer roll diameter.""" required = { "target_roll_diameter_mm": target_roll_diameter_mm, "core_diameter_mm": core_diameter_mm, "thickness_mm": thickness_mm, } invalid = [] normalized = {} for field, value in required.items(): if isinstance(value, bool) or not isinstance(value, (int, float)): invalid.append({"field": field, "message": "must be a number"}) continue number = float(value) if not math.isfinite(number) or number <= 0: invalid.append( {"field": field, "message": "must be greater than zero"} ) continue normalized[field] = number if thickness_stddev_mm is None: thickness_stddev = 0.0 elif isinstance(thickness_stddev_mm, bool) or not isinstance( thickness_stddev_mm, (int, float) ): invalid.append( {"field": "thickness_stddev_mm", "message": "must be a number"} ) thickness_stddev = 0.0 else: thickness_stddev = float(thickness_stddev_mm) if not math.isfinite(thickness_stddev) or thickness_stddev < 0: invalid.append( { "field": "thickness_stddev_mm", "message": "must be zero or greater", } ) if invalid: return {"status": "invalid_parameter", "invalid": invalid} target = normalized["target_roll_diameter_mm"] core = normalized["core_diameter_mm"] thickness = normalized["thickness_mm"] if target <= core: return { "status": "invalid_parameter", "invalid": [ { "field": "target_roll_diameter_mm", "message": "must be greater than core_diameter_mm", } ], } minimum_thickness = thickness - (2 * thickness_stddev) if minimum_thickness <= 0: return { "status": "invalid_parameter", "invalid": [ { "field": "thickness_stddev_mm", "message": "minus two standard deviations must remain positive", } ], } length_factor = math.pi * (target**2 - core**2) / 4000 minimum_length = length_factor / (thickness + (2 * thickness_stddev)) average_length = length_factor / thickness maximum_length = length_factor / minimum_thickness lengths = (minimum_length, average_length, maximum_length) if not all(math.isfinite(value) and value > 0 for value in lengths): return { "status": "invalid_parameter", "invalid": [ { "field": "request", "message": "values are outside the supported calculation range", } ], } normalized["thickness_stddev_mm"] = thickness_stddev material_weights = None if width_m is not None and area_weight_g_m2 is not None: weight_result = calculate_material_weight( roll_length_m=average_length, width_m=width_m, area_weight_g_m2=area_weight_g_m2, ) if weight_result["status"] != "success": return weight_result width = weight_result["inputs"]["width_m"] area_weight = weight_result["inputs"]["area_weight_g_m2"] normalized["width_m"] = width normalized["area_weight_g_m2"] = area_weight material_weights = { "minimum_kg": _material_weight_kg(minimum_length, width, area_weight), "average_kg": weight_result["material_weight_kg"], "maximum_kg": _material_weight_kg(maximum_length, width, area_weight), } return { "status": "success", "inputs": normalized, "calculation": { "minimum_product_length_m": minimum_length, "average_product_length_m": average_length, "maximum_product_length_m": maximum_length, "material_weights_kg": material_weights, }, } 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 _token_counts(value: str) -> dict[str, int]: counts: dict[str, int] = {} for token in value.split(): counts[token] = counts.get(token, 0) + 1 return counts def _tokens_contained( query_tokens: dict[str, int], candidate_tokens: dict[str, int] ) -> bool: return all( candidate_tokens.get(token, 0) >= count for token, count in query_tokens.items() ) def _unmatched_token_count( candidate_tokens: dict[str, int], query_tokens: dict[str, int] ) -> int: return sum( max(0, count - query_tokens.get(token, 0)) for token, count in candidate_tokens.items() ) 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_search_candidate(article: dict[str, Any]) -> dict[str, Any]: name = str(article.get("name", "")) number = str(article.get("nr", "")) return { "article_number": number, "name": " ".join(name.split()), "width_m": _article_width(name), "production_site": article_production_site(number), } def article_production_site(article_number: str) -> str | None: """Return production-site metadata established by article-number rules.""" return "Malaysia" if article_number.startswith("8") else None def _article_width(name: str) -> float | None: patterns = ( r"(? 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 = _material_weight_kg(length, width, area_weight) 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