Add configurable production machine constraints

This commit is contained in:
2026-09-26 10:47:53 +02:00
parent e6a7d4da72
commit 3e13d24405
12 changed files with 904 additions and 3 deletions
+25 -3
View File
@@ -107,6 +107,7 @@ Implemented routes:
| `/api/calculations/roll` | `POST` | Basic Auth | Resolves, validates, and executes a direct roll-diameter request. | | `/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/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/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/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` | `POST` | Basic Auth | Creates an in-memory conversation session. |
| `/api/conversations/<id>/messages` | `POST` | Basic Auth | Interprets one utterance and executes the deterministic workflow. | | `/api/conversations/<id>/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 `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, a thin adapter over the same domain services used by Flask: article lookup,
direct roll calculation, product-length calculation, material-weight calculation, 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`, The available PoC tools are `get_article`, `search_articles`,
`calculate_material_weight`, `calculate_product_length`, `calculate_material_weight`, `calculate_product_length`,
`calculate_roll_diameter`, and `calculate_roll_diameter`, `analyze_transport_capacity`, `get_machine`, and
`analyze_transport_capacity`. `check_production_feasibility`.
Install the pinned dependencies, including `mcp==1.26.0`, in the existing Install the pinned dependencies, including `mcp==1.26.0`, in the existing
environment, then start it with: environment, then start it with:
@@ -165,6 +166,16 @@ Tool calling semantics:
`transport_roll_inputs` unchanged. The supplied `roll_weight_kg` is passed `transport_roll_inputs` unchanged. The supplied `roll_weight_kg` is passed
through unchanged. Callers may use a canonical `transport_preset` or provide through unchanged. Callers may use a canonical `transport_preset` or provide
custom transport dimensions and `max_weight_kg`. 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: 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. 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` ### `access_log.json`
JSON audit log written by `app.py`. JSON audit log written by `app.py`.
+14
View File
@@ -20,6 +20,7 @@ from flask_httpauth import HTTPBasicAuth
from werkzeug.security import check_password_hash from werkzeug.security import check_password_hash
from core_presets import CORE_PRESETS from core_presets import CORE_PRESETS
from machine_constraints import check_production_feasibility
from conversation_service import ( from conversation_service import (
ConversationNotFoundError, ConversationNotFoundError,
ConversationReportNotFoundError, ConversationReportNotFoundError,
@@ -264,6 +265,19 @@ def create_transport_calculation():
return jsonify(result), 200 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): def _create_pdf(calculation_request):
result = calculate_roll(calculation_request, build_info=BUILD_INFO) result = calculate_roll(calculation_request, build_info=BUILD_INFO)
report = report_from_calculation_result(result) report = report_from_calculation_result(result)
+153
View File
@@ -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
+16
View File
@@ -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)
1 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
2 Bento 1 Bento1, B1 2501 876 300 150, 170, 200, 235 6 0,9 5 4,7 Espelkamp (Germany)
3 Bento 2 Bento2, B2 2501 1000 250 150, 170, 200, 235 6 0,9 5 4,7 Selangor (Malaysia)
4 K1 K 1 800 1100 200 130, 150, 170 20 1 6,1 4 Selangor (Malaysia)
5 K3/4 K 34, K 3/4 800 1100 200 130, 150, 170 15 1 6,1 4 Espelkamp (Germany)
6 K5 K 5 800 1200 200 130, 150, 170 20 1 6,1 4 Adorf (Germany)
7 K6 K 6 800 1100 200 130, 150, 170 20 1 6,1 4 Selangor (Malaysia)
8 K7 K 7 700 1100 300 130, 150, 170 40 0,6 6,1 4 Espelkamp (Germany)
9 E1 E 1 2500 1000 200 150, 170 10 0,6 7,5 7,5 Espelkamp (Germany)
10 E4 E 4 2000 900 200 150, 170 10 0,5 5,1 5,1 Espelkamp (Germany)
11 Gitter 1 G1, G 1 400 600 200 140 4,75 4 Adorf (Germany)
12 Gitter 2 G2, G 2 400 600 200 140 4,75 4 Adorf (Germany)
13 Gitter 3 G3, G 3 400 600 200 140 4,75 4 Adorf (Germany)
14 Gitter 4 G4, G 4 400 600 200 140 4,75 4 Adorf (Germany)
15 Secudrain 1/2 Sec 1/2, Secudrain 12, Sec 12 400 800 200 120 4 2 Adorf (Germany)
16 Secudrain 3 Sec 3 400 800 200 120 2,1 1,8 Adorf (Germany)
+17
View File
@@ -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. 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 ## Constrained Natural-Language Flow
```text ```text
+402
View File
@@ -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,
}
+28
View File
@@ -10,6 +10,8 @@ from rollcalc_mcp_tools import (
calculate_product_length_result, calculate_product_length_result,
calculate_roll_diameter_result, calculate_roll_diameter_result,
get_article_result, get_article_result,
get_machine_result,
check_production_feasibility_result,
search_articles_result, search_articles_result,
) )
from transport_calculation import TransportPresetKey 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.""" """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) 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() @server.tool()
def calculate_material_weight( def calculate_material_weight(
roll_length_m: float, roll_length_m: float,
+1
View File
@@ -6,3 +6,4 @@ itsdangerous==2.1.2
Jinja2==3.1.2 Jinja2==3.1.2
MarkupSafe==2.1.3 MarkupSafe==2.1.3
mcp==1.26.0 mcp==1.26.0
PyYAML==6.0.2
+28
View File
@@ -4,6 +4,7 @@ from __future__ import annotations
from typing import Any from typing import Any
from machine_constraints import check_production_feasibility, get_machine
from roll_calculation import ( from roll_calculation import (
ArticleRepository, ArticleRepository,
calculate_material_weight, calculate_material_weight,
@@ -28,6 +29,33 @@ def search_articles_result(query: str) -> dict[str, Any]:
return ArticleRepository.load().search(query) 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( def calculate_material_weight_result(
*, *,
roll_length_m: float, roll_length_m: float,
+19
View File
@@ -120,6 +120,25 @@ class CalculationApiTests(unittest.TestCase):
self.assertEqual(response.status_code, 200) self.assertEqual(response.status_code, 200)
self.assertEqual(response.get_json()["status"], "invalid_parameter") 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__": if __name__ == "__main__":
unittest.main() unittest.main()
+165
View File
@@ -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")
+36
View File
@@ -14,12 +14,15 @@ from roll_calculation import (
) )
from rollcalc_mcp_tools import ( from rollcalc_mcp_tools import (
analyze_transport_capacity_result, analyze_transport_capacity_result,
check_production_feasibility_result,
calculate_material_weight_result, calculate_material_weight_result,
calculate_product_length_result, calculate_product_length_result,
calculate_roll_diameter_result, calculate_roll_diameter_result,
get_article_result, get_article_result,
get_machine_result,
search_articles_result, search_articles_result,
) )
from machine_constraints import check_production_feasibility, get_machine
from transport_calculation import analyze_transport, transport_presets from transport_calculation import analyze_transport, transport_presets
@@ -76,6 +79,25 @@ class McpAdapterTests(unittest.TestCase):
"candidates": [], "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): def test_material_weight_adapter_delegates_without_a_core_diameter(self):
arguments = { arguments = {
"roll_length_m": 60.0, "roll_length_m": 60.0,
@@ -196,6 +218,8 @@ class McpAdapterTests(unittest.TestCase):
[ [
"get_article", "get_article",
"search_articles", "search_articles",
"get_machine",
"check_production_feasibility",
"calculate_material_weight", "calculate_material_weight",
"calculate_product_length", "calculate_product_length",
"calculate_roll_diameter", "calculate_roll_diameter",
@@ -207,6 +231,8 @@ class McpAdapterTests(unittest.TestCase):
instance = mcp_server.create_server() instance = mcp_server.create_server()
tools = instance._tool_manager._tools tools = instance._tool_manager._tools
search_tool = tools["search_articles"] search_tool = tools["search_articles"]
machine_tool = tools["get_machine"]
feasibility_tool = tools["check_production_feasibility"]
material_weight_tool = tools["calculate_material_weight"] material_weight_tool = tools["calculate_material_weight"]
product_length_tool = tools["calculate_product_length"] product_length_tool = tools["calculate_product_length"]
roll_schema = tools["calculate_roll_diameter"].parameters roll_schema = tools["calculate_roll_diameter"].parameters
@@ -214,6 +240,16 @@ class McpAdapterTests(unittest.TestCase):
transport_schema = transport_tool.parameters transport_schema = transport_tool.parameters
self.assertEqual(search_tool.parameters["required"], ["query"]) 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( self.assertEqual(
search_tool.parameters["properties"]["query"]["type"], "string" search_tool.parameters["properties"]["query"]["type"], "string"
) )