From 3e13d2440552cf4795f9a9ed4339782205d83e3e Mon Sep 17 00:00:00 2001 From: Martin Date: Sat, 26 Sep 2026 10:47:53 +0200 Subject: [PATCH] Add configurable production machine constraints --- README.md | 28 +- app.py | 14 + config/machines.yaml | 153 +++++++++ data/source/LineLimitations_RollCalc.csv | 16 + docs/architecture.md | 17 + machine_constraints.py | 402 +++++++++++++++++++++++ mcp_server.py | 28 ++ requirements.txt | 1 + rollcalc_mcp_tools.py | 28 ++ tests/test_calculation_api.py | 19 ++ tests/test_machine_constraints.py | 165 ++++++++++ tests/test_mcp_adapter.py | 36 ++ 12 files changed, 904 insertions(+), 3 deletions(-) create mode 100644 config/machines.yaml create mode 100644 data/source/LineLimitations_RollCalc.csv create mode 100644 machine_constraints.py create mode 100644 tests/test_machine_constraints.py diff --git a/README.md b/README.md index 6090191..a27ebac 100644 --- a/README.md +++ b/README.md @@ -107,6 +107,7 @@ Implemented routes: | `/api/calculations/roll` | `POST` | Basic Auth | Resolves, validates, and executes a direct roll-diameter request. | | `/api/calculations/roll/modify` | `POST` | Basic Auth | Applies explicit changes to structured calculation state and recalculates. | | `/api/calculations/transport` | `POST` | Basic Auth | Executes the shared transport-capacity analysis. | +| `/api/calculations/production-feasibility` | `POST` | Basic Auth | Checks configured production-machine constraints. | | `/api/reports/roll-calculation.pdf` | `POST` | Basic Auth | Recalculates a request authoritatively and returns a one-page PDF. | | `/api/conversations` | `POST` | Basic Auth | Creates an in-memory conversation session. | | `/api/conversations//messages` | `POST` | Basic Auth | Interprets one utterance and executes the deterministic workflow. | @@ -119,11 +120,11 @@ Authentication is implemented with `Flask-HTTPAuth`. The current code checks `BE `mcp_server.py` provides a local stdio-only Model Context Protocol server. It is a thin adapter over the same domain services used by Flask: article lookup, direct roll calculation, product-length calculation, material-weight calculation, -and transport analysis. +transport analysis, and configured production-machine feasibility. The available PoC tools are `get_article`, `search_articles`, `calculate_material_weight`, `calculate_product_length`, -`calculate_roll_diameter`, and -`analyze_transport_capacity`. +`calculate_roll_diameter`, `analyze_transport_capacity`, `get_machine`, and +`check_production_feasibility`. Install the pinned dependencies, including `mcp==1.26.0`, in the existing environment, then start it with: @@ -165,6 +166,16 @@ Tool calling semantics: `transport_roll_inputs` unchanged. The supplied `roll_weight_kg` is passed through unchanged. Callers may use a canonical `transport_preset` or provide custom transport dimensions and `max_weight_kg`. +- `get_machine` resolves a configured machine by stable ID, display name, or + configured alias. Locations are descriptive metadata only. +- `check_production_feasibility` evaluates all configured independent machine + limits deterministically. Its V1 weight is material-only: it never estimates + core, packaging, or gross weight. Callers pass `average_diameter_mm` and + `maximum_diameter_mm` from `calculate_roll_diameter`; a nominal fit with a + maximum-diameter exceedance is `feasible_with_warnings`, while a nominal + exceedance is `not_feasible`. Missing configured roll inputs produce + `needs_clarification`; line speed is retained in configuration but outside + the initial roll-feasibility scope. Generate a password hash or user entry with: @@ -516,6 +527,17 @@ Primary product/article dataset used by the active calculator UI. JSON configuration for forklift/heavy-roll rules. The active template also contains an inline `window.APP_CONFIG`, so developers should verify which config source is authoritative before changing rule behavior. +### `config/machines.yaml` + +Runtime configuration for production-machine constraints, loaded with +`yaml.safe_load` and strict schema validation by `machine_constraints.py`. +It has `schema_version: 1`, stable IDs, names, aliases, locations, and explicit +unit-bearing constraint keys. `null` line-speed constraints mean unknown/not +configured; they do not mean zero or unlimited capability. The CSV in +`data/source/LineLimitations_RollCalc.csv` remains the source/reference +document. The browser currently has no production-machine control; the shared +Flask API and MCP tools expose the capability for later UI work. + ### `access_log.json` JSON audit log written by `app.py`. diff --git a/app.py b/app.py index 3fb27af..53c73db 100644 --- a/app.py +++ b/app.py @@ -20,6 +20,7 @@ from flask_httpauth import HTTPBasicAuth from werkzeug.security import check_password_hash from core_presets import CORE_PRESETS +from machine_constraints import check_production_feasibility from conversation_service import ( ConversationNotFoundError, ConversationReportNotFoundError, @@ -264,6 +265,19 @@ def create_transport_calculation(): return jsonify(result), 200 +@app.route("/api/calculations/production-feasibility", methods=["POST"]) +@auth.login_required +def create_production_feasibility_calculation(): + """Check configured production-machine constraints through the shared domain.""" + result = check_production_feasibility(request.get_json(silent=True)) + log_access( + auth.current_user(), + "/api/calculations/production-feasibility", + "POST", + ) + return jsonify(result), 200 + + def _create_pdf(calculation_request): result = calculate_roll(calculation_request, build_info=BUILD_INFO) report = report_from_calculation_result(result) diff --git a/config/machines.yaml b/config/machines.yaml new file mode 100644 index 0000000..a737073 --- /dev/null +++ b/config/machines.yaml @@ -0,0 +1,153 @@ +schema_version: 1 + +machines: + - id: bento_1 + name: Bento 1 + aliases: [Bento1, B1] + location: Espelkamp (Germany) + constraints: + roll_weight_kg: {maximum: 2500} + roll_diameter_mm: {minimum: 300, maximum: 876} + allowed_core_diameters_mm: [150, 168, 170, 200, 235] + product_width_m: {minimum: 4.7, maximum: 5.0} + line_speed_m_per_min: {minimum: 0.9, maximum: 6.0} + - id: bento_2 + name: Bento 2 + aliases: [Bento2, B2] + location: Selangor (Malaysia) + constraints: + roll_weight_kg: {maximum: 2500} + roll_diameter_mm: {minimum: 250, maximum: 1000} + allowed_core_diameters_mm: [150, 168, 170, 200, 235] + product_width_m: {minimum: 4.7, maximum: 5.0} + line_speed_m_per_min: {minimum: 0.9, maximum: 6.0} + - id: k1 + name: K1 + aliases: [K 1] + location: Selangor (Malaysia) + constraints: + roll_weight_kg: {maximum: 800} + roll_diameter_mm: {minimum: 200, maximum: 1100} + allowed_core_diameters_mm: [130, 133, 150, 168, 170] + product_width_m: {minimum: 4.0, maximum: 6.1} + line_speed_m_per_min: {minimum: 1.0, maximum: 20.0} + - id: k_3_4 + name: K3/4 + aliases: [K 34, K 3/4] + location: Espelkamp (Germany) + constraints: + roll_weight_kg: {maximum: 800} + roll_diameter_mm: {minimum: 200, maximum: 1100} + allowed_core_diameters_mm: [130, 133, 150, 168, 170] + product_width_m: {minimum: 4.0, maximum: 6.1} + line_speed_m_per_min: {minimum: 1.0, maximum: 15.0} + - id: k5 + name: K5 + aliases: [K 5] + location: Adorf (Germany) + constraints: + roll_weight_kg: {maximum: 800} + roll_diameter_mm: {minimum: 200, maximum: 1200} + allowed_core_diameters_mm: [130, 133, 150, 168, 170] + product_width_m: {minimum: 4.0, maximum: 6.1} + line_speed_m_per_min: {minimum: 1.0, maximum: 20.0} + - id: k6 + name: K6 + aliases: [K 6] + location: Selangor (Malaysia) + constraints: + roll_weight_kg: {maximum: 800} + roll_diameter_mm: {minimum: 200, maximum: 1100} + allowed_core_diameters_mm: [130, 133, 150, 168, 170] + product_width_m: {minimum: 4.0, maximum: 6.1} + line_speed_m_per_min: {minimum: 1.0, maximum: 20.0} + - id: k7 + name: K7 + aliases: [K 7] + location: Espelkamp (Germany) + constraints: + roll_weight_kg: {maximum: 700} + roll_diameter_mm: {minimum: 300, maximum: 1100} + allowed_core_diameters_mm: [130, 133, 150, 168, 170] + product_width_m: {minimum: 4.0, maximum: 6.1} + line_speed_m_per_min: {minimum: 0.6, maximum: 40.0} + - id: e1 + name: E1 + aliases: [E 1] + location: Espelkamp (Germany) + constraints: + roll_weight_kg: {maximum: 2500} + roll_diameter_mm: {minimum: 200, maximum: 1000} + allowed_core_diameters_mm: [150, 168, 170] + product_width_m: {minimum: 7.5, maximum: 7.5} + line_speed_m_per_min: {minimum: 0.6, maximum: 10.0} + - id: e4 + name: E4 + aliases: [E 4] + location: Espelkamp (Germany) + constraints: + roll_weight_kg: {maximum: 2000} + roll_diameter_mm: {minimum: 200, maximum: 900} + allowed_core_diameters_mm: [150, 168, 170] + product_width_m: {minimum: 5.1, maximum: 5.1} + line_speed_m_per_min: {minimum: 0.5, maximum: 10.0} + - id: gitter_1 + name: Gitter 1 + aliases: [G1, G 1] + location: Adorf (Germany) + constraints: + roll_weight_kg: {maximum: 400} + roll_diameter_mm: {minimum: 200, maximum: 600} + allowed_core_diameters_mm: [140] + product_width_m: {minimum: 4.0, maximum: 4.75} + line_speed_m_per_min: null + - id: gitter_2 + name: Gitter 2 + aliases: [G2, G 2] + location: Adorf (Germany) + constraints: + roll_weight_kg: {maximum: 400} + roll_diameter_mm: {minimum: 200, maximum: 600} + allowed_core_diameters_mm: [140] + product_width_m: {minimum: 4.0, maximum: 4.75} + line_speed_m_per_min: null + - id: gitter_3 + name: Gitter 3 + aliases: [G3, G 3] + location: Adorf (Germany) + constraints: + roll_weight_kg: {maximum: 400} + roll_diameter_mm: {minimum: 200, maximum: 600} + allowed_core_diameters_mm: [140] + product_width_m: {minimum: 4.0, maximum: 4.75} + line_speed_m_per_min: null + - id: gitter_4 + name: Gitter 4 + aliases: [G4, G 4] + location: Adorf (Germany) + constraints: + roll_weight_kg: {maximum: 400} + roll_diameter_mm: {minimum: 200, maximum: 600} + allowed_core_diameters_mm: [140] + product_width_m: {minimum: 4.0, maximum: 4.75} + line_speed_m_per_min: null + - id: secudrain_1_2 + name: Secudrain 1/2 + aliases: [Sec 1/2, Secudrain 12, Sec 12] + location: Adorf (Germany) + constraints: + roll_weight_kg: {maximum: 400} + roll_diameter_mm: {minimum: 200, maximum: 800} + allowed_core_diameters_mm: [120] + product_width_m: {minimum: 2.0, maximum: 4.0} + line_speed_m_per_min: null + - id: secudrain_3 + name: Secudrain 3 + aliases: [Sec 3] + location: Adorf (Germany) + constraints: + roll_weight_kg: {maximum: 400} + roll_diameter_mm: {minimum: 200, maximum: 800} + allowed_core_diameters_mm: [120] + product_width_m: {minimum: 1.8, maximum: 2.1} + line_speed_m_per_min: null diff --git a/data/source/LineLimitations_RollCalc.csv b/data/source/LineLimitations_RollCalc.csv new file mode 100644 index 0000000..66adc02 --- /dev/null +++ b/data/source/LineLimitations_RollCalc.csv @@ -0,0 +1,16 @@ +Production line;Aliasses;Max. Rollweight [kg];Max Rolldiameter [mm];Min Rolldiameter [mm];allowed core diameters [mm];Max linespeed [m/min];Min linespeed [m/min];max. Product Width;min. Product Width;Location +Bento 1;Bento1, B1;2501;876;300;150, 170, 200, 235;6;0,9;5;4,7;Espelkamp (Germany) +Bento 2;Bento2, B2;2501;1000;250;150, 170, 200, 235;6;0,9;5;4,7;Selangor (Malaysia) +K1;K 1;800;1100;200;130, 150, 170;20;1;6,1;4;Selangor (Malaysia) +K3/4;K 34, K 3/4;800;1100;200;130, 150, 170;15;1;6,1;4;Espelkamp (Germany) +K5;K 5;800;1200;200;130, 150, 170;20;1;6,1;4;Adorf (Germany) +K6;K 6;800;1100;200;130, 150, 170;20;1;6,1;4;Selangor (Malaysia) +K7;K 7;700;1100;300;130, 150, 170;40;0,6;6,1;4;Espelkamp (Germany) +E1;E 1;2500;1000;200;150, 170;10;0,6;7,5;7,5;Espelkamp (Germany) +E4;E 4;2000;900;200;150, 170;10;0,5;5,1;5,1;Espelkamp (Germany) +Gitter 1;G1, G 1;400;600;200;140;;;4,75;4;Adorf (Germany) +Gitter 2;G2, G 2;400;600;200;140;;;4,75;4;Adorf (Germany) +Gitter 3;G3, G 3;400;600;200;140;;;4,75;4;Adorf (Germany) +Gitter 4;G4, G 4;400;600;200;140;;;4,75;4;Adorf (Germany) +Secudrain 1/2;Sec 1/2, Secudrain 12, Sec 12;400;800;200;120;;;4;2;Adorf (Germany) + Secudrain 3;Sec 3;400;800;200;120;;;2,1;1,8;Adorf (Germany) \ No newline at end of file diff --git a/docs/architecture.md b/docs/architecture.md index f898999..20029d0 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -31,6 +31,23 @@ roll_calculation.calculate_roll() The PDF endpoint accepts the same calculation request, recalculates it through the core, then adapts the authoritative result to the established one-page PDF layout. It does not accept calculated values from the browser. +## Production-Machine Feasibility + +`config/machines.yaml` is the runtime source for configured production-machine +limits; `data/source/LineLimitations_RollCalc.csv` is retained as the reference +source document. `machine_constraints.py` loads and validates the YAML through +`MachineRepository`, resolves only stable IDs/names/configured aliases, and +evaluates independent constraints through `check_production_feasibility()`. + +V1 compares material-only roll weight, nominal/average diameter, calculated +maximum diameter, core diameter, and product width. All configured V1 inputs +are required before a feasible result may be claimed. Nominal diameter is used +for a configured minimum; an average diameter over the maximum fails, whereas +an average fit whose calculated maximum exceeds the maximum is a retained +warning. Line speed remains configured but is outside the initial roll-only +scope. Flask and MCP are thin adapters; no machine constraints are duplicated +in those layers or in browser code. + ## Constrained Natural-Language Flow ```text diff --git a/machine_constraints.py b/machine_constraints.py new file mode 100644 index 0000000..e624f50 --- /dev/null +++ b/machine_constraints.py @@ -0,0 +1,402 @@ +"""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 + + +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", +} + + +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, + } diff --git a/mcp_server.py b/mcp_server.py index f2ada8b..97f3ca3 100644 --- a/mcp_server.py +++ b/mcp_server.py @@ -10,6 +10,8 @@ from rollcalc_mcp_tools import ( calculate_product_length_result, calculate_roll_diameter_result, get_article_result, + get_machine_result, + check_production_feasibility_result, search_articles_result, ) from transport_calculation import TransportPresetKey @@ -34,6 +36,32 @@ def create_server() -> Any: """Use search_articles when the user identifies a RollCalc product by name, family, designation, or descriptive article text instead of an exact article number. Do not guess or invent an article number: results are deterministic retrieval candidates from the RollCalc article master data, not product recommendations. For an exact article number, use get_article -> calculate_roll_diameter -> optionally analyze_transport_capacity. For a product name, use search_articles first. Only when total_matches is exactly 1 may you use that candidate's exact article_number and continue through get_article and the requested workflow. When multiple plausible candidates are returned, do NOT select the first or highest-ranked candidate, do not treat ranking as permission to choose, and do not calculate yet: present relevant candidates with their article numbers, distinguishing names, and production_site when supplied; ask the user which article number is intended, then stop until they select it. Ranking/order expresses retrieval relevance only and never authorizes automatic selection. If no candidate is returned, say that no matching RollCalc article was found, ask for a more specific designation or article number, and do not invent an article. production_site=Malaysia is descriptive metadata only: it does not imply Bentofix, Bento 2, or any production machine.""" return search_articles_result(query) + @server.tool() + def get_machine(machine: str) -> dict[str, Any]: + """Resolve a configured production machine by stable ID, exact name, or configured alias. Machine limits are deterministic configured domain data; aliases and locations are lookup/metadata only and do not imply capabilities.""" + return get_machine_result(machine) + + @server.tool() + def check_production_feasibility( + 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, + ) -> dict[str, Any]: + """Deterministically check a material-only calculated roll against one configured machine. The tool, not the LLM, decides every independent limit; never override failed constraints. Report returned warnings, especially maximum-diameter variation warnings. Missing configured V1 roll inputs require clarification rather than assumptions. average_diameter_mm and maximum_diameter_mm are the existing RollCalc calculation outputs; line speed is retained but outside the initial roll-feasibility scope.""" + return check_production_feasibility_result( + machine=machine, + roll_weight_kg=roll_weight_kg, + average_diameter_mm=average_diameter_mm, + maximum_diameter_mm=maximum_diameter_mm, + core_diameter_mm=core_diameter_mm, + product_width_m=product_width_m, + line_speed_m_per_min=line_speed_m_per_min, + ) + @server.tool() def calculate_material_weight( roll_length_m: float, diff --git a/requirements.txt b/requirements.txt index d2b106c..716f8d5 100644 --- a/requirements.txt +++ b/requirements.txt @@ -6,3 +6,4 @@ itsdangerous==2.1.2 Jinja2==3.1.2 MarkupSafe==2.1.3 mcp==1.26.0 +PyYAML==6.0.2 diff --git a/rollcalc_mcp_tools.py b/rollcalc_mcp_tools.py index 5f3d344..64ad28a 100644 --- a/rollcalc_mcp_tools.py +++ b/rollcalc_mcp_tools.py @@ -4,6 +4,7 @@ from __future__ import annotations from typing import Any +from machine_constraints import check_production_feasibility, get_machine from roll_calculation import ( ArticleRepository, calculate_material_weight, @@ -28,6 +29,33 @@ def search_articles_result(query: str) -> dict[str, Any]: return ArticleRepository.load().search(query) +def get_machine_result(machine: str | None = None) -> dict[str, Any]: + """Delegate configured-machine resolution to the shared repository.""" + return get_machine(machine) + + +def check_production_feasibility_result( + *, + 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, +) -> dict[str, Any]: + """Delegate configured production feasibility to the shared domain.""" + return check_production_feasibility({ + "machine": machine, + "roll_weight_kg": roll_weight_kg, + "average_diameter_mm": average_diameter_mm, + "maximum_diameter_mm": maximum_diameter_mm, + "core_diameter_mm": core_diameter_mm, + "product_width_m": product_width_m, + "line_speed_m_per_min": line_speed_m_per_min, + }) + + def calculate_material_weight_result( *, roll_length_m: float, diff --git a/tests/test_calculation_api.py b/tests/test_calculation_api.py index 5e5051d..2f5518f 100644 --- a/tests/test_calculation_api.py +++ b/tests/test_calculation_api.py @@ -120,6 +120,25 @@ class CalculationApiTests(unittest.TestCase): self.assertEqual(response.status_code, 200) self.assertEqual(response.get_json()["status"], "invalid_parameter") + def test_production_feasibility_endpoint_uses_shared_domain(self): + response = self.client.post( + "/api/calculations/production-feasibility", + json={ + "machine": "B1", + "roll_weight_kg": 2500, + "average_diameter_mm": 876, + "maximum_diameter_mm": 877, + "core_diameter_mm": 168, + "product_width_m": 5, + }, + headers=self.headers, + ) + result = response.get_json() + + self.assertEqual(response.status_code, 200) + self.assertEqual(result["feasibility"], "feasible_with_warnings") + self.assertEqual(result["warning_constraints"], ["roll_diameter_mm"]) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_machine_constraints.py b/tests/test_machine_constraints.py new file mode 100644 index 0000000..5bfb059 --- /dev/null +++ b/tests/test_machine_constraints.py @@ -0,0 +1,165 @@ +import tempfile +import unittest +from pathlib import Path + +from machine_constraints import ( + MACHINE_CONFIG_FILE, + MachineConfigurationError, + MachineRepository, + check_production_feasibility, + get_machine, +) + + +def request(**changes): + payload = { + "machine": "bento_1", + "roll_weight_kg": 2500.0, + "average_diameter_mm": 800.0, + "maximum_diameter_mm": 850.0, + "core_diameter_mm": 168.0, + "product_width_m": 4.8, + } + payload.update(changes) + return payload + + +class MachineRepositoryTests(unittest.TestCase): + def setUp(self): + self.repository = MachineRepository.load() + + def test_yaml_load_represents_all_source_machines_and_bento_transformation(self): + self.assertTrue(MACHINE_CONFIG_FILE.exists()) + self.assertEqual(len(self.repository.machines), 15) + self.assertEqual( + {machine.name for machine in self.repository.machines}, + { + "Bento 1", "Bento 2", "K1", "K3/4", "K5", "K6", "K7", + "E1", "E4", "Gitter 1", "Gitter 2", "Gitter 3", "Gitter 4", + "Secudrain 1/2", "Secudrain 3", + }, + ) + for identifier in ("bento_1", "bento_2"): + machine = self.repository.resolve(identifier)["resolved_machine"] + self.assertEqual(machine.constraints.roll_weight_kg.maximum, 2500.0) + + def test_blank_source_line_speed_is_explicitly_not_configured(self): + machine = self.repository.resolve("Gitter 1")["machine"] + self.assertIsNone(machine["constraints"]["line_speed_m_per_min"]) + + def test_lookup_uses_ids_names_aliases_and_harmless_normalization(self): + for value, matched_by in ( + ("bento_1", "id"), (" BENTO 1 ", "name"), (" b1 ", "alias"), + ): + with self.subTest(value=value): + result = self.repository.resolve(value) + self.assertEqual(result["status"], "resolved") + self.assertEqual(result["matched_by"], matched_by) + self.assertEqual(result["machine"]["id"], "bento_1") + self.assertEqual(self.repository.resolve("bento") ["status"], "machine_not_found") + + def test_get_machine_returns_public_configured_data(self): + result = get_machine("K 3/4", repository=self.repository) + self.assertEqual(result["status"], "resolved") + self.assertNotIn("resolved_machine", result) + self.assertEqual(result["machine"]["id"], "k_3_4") + + def config_error(self, body): + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "machines.yaml" + path.write_text(body, encoding="utf-8") + with self.assertRaises(MachineConfigurationError): + MachineRepository.load(path) + + def test_configuration_validation_rejects_malformed_and_unsupported_versions(self): + self.config_error("machines: [") + self.config_error("schema_version: 2\nmachines: []\n") + self.config_error("schema_version: 1\nunknown: true\nmachines: []\n") + + def test_configuration_validation_rejects_invalid_machine_data(self): + valid = """ +schema_version: 1 +machines: + - id: alpha + name: Alpha + aliases: [A] + location: Here + constraints: + roll_weight_kg: {maximum: 3} + roll_diameter_mm: {minimum: 1, maximum: 2} + allowed_core_diameters_mm: [150] + product_width_m: {minimum: 1, maximum: 2} + line_speed_m_per_min: null +""" + self.config_error(valid.replace("minimum: 1, maximum: 2", "minimum: 2, maximum: 1")) + self.config_error(valid.replace("[150]", "[0]")) + self.config_error(valid.replace("aliases: [A]", "aliases: bad")) + self.config_error(valid.replace("line_speed_m_per_min: null", "unsupported: true")) + self.config_error(valid.replace("[150]", "[not-a-number]")) + duplicate_row = valid.split(" - id: alpha", 1)[1].replace( + "name: Alpha", "name: Beta", 1 + ) + duplicate = valid + "\n - id: beta" + duplicate_row + self.config_error(duplicate) + + +class MachineFeasibilityTests(unittest.TestCase): + def check(self, **changes): + return check_production_feasibility(request(**changes)) + + def test_inclusive_weight_and_range_boundaries(self): + self.assertEqual(self.check()["feasibility"], "feasible") + self.assertEqual(self.check(roll_weight_kg=2500.0001)["feasibility"], "not_feasible") + self.assertEqual(self.check(average_diameter_mm=300, maximum_diameter_mm=300)["feasibility"], "feasible") + self.assertEqual(self.check(average_diameter_mm=876, maximum_diameter_mm=876)["feasibility"], "feasible") + self.assertEqual(self.check(average_diameter_mm=299, maximum_diameter_mm=299)["feasibility"], "not_feasible") + + def test_exact_width_machines_do_not_have_an_implicit_tolerance(self): + exact = self.check(machine="e1", product_width_m=7.5, average_diameter_mm=900, maximum_diameter_mm=900) + low = self.check(machine="e1", product_width_m=7.4999, average_diameter_mm=900, maximum_diameter_mm=900) + high = self.check(machine="e1", product_width_m=7.5001, average_diameter_mm=900, maximum_diameter_mm=900) + self.assertEqual(exact["feasibility"], "feasible") + self.assertIn("product_width_m", low["failed_constraints"]) + self.assertIn("product_width_m", high["failed_constraints"]) + + def test_configured_core_equivalents_are_exact_membership_only(self): + for core in (130, 133, 150, 168, 170): + with self.subTest(core=core): + self.assertEqual(self.check(machine="k1", roll_weight_kg=800, core_diameter_mm=core)["feasibility"], "feasible") + failed = self.check(machine="k1", roll_weight_kg=800, core_diameter_mm=169.999) + self.assertIn("allowed_core_diameters_mm", failed["failed_constraints"]) + + def test_diameter_distinguishes_pass_warning_and_failure(self): + passed = self.check(average_diameter_mm=800, maximum_diameter_mm=850) + warning = self.check(average_diameter_mm=876, maximum_diameter_mm=876.1) + failed = self.check(average_diameter_mm=876.1, maximum_diameter_mm=900) + self.assertEqual(passed["feasibility"], "feasible") + self.assertEqual(warning["feasibility"], "feasible_with_warnings") + self.assertEqual(warning["warning_constraints"], ["roll_diameter_mm"]) + self.assertEqual(failed["feasibility"], "not_feasible") + self.assertIn("roll_diameter_mm", failed["failed_constraints"]) + + def test_failures_and_warnings_are_all_retained(self): + result = self.check(roll_weight_kg=3000, average_diameter_mm=800, maximum_diameter_mm=900, core_diameter_mm=151, product_width_m=4.0) + self.assertEqual(result["feasibility"], "not_feasible") + self.assertEqual(result["warning_constraints"], ["roll_diameter_mm"]) + self.assertEqual(set(result["failed_constraints"]), {"roll_weight_kg", "allowed_core_diameters_mm", "product_width_m"}) + + def test_missing_values_keep_evaluable_checks_but_never_claim_feasible(self): + result = self.check(roll_weight_kg=None, maximum_diameter_mm=None, core_diameter_mm=None) + self.assertEqual(result["status"], "needs_clarification") + self.assertIsNone(result["feasible"]) + self.assertEqual(set(result["missing_required_inputs"]), {"roll_weight_kg", "maximum_diameter_mm", "core_diameter_mm"}) + self.assertIn("product_width_m", result["passed_constraints"]) + + def test_line_speed_is_not_required_or_defaulted_in_v1(self): + result = self.check(machine="gitter_1", roll_weight_kg=400, average_diameter_mm=600, maximum_diameter_mm=600, core_diameter_mm=140, product_width_m=4.0) + line_speed = result["checks"][-1] + self.assertEqual(result["feasibility"], "feasible") + self.assertEqual(line_speed["state"], "not_evaluated") + self.assertIsNone(line_speed["limits"]) + + def test_invalid_and_unknown_requests_are_structured(self): + self.assertEqual(check_production_feasibility({"machine": "B1", "unknown": 1})["status"], "invalid_parameter") + self.assertEqual(check_production_feasibility({"machine": "B1", "core_weight_kg": 10})["status"], "invalid_parameter") + self.assertEqual(self.check(machine="not a machine")["status"], "machine_not_found") diff --git a/tests/test_mcp_adapter.py b/tests/test_mcp_adapter.py index b1cfc46..83130b2 100644 --- a/tests/test_mcp_adapter.py +++ b/tests/test_mcp_adapter.py @@ -14,12 +14,15 @@ from roll_calculation import ( ) from rollcalc_mcp_tools import ( analyze_transport_capacity_result, + check_production_feasibility_result, calculate_material_weight_result, calculate_product_length_result, calculate_roll_diameter_result, get_article_result, + get_machine_result, search_articles_result, ) +from machine_constraints import check_production_feasibility, get_machine from transport_calculation import analyze_transport, transport_presets @@ -76,6 +79,25 @@ class McpAdapterTests(unittest.TestCase): "candidates": [], }) + def test_machine_adapters_delegate_to_the_shared_domain(self): + arguments = { + "machine": "B1", + "roll_weight_kg": 2500.0, + "average_diameter_mm": 876.0, + "maximum_diameter_mm": 876.1, + "core_diameter_mm": 168.0, + "product_width_m": 5.0, + } + self.assertEqual(get_machine_result("B1"), get_machine("B1")) + self.assertEqual( + check_production_feasibility_result(**arguments), + check_production_feasibility(arguments), + ) + self.assertEqual( + check_production_feasibility_result(**arguments)["feasibility"], + "feasible_with_warnings", + ) + def test_material_weight_adapter_delegates_without_a_core_diameter(self): arguments = { "roll_length_m": 60.0, @@ -196,6 +218,8 @@ class McpAdapterTests(unittest.TestCase): [ "get_article", "search_articles", + "get_machine", + "check_production_feasibility", "calculate_material_weight", "calculate_product_length", "calculate_roll_diameter", @@ -207,6 +231,8 @@ class McpAdapterTests(unittest.TestCase): instance = mcp_server.create_server() tools = instance._tool_manager._tools search_tool = tools["search_articles"] + machine_tool = tools["get_machine"] + feasibility_tool = tools["check_production_feasibility"] material_weight_tool = tools["calculate_material_weight"] product_length_tool = tools["calculate_product_length"] roll_schema = tools["calculate_roll_diameter"].parameters @@ -214,6 +240,16 @@ class McpAdapterTests(unittest.TestCase): transport_schema = transport_tool.parameters self.assertEqual(search_tool.parameters["required"], ["query"]) + self.assertEqual(machine_tool.parameters["required"], ["machine"]) + self.assertEqual( + feasibility_tool.parameters["required"], ["machine"] + ) + self.assertIn("average_diameter_mm", feasibility_tool.parameters["properties"]) + self.assertIn("maximum_diameter_mm", feasibility_tool.parameters["properties"]) + self.assertIn("tool, not the LLM, decides", feasibility_tool.description) + self.assertIn("never override failed constraints", feasibility_tool.description) + self.assertIn("warnings", feasibility_tool.description) + self.assertIn("Missing configured V1 roll inputs", feasibility_tool.description) self.assertEqual( search_tool.parameters["properties"]["query"]["type"], "string" )