719 lines
31 KiB
Python
719 lines
31 KiB
Python
"""Configured, deterministic production-machine feasibility checks."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
import math
|
|
from pathlib import Path
|
|
import unicodedata
|
|
from typing import Any
|
|
|
|
import yaml
|
|
|
|
from roll_calculation import (
|
|
calculate_material_length_for_weight,
|
|
calculate_material_weight,
|
|
calculate_product_length,
|
|
calculate_roll,
|
|
)
|
|
|
|
|
|
MACHINE_CONFIG_FILE = Path(__file__).parent / "config" / "machines.yaml"
|
|
SCHEMA_VERSION = 1
|
|
CONSTRAINT_KEYS = {
|
|
"roll_weight_kg",
|
|
"roll_diameter_mm",
|
|
"allowed_core_diameters_mm",
|
|
"product_width_m",
|
|
"line_speed_m_per_min",
|
|
}
|
|
MAX_LENGTH_TIE_REL_TOLERANCE = 1e-12
|
|
MAX_LENGTH_TIE_ABS_TOLERANCE_M = 1e-9
|
|
|
|
|
|
class MachineConfigurationError(ValueError):
|
|
"""Raised when the machine configuration is invalid."""
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class NumericRange:
|
|
minimum: float | None = None
|
|
maximum: float | None = None
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class MachineConstraints:
|
|
roll_weight_kg: NumericRange | None
|
|
roll_diameter_mm: NumericRange | None
|
|
allowed_core_diameters_mm: tuple[float, ...] | None
|
|
product_width_m: NumericRange | None
|
|
line_speed_m_per_min: NumericRange | None
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Machine:
|
|
id: str
|
|
name: str
|
|
aliases: tuple[str, ...]
|
|
location: str
|
|
constraints: MachineConstraints
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class MachineFeasibilityRequest:
|
|
machine: str
|
|
roll_weight_kg: float | None = None
|
|
average_diameter_mm: float | None = None
|
|
maximum_diameter_mm: float | None = None
|
|
core_diameter_mm: float | None = None
|
|
product_width_m: float | None = None
|
|
line_speed_m_per_min: float | None = None
|
|
|
|
@classmethod
|
|
def from_dict(cls, payload: Any) -> "MachineFeasibilityRequest":
|
|
if not isinstance(payload, dict):
|
|
raise ValueError("request must be a JSON object")
|
|
allowed = set(cls.__dataclass_fields__)
|
|
unknown = sorted(set(payload) - allowed)
|
|
if unknown:
|
|
raise ValueError("unknown request fields: " + ", ".join(unknown))
|
|
machine = payload.get("machine")
|
|
if not isinstance(machine, str) or not machine.strip():
|
|
raise ValueError("machine must be non-empty text")
|
|
values: dict[str, Any] = {"machine": machine.strip()}
|
|
for field in allowed - {"machine"}:
|
|
value = payload.get(field)
|
|
if value is None:
|
|
values[field] = None
|
|
continue
|
|
if isinstance(value, bool) or not isinstance(value, (int, float)):
|
|
raise ValueError(f"{field} must be a number")
|
|
number = float(value)
|
|
if not math.isfinite(number) or number <= 0:
|
|
raise ValueError(f"{field} must be greater than zero")
|
|
values[field] = number
|
|
average = values["average_diameter_mm"]
|
|
maximum = values["maximum_diameter_mm"]
|
|
if average is not None and maximum is not None and maximum < average:
|
|
raise ValueError(
|
|
"maximum_diameter_mm must be greater than or equal to "
|
|
"average_diameter_mm"
|
|
)
|
|
return cls(**values)
|
|
|
|
|
|
def _normalize_lookup(value: str) -> str:
|
|
return " ".join(unicodedata.normalize("NFKC", value).casefold().split())
|
|
|
|
|
|
def _number(value: Any, field: str) -> float:
|
|
if isinstance(value, bool) or not isinstance(value, (int, float)):
|
|
raise MachineConfigurationError(f"{field} must be a number")
|
|
value = float(value)
|
|
if not math.isfinite(value) or value <= 0:
|
|
raise MachineConfigurationError(f"{field} must be greater than zero")
|
|
return value
|
|
|
|
|
|
def _range(value: Any, field: str) -> NumericRange | None:
|
|
if value is None:
|
|
return None
|
|
if not isinstance(value, dict):
|
|
raise MachineConfigurationError(f"{field} must be an object or null")
|
|
unknown = sorted(set(value) - {"minimum", "maximum"})
|
|
if unknown:
|
|
raise MachineConfigurationError(f"{field} has unknown keys: {', '.join(unknown)}")
|
|
if not value:
|
|
raise MachineConfigurationError(f"{field} must contain a limit")
|
|
minimum = _number(value["minimum"], f"{field}.minimum") if "minimum" in value else None
|
|
maximum = _number(value["maximum"], f"{field}.maximum") if "maximum" in value else None
|
|
if minimum is None and maximum is None:
|
|
raise MachineConfigurationError(f"{field} must contain a limit")
|
|
if minimum is not None and maximum is not None and minimum > maximum:
|
|
raise MachineConfigurationError(f"{field}.minimum must not exceed maximum")
|
|
return NumericRange(minimum=minimum, maximum=maximum)
|
|
|
|
|
|
def _core_values(value: Any) -> tuple[float, ...] | None:
|
|
if value is None:
|
|
return None
|
|
if not isinstance(value, list) or not value:
|
|
raise MachineConfigurationError(
|
|
"allowed_core_diameters_mm must be a non-empty list or null"
|
|
)
|
|
values = tuple(_number(item, "allowed_core_diameters_mm item") for item in value)
|
|
if len(set(values)) != len(values):
|
|
raise MachineConfigurationError("allowed_core_diameters_mm must not repeat values")
|
|
return values
|
|
|
|
|
|
class MachineRepository:
|
|
"""Load, validate, and resolve configured production machines."""
|
|
|
|
def __init__(self, machines: tuple[Machine, ...]):
|
|
self.machines = machines
|
|
self._by_lookup: dict[str, Machine] = {}
|
|
for machine in machines:
|
|
identifiers = (machine.id, machine.name, *machine.aliases)
|
|
local: set[str] = set()
|
|
for position, value in enumerate(identifiers):
|
|
normalized = _normalize_lookup(value)
|
|
# A machine such as K1 legitimately has identical stable ID
|
|
# and display name. Any alias collision remains invalid.
|
|
if normalized in local and position > 1:
|
|
raise MachineConfigurationError(
|
|
f"duplicate machine lookup identifier: {value}"
|
|
)
|
|
local.add(normalized)
|
|
existing = self._by_lookup.get(normalized)
|
|
if existing is not None and existing is not machine:
|
|
raise MachineConfigurationError(
|
|
f"duplicate machine lookup identifier: {value}"
|
|
)
|
|
self._by_lookup[normalized] = machine
|
|
|
|
@classmethod
|
|
def load(cls, path: Path = MACHINE_CONFIG_FILE) -> "MachineRepository":
|
|
try:
|
|
with path.open(encoding="utf-8") as handle:
|
|
data = yaml.safe_load(handle)
|
|
except OSError as error:
|
|
raise MachineConfigurationError(
|
|
f"unable to load machine configuration: {error}"
|
|
) from error
|
|
except yaml.YAMLError as error:
|
|
raise MachineConfigurationError(
|
|
f"invalid machine configuration YAML: {error}"
|
|
) from error
|
|
return cls(_parse_config(data))
|
|
|
|
def resolve(self, machine: str) -> dict[str, Any]:
|
|
if not isinstance(machine, str) or not machine.strip():
|
|
return {
|
|
"status": "invalid_parameter",
|
|
"invalid": [{"field": "machine", "message": "machine must be non-empty text"}],
|
|
}
|
|
resolved = self._by_lookup.get(_normalize_lookup(machine))
|
|
if resolved is None:
|
|
return {"status": "machine_not_found", "machine": machine.strip()}
|
|
matched_by = "id" if _normalize_lookup(machine) == _normalize_lookup(resolved.id) else "name"
|
|
if matched_by == "name" and _normalize_lookup(machine) != _normalize_lookup(resolved.name):
|
|
matched_by = "alias"
|
|
return {
|
|
"status": "resolved",
|
|
"machine": _public_machine(resolved),
|
|
"resolved_machine": resolved,
|
|
"matched_by": matched_by,
|
|
"matched_value": machine.strip(),
|
|
}
|
|
|
|
|
|
def _parse_config(data: Any) -> tuple[Machine, ...]:
|
|
if not isinstance(data, dict):
|
|
raise MachineConfigurationError("machine configuration must be an object")
|
|
unknown = sorted(set(data) - {"schema_version", "machines"})
|
|
if unknown:
|
|
raise MachineConfigurationError("unknown configuration keys: " + ", ".join(unknown))
|
|
if (
|
|
isinstance(data.get("schema_version"), bool)
|
|
or data.get("schema_version") != SCHEMA_VERSION
|
|
):
|
|
raise MachineConfigurationError(f"unsupported schema_version: {data.get('schema_version')!r}")
|
|
rows = data.get("machines")
|
|
if not isinstance(rows, list) or not rows:
|
|
raise MachineConfigurationError("machines must be a non-empty list")
|
|
machines: list[Machine] = []
|
|
ids: set[str] = set()
|
|
for index, row in enumerate(rows):
|
|
prefix = f"machines[{index}]"
|
|
if not isinstance(row, dict):
|
|
raise MachineConfigurationError(f"{prefix} must be an object")
|
|
unknown = sorted(set(row) - {"id", "name", "aliases", "location", "constraints"})
|
|
if unknown:
|
|
raise MachineConfigurationError(f"{prefix} has unknown keys: {', '.join(unknown)}")
|
|
if set(row) != {"id", "name", "aliases", "location", "constraints"}:
|
|
raise MachineConfigurationError(f"{prefix} must contain id, name, aliases, location, and constraints")
|
|
text = {}
|
|
for key in ("id", "name", "location"):
|
|
value = row[key]
|
|
if not isinstance(value, str) or not value.strip():
|
|
raise MachineConfigurationError(f"{prefix}.{key} must be non-empty text")
|
|
text[key] = value.strip()
|
|
if text["id"] in ids:
|
|
raise MachineConfigurationError(f"duplicate machine id: {text['id']}")
|
|
ids.add(text["id"])
|
|
aliases = row["aliases"]
|
|
if not isinstance(aliases, list) or not all(isinstance(item, str) and item.strip() for item in aliases):
|
|
raise MachineConfigurationError(f"{prefix}.aliases must be a list of non-empty text")
|
|
constraints = row["constraints"]
|
|
if not isinstance(constraints, dict) or set(constraints) != CONSTRAINT_KEYS:
|
|
raise MachineConfigurationError(f"{prefix}.constraints must contain exactly the supported constraint keys")
|
|
machines.append(Machine(
|
|
id=text["id"], name=text["name"], aliases=tuple(item.strip() for item in aliases),
|
|
location=text["location"], constraints=MachineConstraints(
|
|
roll_weight_kg=_range(constraints["roll_weight_kg"], f"{prefix}.constraints.roll_weight_kg"),
|
|
roll_diameter_mm=_range(constraints["roll_diameter_mm"], f"{prefix}.constraints.roll_diameter_mm"),
|
|
allowed_core_diameters_mm=_core_values(constraints["allowed_core_diameters_mm"]),
|
|
product_width_m=_range(constraints["product_width_m"], f"{prefix}.constraints.product_width_m"),
|
|
line_speed_m_per_min=_range(constraints["line_speed_m_per_min"], f"{prefix}.constraints.line_speed_m_per_min"),
|
|
),
|
|
))
|
|
return tuple(machines)
|
|
|
|
|
|
def _range_public(value: NumericRange | None, unit: str) -> dict[str, Any] | None:
|
|
if value is None:
|
|
return None
|
|
return {"minimum": value.minimum, "maximum": value.maximum, "unit": unit}
|
|
|
|
|
|
def _public_machine(machine: Machine) -> dict[str, Any]:
|
|
constraints = machine.constraints
|
|
return {
|
|
"id": machine.id, "name": machine.name, "aliases": list(machine.aliases),
|
|
"location": machine.location,
|
|
"constraints": {
|
|
"roll_weight_kg": _range_public(constraints.roll_weight_kg, "kg"),
|
|
"roll_diameter_mm": _range_public(constraints.roll_diameter_mm, "mm"),
|
|
"allowed_core_diameters_mm": list(constraints.allowed_core_diameters_mm) if constraints.allowed_core_diameters_mm is not None else None,
|
|
"product_width_m": _range_public(constraints.product_width_m, "m"),
|
|
"line_speed_m_per_min": _range_public(constraints.line_speed_m_per_min, "m/min"),
|
|
},
|
|
}
|
|
|
|
|
|
def get_machine(machine: str | None, *, repository: MachineRepository | None = None) -> dict[str, Any]:
|
|
"""Resolve one configured machine without applying production rules."""
|
|
result = (repository or MachineRepository.load()).resolve(machine) # type: ignore[arg-type]
|
|
return {key: value for key, value in result.items() if key != "resolved_machine"}
|
|
|
|
|
|
def _range_check(name: str, actual: float | None, limits: NumericRange | None, unit: str) -> dict[str, Any]:
|
|
if limits is None:
|
|
return {"constraint": name, "state": "not_configured", "unit": unit}
|
|
if actual is None:
|
|
return {"constraint": name, "state": "not_evaluated", "reason": "missing_actual_value", "limits": _range_public(limits, unit), "unit": unit}
|
|
failures = []
|
|
margins: dict[str, dict[str, float | str]] = {}
|
|
if limits.minimum is not None:
|
|
margins["margin_to_minimum"] = {"value": actual - limits.minimum, "unit": unit}
|
|
if actual < limits.minimum:
|
|
failures.append("below_minimum")
|
|
if limits.maximum is not None:
|
|
margins["margin_to_maximum"] = {"value": limits.maximum - actual, "unit": unit}
|
|
if actual > limits.maximum:
|
|
failures.append("above_maximum")
|
|
return {"constraint": name, "state": "failed" if failures else "passed", "actual": {"value": actual, "unit": unit}, "limits": _range_public(limits, unit), "violations": failures, "margins": margins}
|
|
|
|
|
|
def _diameter_check(average: float | None, maximum: float | None, limits: NumericRange | None) -> dict[str, Any]:
|
|
if limits is None:
|
|
return {"constraint": "roll_diameter_mm", "state": "not_configured", "unit": "mm"}
|
|
if average is None or maximum is None:
|
|
return {"constraint": "roll_diameter_mm", "state": "not_evaluated", "reason": "missing_actual_value", "limits": _range_public(limits, "mm"), "unit": "mm"}
|
|
violations = []
|
|
if limits.minimum is not None and average < limits.minimum:
|
|
violations.append("average_below_minimum")
|
|
if limits.maximum is not None and average > limits.maximum:
|
|
violations.append("average_above_maximum")
|
|
warning = limits.maximum is not None and average <= limits.maximum and maximum > limits.maximum
|
|
state = "failed" if violations else "warning" if warning else "passed"
|
|
margins: dict[str, dict[str, float | str]] = {}
|
|
if limits.minimum is not None:
|
|
margins["average_margin_to_minimum"] = {"value": average - limits.minimum, "unit": "mm"}
|
|
if limits.maximum is not None:
|
|
margins["average_margin_to_maximum"] = {"value": limits.maximum - average, "unit": "mm"}
|
|
margins["maximum_margin_to_maximum"] = {"value": limits.maximum - maximum, "unit": "mm"}
|
|
return {
|
|
"constraint": "roll_diameter_mm", "state": state,
|
|
"actual": {"average_diameter_mm": average, "maximum_diameter_mm": maximum, "unit": "mm"},
|
|
"limits": _range_public(limits, "mm"), "violations": violations,
|
|
"warnings": ["maximum_diameter_above_maximum"] if warning else [], "margins": margins,
|
|
"minimum_evaluation": "average_diameter_mm",
|
|
}
|
|
|
|
|
|
def _core_check(actual: float | None, allowed: tuple[float, ...] | None) -> dict[str, Any]:
|
|
if allowed is None:
|
|
return {"constraint": "allowed_core_diameters_mm", "state": "not_configured", "unit": "mm"}
|
|
if actual is None:
|
|
return {"constraint": "allowed_core_diameters_mm", "state": "not_evaluated", "reason": "missing_actual_value", "allowed_values": list(allowed), "unit": "mm"}
|
|
return {"constraint": "allowed_core_diameters_mm", "state": "passed" if actual in allowed else "failed", "actual": {"value": actual, "unit": "mm"}, "allowed_values": list(allowed), "unit": "mm", "violations": [] if actual in allowed else ["not_allowed"]}
|
|
|
|
|
|
def check_production_feasibility(
|
|
request: MachineFeasibilityRequest | dict[str, Any], *, repository: MachineRepository | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Evaluate all configured V1 roll constraints without short-circuiting.
|
|
|
|
Diameter minima use RollCalc's nominal/average diameter. A maximum-diameter
|
|
variation exceedance is a warning only when the nominal diameter still fits.
|
|
Roll weight is the caller-supplied material-only weight in V1.
|
|
"""
|
|
if not isinstance(request, MachineFeasibilityRequest):
|
|
try:
|
|
request = MachineFeasibilityRequest.from_dict(request)
|
|
except ValueError as error:
|
|
return {"status": "invalid_parameter", "invalid": [{"field": "request", "message": str(error)}]}
|
|
resolved = (repository or MachineRepository.load()).resolve(request.machine)
|
|
if resolved["status"] != "resolved":
|
|
return {key: value for key, value in resolved.items() if key != "resolved_machine"}
|
|
machine: Machine = resolved["resolved_machine"]
|
|
constraints = machine.constraints
|
|
checks = [
|
|
_range_check("roll_weight_kg", request.roll_weight_kg, constraints.roll_weight_kg, "kg"),
|
|
_diameter_check(request.average_diameter_mm, request.maximum_diameter_mm, constraints.roll_diameter_mm),
|
|
_core_check(request.core_diameter_mm, constraints.allowed_core_diameters_mm),
|
|
_range_check("product_width_m", request.product_width_m, constraints.product_width_m, "m"),
|
|
{"constraint": "line_speed_m_per_min", "state": "not_evaluated", "reason": "outside_initial_roll_feasibility_scope", "limits": _range_public(constraints.line_speed_m_per_min, "m/min"), "unit": "m/min"},
|
|
]
|
|
required = {}
|
|
if constraints.roll_weight_kg is not None:
|
|
required["roll_weight_kg"] = request.roll_weight_kg
|
|
if constraints.roll_diameter_mm is not None:
|
|
required["average_diameter_mm"] = request.average_diameter_mm
|
|
required["maximum_diameter_mm"] = request.maximum_diameter_mm
|
|
if constraints.allowed_core_diameters_mm is not None:
|
|
required["core_diameter_mm"] = request.core_diameter_mm
|
|
if constraints.product_width_m is not None:
|
|
required["product_width_m"] = request.product_width_m
|
|
missing = [field for field, value in required.items() if value is None]
|
|
failed = [check["constraint"] for check in checks if check["state"] == "failed"]
|
|
warnings = [check["constraint"] for check in checks if check["state"] == "warning"]
|
|
if missing:
|
|
feasibility = "needs_clarification"
|
|
status = "needs_clarification"
|
|
feasible: bool | None = None
|
|
elif failed:
|
|
feasibility = "not_feasible"
|
|
status = "success"
|
|
feasible = False
|
|
elif warnings:
|
|
feasibility = "feasible_with_warnings"
|
|
status = "success"
|
|
feasible = True
|
|
else:
|
|
feasibility = "feasible"
|
|
status = "success"
|
|
feasible = True
|
|
return {
|
|
"status": status,
|
|
"machine": resolved["machine"],
|
|
"resolution": {"matched_by": resolved["matched_by"], "matched_value": resolved["matched_value"]},
|
|
"feasibility": feasibility,
|
|
"feasible": feasible,
|
|
"weight_scope": "material_only",
|
|
"checks": checks,
|
|
"passed_constraints": [check["constraint"] for check in checks if check["state"] == "passed"],
|
|
"warning_constraints": warnings,
|
|
"failed_constraints": failed,
|
|
"missing_required_inputs": missing,
|
|
}
|
|
|
|
|
|
def _machine_max_input_error(field: str, value: Any, *, allow_zero: bool = False) -> dict[str, str] | None:
|
|
if isinstance(value, bool) or not isinstance(value, (int, float)):
|
|
return {"field": field, "message": "must be a number"}
|
|
if not math.isfinite(float(value)) or (float(value) < 0 if allow_zero else float(value) <= 0):
|
|
return {
|
|
"field": field,
|
|
"message": "must be zero or greater" if allow_zero else "must be greater than zero",
|
|
}
|
|
return None
|
|
|
|
|
|
def _maximum_limit_state(value: NumericRange | None) -> str:
|
|
if value is None or value.maximum is None:
|
|
return "not_configured"
|
|
return "configured"
|
|
|
|
|
|
def _governing_maximum(
|
|
candidates: tuple[tuple[str, float], ...],
|
|
) -> tuple[float, list[str]]:
|
|
"""Choose all co-governing limits with a fixed, calculation-only tolerance."""
|
|
maximum = min(value for _, value in candidates)
|
|
governing = [
|
|
name
|
|
for name, value in candidates
|
|
if math.isclose(
|
|
value,
|
|
maximum,
|
|
rel_tol=MAX_LENGTH_TIE_REL_TOLERANCE,
|
|
abs_tol=MAX_LENGTH_TIE_ABS_TOLERANCE_M,
|
|
)
|
|
]
|
|
return maximum, governing
|
|
|
|
|
|
def _verification_boundary_value(value: float | None, limits: NumericRange | None) -> float | None:
|
|
"""Normalize only inverse-calculation floating-point noise at a hard bound."""
|
|
if value is None or limits is None:
|
|
return value
|
|
if (
|
|
limits.minimum is not None
|
|
and value < limits.minimum
|
|
and math.isclose(
|
|
value,
|
|
limits.minimum,
|
|
rel_tol=MAX_LENGTH_TIE_REL_TOLERANCE,
|
|
abs_tol=MAX_LENGTH_TIE_ABS_TOLERANCE_M,
|
|
)
|
|
):
|
|
return limits.minimum
|
|
if (
|
|
limits.maximum is not None
|
|
and value > limits.maximum
|
|
and math.isclose(
|
|
value,
|
|
limits.maximum,
|
|
rel_tol=MAX_LENGTH_TIE_REL_TOLERANCE,
|
|
abs_tol=MAX_LENGTH_TIE_ABS_TOLERANCE_M,
|
|
)
|
|
):
|
|
return limits.maximum
|
|
return value
|
|
|
|
|
|
def calculate_machine_max_product_length(
|
|
*,
|
|
machine: str,
|
|
core_diameter_mm: Any,
|
|
thickness_mm: Any,
|
|
thickness_stddev_mm: Any = None,
|
|
product_width_m: Any = None,
|
|
area_weight_g_m2: Any = None,
|
|
repository: MachineRepository | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Calculate deterministic machine-limited product lengths from shared domains.
|
|
|
|
Diameter limits are delegated to ``calculate_product_length`` and material
|
|
weight limits to ``calculate_material_length_for_weight``. This function
|
|
only composes configured machine constraints; it contains no roll formula.
|
|
"""
|
|
resolved = (repository or MachineRepository.load()).resolve(machine)
|
|
if resolved["status"] != "resolved":
|
|
return {key: value for key, value in resolved.items() if key != "resolved_machine"}
|
|
|
|
invalid = []
|
|
for field, value in (("core_diameter_mm", core_diameter_mm), ("thickness_mm", thickness_mm)):
|
|
error = _machine_max_input_error(field, value)
|
|
if error:
|
|
invalid.append(error)
|
|
if thickness_stddev_mm is None:
|
|
thickness_stddev = 0.0
|
|
else:
|
|
error = _machine_max_input_error(
|
|
"thickness_stddev_mm", thickness_stddev_mm, allow_zero=True
|
|
)
|
|
if error:
|
|
invalid.append(error)
|
|
thickness_stddev = 0.0
|
|
else:
|
|
thickness_stddev = float(thickness_stddev_mm)
|
|
for field, value in (("product_width_m", product_width_m), ("area_weight_g_m2", area_weight_g_m2)):
|
|
if value is not None:
|
|
error = _machine_max_input_error(field, value)
|
|
if error:
|
|
invalid.append(error)
|
|
if invalid:
|
|
return {"status": "invalid_parameter", "invalid": invalid}
|
|
|
|
core = float(core_diameter_mm)
|
|
thickness = float(thickness_mm)
|
|
width = float(product_width_m) if product_width_m is not None else None
|
|
area_weight = float(area_weight_g_m2) if area_weight_g_m2 is not None else None
|
|
machine_model: Machine = resolved["resolved_machine"]
|
|
constraints = machine_model.constraints
|
|
compatibility_checks = [
|
|
_core_check(core, constraints.allowed_core_diameters_mm),
|
|
_range_check("product_width_m", width, constraints.product_width_m, "m"),
|
|
]
|
|
base = {
|
|
"machine": resolved["machine"],
|
|
"resolution": {
|
|
"matched_by": resolved["matched_by"],
|
|
"matched_value": resolved["matched_value"],
|
|
},
|
|
"weight_scope": "material_only",
|
|
"inputs": {
|
|
"core_diameter_mm": core,
|
|
"thickness_mm": thickness,
|
|
"thickness_stddev_mm": thickness_stddev,
|
|
"product_width_m": width,
|
|
"area_weight_g_m2": area_weight,
|
|
},
|
|
"compatibility_checks": compatibility_checks,
|
|
"units": {"length_m": "m", "diameter_mm": "mm", "weight_kg": "kg"},
|
|
}
|
|
failed_compatibility = [
|
|
check["constraint"] for check in compatibility_checks if check["state"] == "failed"
|
|
]
|
|
if failed_compatibility:
|
|
return {
|
|
"status": "incompatible",
|
|
**base,
|
|
"failed_compatibility_constraints": failed_compatibility,
|
|
"production_maximums": None,
|
|
"missing_required_inputs": [],
|
|
}
|
|
|
|
missing = []
|
|
if constraints.product_width_m is not None and width is None:
|
|
missing.append("product_width_m")
|
|
if constraints.roll_weight_kg is not None:
|
|
if width is None and "product_width_m" not in missing:
|
|
missing.append("product_width_m")
|
|
if area_weight is None:
|
|
missing.append("area_weight_g_m2")
|
|
|
|
diameter_limits: dict[str, float | None] = {
|
|
"nominal_diameter_length_m": None,
|
|
"conservative_diameter_length_m": None,
|
|
}
|
|
if constraints.roll_diameter_mm is not None and constraints.roll_diameter_mm.maximum is not None:
|
|
diameter_result = calculate_product_length(
|
|
target_roll_diameter_mm=constraints.roll_diameter_mm.maximum,
|
|
core_diameter_mm=core,
|
|
thickness_mm=thickness,
|
|
thickness_stddev_mm=thickness_stddev,
|
|
)
|
|
if diameter_result["status"] != "success":
|
|
return {**base, **diameter_result, "production_maximums": None}
|
|
calculation = diameter_result["calculation"]
|
|
diameter_limits = {
|
|
"nominal_diameter_length_m": calculation["average_product_length_m"],
|
|
"conservative_diameter_length_m": calculation["minimum_product_length_m"],
|
|
}
|
|
|
|
weight_limit: float | None = None
|
|
if constraints.roll_weight_kg is not None and constraints.roll_weight_kg.maximum is not None and not missing:
|
|
weight_result = calculate_material_length_for_weight(
|
|
target_material_weight_kg=constraints.roll_weight_kg.maximum,
|
|
width_m=width,
|
|
area_weight_g_m2=area_weight,
|
|
)
|
|
if weight_result["status"] != "success":
|
|
return {**base, **weight_result, "production_maximums": None}
|
|
weight_limit = weight_result["material_length_m"]
|
|
|
|
limits = {
|
|
**diameter_limits,
|
|
"material_weight_length_m": weight_limit,
|
|
"roll_diameter_maximum_state": _maximum_limit_state(constraints.roll_diameter_mm),
|
|
"material_weight_maximum_state": _maximum_limit_state(constraints.roll_weight_kg),
|
|
}
|
|
if missing:
|
|
partial_limits = {
|
|
key: value for key, value in diameter_limits.items() if value is not None
|
|
}
|
|
return {
|
|
"status": "needs_clarification",
|
|
**base,
|
|
"limits": limits,
|
|
"partial_limits": partial_limits or None,
|
|
"production_maximums": None,
|
|
"missing_required_inputs": missing,
|
|
}
|
|
|
|
nominal_candidates: list[tuple[str, float]] = []
|
|
conservative_candidates: list[tuple[str, float]] = []
|
|
if diameter_limits["nominal_diameter_length_m"] is not None:
|
|
nominal_candidates.append(("roll_diameter_nominal", diameter_limits["nominal_diameter_length_m"]))
|
|
conservative_candidates.append(("roll_diameter_conservative", diameter_limits["conservative_diameter_length_m"]))
|
|
if weight_limit is not None:
|
|
nominal_candidates.append(("material_weight", weight_limit))
|
|
conservative_candidates.append(("material_weight", weight_limit))
|
|
if not nominal_candidates:
|
|
return {
|
|
"status": "no_maximum_length_constraint",
|
|
**base,
|
|
"limits": limits,
|
|
"production_maximums": None,
|
|
"missing_required_inputs": [],
|
|
}
|
|
|
|
nominal_length, nominal_governing = _governing_maximum(tuple(nominal_candidates))
|
|
conservative_length, conservative_governing = _governing_maximum(
|
|
tuple(conservative_candidates)
|
|
)
|
|
production_maximums = {
|
|
"nominal_length_m": nominal_length,
|
|
"nominal_governing_constraints": nominal_governing,
|
|
"conservative_length_m": conservative_length,
|
|
"conservative_governing_constraints": conservative_governing,
|
|
}
|
|
|
|
material_weights = None
|
|
if width is not None and area_weight is not None:
|
|
nominal_weight = calculate_material_weight(
|
|
roll_length_m=nominal_length, width_m=width, area_weight_g_m2=area_weight
|
|
)
|
|
conservative_weight = calculate_material_weight(
|
|
roll_length_m=conservative_length, width_m=width, area_weight_g_m2=area_weight
|
|
)
|
|
material_weights = {
|
|
"at_nominal_production_maximum": nominal_weight["material_weight_kg"],
|
|
"at_conservative_production_maximum": conservative_weight["material_weight_kg"],
|
|
}
|
|
|
|
final_feasibility = {}
|
|
for label, length in (("nominal", nominal_length), ("conservative", conservative_length)):
|
|
roll_result = calculate_roll({
|
|
"roll_length_m": length,
|
|
"core_diameter_mm": core,
|
|
"thickness_mm": thickness,
|
|
"thickness_stddev_mm": thickness_stddev,
|
|
"width_m": width,
|
|
"area_weight_g_m2": area_weight,
|
|
"include_roll_weight": constraints.roll_weight_kg is not None,
|
|
})
|
|
if roll_result["status"] != "success":
|
|
return {
|
|
"status": "incompatible",
|
|
**base,
|
|
"limits": limits,
|
|
"production_maximums": None,
|
|
"final_feasibility": {label: roll_result},
|
|
"missing_required_inputs": [],
|
|
}
|
|
calculated = roll_result["calculation"]
|
|
feasibility = check_production_feasibility({
|
|
"machine": machine_model.id,
|
|
"roll_weight_kg": _verification_boundary_value(
|
|
calculated["roll_weight_kg"], constraints.roll_weight_kg
|
|
),
|
|
"average_diameter_mm": _verification_boundary_value(
|
|
calculated["average_diameter_mm"], constraints.roll_diameter_mm
|
|
),
|
|
"maximum_diameter_mm": _verification_boundary_value(
|
|
calculated["maximum_diameter_mm"], constraints.roll_diameter_mm
|
|
),
|
|
"core_diameter_mm": core,
|
|
"product_width_m": width,
|
|
}, repository=repository)
|
|
final_feasibility[label] = feasibility
|
|
if feasibility.get("feasible") is not True:
|
|
return {
|
|
"status": "incompatible",
|
|
**base,
|
|
"limits": limits,
|
|
"production_maximums": None,
|
|
"final_feasibility": final_feasibility,
|
|
"missing_required_inputs": [],
|
|
}
|
|
return {
|
|
"status": "success",
|
|
**base,
|
|
"limits": limits,
|
|
"production_maximums": production_maximums,
|
|
"material_weights_kg": material_weights,
|
|
"final_feasibility": final_feasibility,
|
|
"warnings": {
|
|
label: value["warning_constraints"]
|
|
for label, value in final_feasibility.items()
|
|
if value["warning_constraints"]
|
|
},
|
|
"missing_required_inputs": [],
|
|
}
|