120 lines
3.7 KiB
Python
120 lines
3.7 KiB
Python
"""Thin MCP-facing adapters over the shared RollCalc domain services."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from roll_calculation import (
|
|
ArticleRepository,
|
|
calculate_material_weight,
|
|
calculate_product_length,
|
|
calculate_roll,
|
|
get_article,
|
|
transport_roll_inputs_from_roll_calculation,
|
|
)
|
|
from transport_calculation import analyze_transport
|
|
|
|
|
|
def get_article_result(
|
|
article_number: str | None = None,
|
|
article_name_hint: str | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Delegate article resolution to the shared article repository."""
|
|
return get_article(article_number, article_name_hint)
|
|
|
|
|
|
def search_articles_result(query: str) -> dict[str, Any]:
|
|
"""Delegate article discovery to the shared article repository."""
|
|
return ArticleRepository.load().search(query)
|
|
|
|
|
|
def calculate_material_weight_result(
|
|
*,
|
|
roll_length_m: float,
|
|
width_m: float,
|
|
area_weight_g_m2: float,
|
|
) -> dict[str, Any]:
|
|
"""Delegate material-only weight calculation to the shared domain service."""
|
|
return calculate_material_weight(
|
|
roll_length_m=roll_length_m,
|
|
width_m=width_m,
|
|
area_weight_g_m2=area_weight_g_m2,
|
|
)
|
|
|
|
|
|
def calculate_product_length_result(
|
|
*,
|
|
target_roll_diameter_mm: float,
|
|
core_diameter_mm: float,
|
|
thickness_mm: float,
|
|
thickness_stddev_mm: float | None = None,
|
|
width_m: float | None = None,
|
|
area_weight_g_m2: float | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Delegate target-diameter product-length calculation to the domain."""
|
|
return calculate_product_length(
|
|
target_roll_diameter_mm=target_roll_diameter_mm,
|
|
core_diameter_mm=core_diameter_mm,
|
|
thickness_mm=thickness_mm,
|
|
thickness_stddev_mm=thickness_stddev_mm,
|
|
width_m=width_m,
|
|
area_weight_g_m2=area_weight_g_m2,
|
|
)
|
|
|
|
|
|
def calculate_roll_diameter_result(
|
|
*,
|
|
article_number: str | None = None,
|
|
article_name_hint: str | None = None,
|
|
roll_length_m: float | None = None,
|
|
width_m: float | None = None,
|
|
thickness_mm: float | None = None,
|
|
thickness_stddev_mm: float | None = None,
|
|
area_weight_g_m2: float | None = None,
|
|
core_diameter_mm: float | None = None,
|
|
core_type: str | None = None,
|
|
category: str | None = None,
|
|
production_site: str | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Map MCP inputs to the existing calculation, always requesting weight."""
|
|
result = calculate_roll({
|
|
"article_number": article_number,
|
|
"article_name_hint": article_name_hint,
|
|
"roll_length_m": roll_length_m,
|
|
"width_m": width_m,
|
|
"thickness_mm": thickness_mm,
|
|
"thickness_stddev_mm": thickness_stddev_mm,
|
|
"area_weight_g_m2": area_weight_g_m2,
|
|
"core_diameter_mm": core_diameter_mm,
|
|
"core_type": core_type,
|
|
"category": category,
|
|
"production_site": production_site,
|
|
"include_roll_weight": True,
|
|
})
|
|
if result.get("status") == "success":
|
|
result = dict(result)
|
|
result["transport_roll_inputs"] = (
|
|
transport_roll_inputs_from_roll_calculation(result)
|
|
)
|
|
return result
|
|
|
|
|
|
def analyze_transport_capacity_result(
|
|
*,
|
|
transport_preset: str,
|
|
roll_diameter_mm: float,
|
|
core_diameter_mm: float,
|
|
roll_width_m: float,
|
|
roll_weight_kg: float,
|
|
product_length_m: float,
|
|
) -> dict[str, Any]:
|
|
"""Map the preset-only MCP contract to the shared transport domain."""
|
|
return analyze_transport({
|
|
"transport_preset": transport_preset,
|
|
"roll_diameter_mm": roll_diameter_mm,
|
|
"core_diameter_mm": core_diameter_mm,
|
|
"roll_width_m": roll_width_m,
|
|
"roll_weight_kg": roll_weight_kg,
|
|
"product_length_m": product_length_m,
|
|
})
|