M/M/1 Queue Simulation Server

""" Simulation MCP Server (v0.3.2) Model Context Protocol server for simulation systems: - M/M/1: Single-server queuing - M/M/c: Multi-server queuing - MPS: Matrix Production System (manufacturing) New in v0.3.2: - verify_mps_schedule: Verify schedule and calculate makespan - Schedule verification for experiment evaluation - MAPE comparison with ground truth makespan New in v0.3.1: - MPS updated to match KTH Master Thesis (2025) definition - Grid layout with Manhattan distance calculation - Material Handling Vehicle (MHV) transport with repositioning - Post-task buffer constraints - Task-machine eligibility New in v0.3.0: - Matrix Production System (MPS) support - mps:// resources for manufacturing simulation New in v0.2.x: - M/M/c simulation with Erlang C - mmc:// resources for M/M/c schema - Comparison tools (pooled vs separate queues) Resources (mm1:// prefix - M/M/1 queuing): - mm1://schema, mm1://parameters, mm1://metrics - mm1://formulas, mm1://guidelines, mm1://examples Resources (mmc:// prefix - M/M/c queuing): - mmc://schema, mmc://parameters, mmc://formulas - mmc://metrics, mmc://examples, mmc://comparison Resources (mps:// prefix - Manufacturing): - mps://schema - Complete MPS schema (v2.0, KTH thesis format) - mps://parameters - Grid layout, machines, tasks, products, MHV - mps://formulas - Scheduling constraints, Manhattan distance - mps://metrics - Makespan, utilization, transport times - mps://examples - Manufacturing scenarios - mps://comparison - MPS vs queuing comparison - mps://guidelines - SimPy implementation patterns Tools - Queuing: - validate_config - Validate queue parameters - calculate_metrics - Compute M/M/1 theory - run_simulation - Execute M/M/1 simulation - run_mmc_simulation - Execute M/M/c simulation - compare_mm1_vs_mmc - Compare strategies - analyze_cashier_problem - Supermarket analysis - recommend_parameters - Suggest optimal parameters Tools - Manufacturing (MPS v2.0): - run_mps_sim - Execute MPS simulation (grid layout, MHV transport) - calculate_mps_metrics - Theoretical analysis (load, bottleneck) - analyze_mps_schedule - Detailed schedule analysis - get_mps_default_config - Default 3x2 grid configuration - verify_mps_schedule - Verify schedule and calculate makespan (NEW) Prompts: - generate_simulation_code - Create SimPy code - explain_mm1_theory - Explain M/M/1 theory - explain_mmc_theory - Explain M/M/c theory - analyze_results - Analyze simulation results - debug_simulation - Debug simulation issues """ try: from fastmcp import FastMCP except ImportError: try: from mcp.server.fastmcp import FastMCP except ImportError: # Fallback for older MCP installations from mcp.server import Server as FastMCP from typing import Dict, Any, Optional import json import math import time import random import numpy as np # Import our modules from .schemas import MM1_SCHEMA, MMC_SCHEMA, MPS_SCHEMA from .utils.metrics import ( validate_mm1_config, calculate_theoretical_metrics, compare_simulation_to_theory, get_recommended_simulation_time ) from .simulations.mm1_queue import run_mm1_simulation # Import M/M/c support try: from .simulations.mmc_queue_fixed import ( MMcConfig, MMcQueueSimulation, calculate_mmc_theoretical ) MMC_AVAILABLE = True except ImportError: MMC_AVAILABLE = False print("Warning: M/M/c simulation not available, using fallback") # Import MPS support (v2.0 - matches KTH thesis definition) try: from .simulations.mps_simulation import ( MPSConfig, MPSSimulation, MPSResult, Machine, Task, Product, Job, MHV, run_mps_simulation, create_config_from_dict, verify_mps_schedule as _verify_mps_schedule, get_default_mps_config as _get_default_mps_config ) MPS_AVAILABLE = True except ImportError: MPS_AVAILABLE = False print("Warning: MPS simulation not available") # ============================================================================ # Initialize FastMCP Server # ============================================================================ mcp = FastMCP("Simulation Server (M/M/1, M/M/c, MPS)", version="0.3.2") # ============================================================================ # RESOURCES: Data Providers # ============================================================================ @mcp.resource("mm1://schema") def get_full_schema() -> str: """ Complete M/M/1 and M/M/c queue system schema Returns the full schema including parameters, metrics, formulas, implementation guidelines, and example use cases. Now includes M/M/c multi-server support. """ schema = dict(MM1_SCHEMA) schema["version"] = "0.2.0" schema["supports"] = ["M/M/1", "M/M/c (c servers)"] return json.dumps(schema, indent=2) @mcp.resource("mm1://parameters") def get_parameters() -> str: """ M/M/1 and M/M/c parameter definitions Returns detailed information about input parameters including symbols, types, constraints, and examples. Now includes num_servers parameter for M/M/c. """ params = dict(MM1_SCHEMA["parameters"]) params["num_servers"] = { "symbol": "c", "description": "Number of parallel servers (1 for M/M/1, >1 for M/M/c)", "type": "integer", "constraints": "c >= 1", "default": 1, "examples": [1, 2, 3, 10] } return json.dumps({ "parameters": params, "system_characteristics": MM1_SCHEMA["system_characteristics"] }, indent=2) @mcp.resource("mm1://metrics") def get_metrics() -> str: """ M/M/1 and M/M/c performance metrics Returns definitions of all performance metrics that can be calculated, including required and optional metrics. """ return json.dumps({ "performance_metrics": MM1_SCHEMA["performance_metrics"], "mmc_specific": { "erlang_c": "Probability of queueing in M/M/c", "server_utilization": "Average utilization across all servers" } }, indent=2) @mcp.resource("mm1://formulas") def get_formulas() -> str: """ M/M/1 and M/M/c theoretical formulas Returns LaTeX and plain text formulas for calculating theoretical performance metrics. """ formulas = dict(MM1_SCHEMA["theoretical_formulas"]) formulas["mmc_formulas"] = { "utilization": "ρ = λ/(c×μ)", "erlang_c": "Pc = ((λ/μ)^c / c!) × P0 / (1-ρ)", "avg_queue_length": "Lq = Pc × ρ / (1-ρ)", "avg_waiting_time": "Wq = Lq / λ" } return json.dumps(formulas, indent=2) @mcp.resource("mm1://guidelines") def get_guidelines() -> str: """ Implementation guidelines for M/M/1 and M/M/c simulations """ return json.dumps(MM1_SCHEMA["implementation_guidelines"], indent=2) @mcp.resource("mm1://examples") def get_examples() -> str: """ Example configurations for common scenarios Now includes M/M/c examples like supermarket cashiers, bank tellers, call centers, etc. """ examples = list(MM1_SCHEMA["example_configurations"]) examples.extend([ { "name": "Supermarket Cashiers", "description": "3 cashiers with pooled queue", "model": "M/M/3", "parameters": { "arrival_rate": 12, "service_rate": 5, "num_servers": 3 }, "expected_utilization": 0.8 }, { "name": "Call Center", "description": "10 agents handling calls", "model": "M/M/10", "parameters": { "arrival_rate": 45, "service_rate": 5, "num_servers": 10 }, "expected_utilization": 0.9 } ]) return json.dumps(examples, indent=2) @mcp.resource("mm1://literature") def get_literature() -> str: """ Academic references and literature """ return json.dumps(MM1_SCHEMA["literature_references"], indent=2) # ============================================================================ # RESOURCES: M/M/c Schema (mmc:// prefix) # ============================================================================ @mcp.resource("mmc://schema") def get_mmc_schema() -> str: """ Complete M/M/c queue system schema Returns the full schema for multi-server queuing systems including parameters, metrics, Erlang C formulas, and implementation guidelines. """ return json.dumps(MMC_SCHEMA, indent=2, ensure_ascii=False) @mcp.resource("mmc://parameters") def get_mmc_parameters() -> str: """ M/M/c parameter definitions Returns detailed information about input parameters for M/M/c systems including arrival rate, service rate per server, and number of servers. """ return json.dumps(MMC_SCHEMA["parameters"], indent=2) @mcp.resource("mmc://formulas") def get_mmc_formulas() -> str: """ M/M/c theoretical formulas including Erlang C Returns formulas for: - Utilization: ρ = λ/(c×μ) - P₀: Probability of empty system - Erlang C: Probability of waiting - Lq, Wq, L, W metrics """ return json.dumps(MMC_SCHEMA["theoretical_validation"], indent=2) @mcp.resource("mmc://metrics") def get_mmc_metrics() -> str: """ M/M/c performance metrics Returns definitions of all performance metrics for multi-server systems including Erlang C probability and server utilization. """ return json.dumps(MMC_SCHEMA["performance_metrics"], indent=2) @mcp.resource("mmc://examples") def get_mmc_examples() -> str: """ M/M/c example configurations Returns practical examples including call centers, hospital triage, bank tellers, and other multi-server scenarios. """ return json.dumps(MMC_SCHEMA["example_use_cases"], indent=2) @mcp.resource("mmc://comparison") def get_mmc_comparison() -> str: """ M/M/1 vs M/M/c comparison Returns key differences between single-server and multi-server systems, including when to use each model. """ return json.dumps(MMC_SCHEMA["comparison_with_mm1"], indent=2) @mcp.resource("mmc://literature") def get_mmc_literature() -> str: """ M/M/c academic references Returns literature references for Erlang C formula and multi-server queuing theory. """ return json.dumps(MMC_SCHEMA["literature_references"], indent=2) # ============================================================================ # RESOURCES: Matrix Production System (mps:// prefix) # ============================================================================ @mcp.resource("mps://schema") def get_mps_schema() -> str: """ Complete Matrix Production System schema Returns the full schema for manufacturing simulation including BOM, routing, work stations, and capacity planning. """ return json.dumps(MPS_SCHEMA, indent=2, ensure_ascii=False) @mcp.resource("mps://parameters") def get_mps_parameters() -> str: """ MPS parameter definitions Returns detailed information about input parameters for MPS systems including products, BOM, routing, and work stations. """ return json.dumps(MPS_SCHEMA["parameters"], indent=2) @mcp.resource("mps://formulas") def get_mps_formulas() -> str: """ MPS theoretical formulas Returns Factory Physics formulas for: - Little's Law: WIP = TH × LT - Bottleneck identification - Utilization calculations """ return json.dumps(MPS_SCHEMA["theoretical_validation"], indent=2) @mcp.resource("mps://metrics") def get_mps_metrics() -> str: """ MPS performance metrics Returns definitions of manufacturing performance metrics including throughput, lead time, WIP, utilization, and makespan. """ return json.dumps(MPS_SCHEMA["performance_metrics"], indent=2) @mcp.resource("mps://examples") def get_mps_examples() -> str: """ MPS example configurations Returns practical examples including job shops, assembly lines, and electronics manufacturing. """ return json.dumps(MPS_SCHEMA["example_use_cases"], indent=2) @mcp.resource("mps://comparison") def get_mps_vs_queuing() -> str: """ MPS vs Queuing Systems comparison Returns key differences between manufacturing systems and simple queuing models. """ return json.dumps(MPS_SCHEMA["comparison_with_queuing"], indent=2) @mcp.resource("mps://literature") def get_mps_literature() -> str: """ MPS academic references Returns literature references including Factory Physics, Theory of Constraints, and scheduling theory. """ return json.dumps(MPS_SCHEMA["literature_references"], indent=2) @mcp.resource("mps://guidelines") def get_mps_guidelines() -> str: """ MPS implementation guidelines Returns guidelines for implementing MPS simulations in SimPy including class structure and patterns. """ return json.dumps(MPS_SCHEMA["implementation_guidelines"], indent=2) # ============================================================================ # TOOLS: M/M/1 Legacy Support # ============================================================================ @mcp.tool() async def validate_config( arrival_rate: float, service_rate: float, simulation_time: float = 10000.0 ) -> Dict[str, Any]: """ Validate M/M/1 queue configuration Checks if parameters are valid and system is stable (ρ < 1). Args: arrival_rate: λ (customers per time unit) service_rate: μ (customers per time unit) simulation_time: Duration of simulation Returns: Validation results with stability analysis """ return validate_mm1_config(arrival_rate, service_rate, simulation_time) @mcp.tool() async def calculate_metrics( arrival_rate: float, service_rate: float ) -> Dict[str, Any]: """ Calculate theoretical M/M/1 performance metrics Uses exact formulas to compute steady-state metrics. Args: arrival_rate: λ (customers per time unit) service_rate: μ (customers per time unit) Returns: Dictionary with theoretical performance metrics """ return calculate_theoretical_metrics(arrival_rate, service_rate) @mcp.tool() async def run_simulation( arrival_rate: float, service_rate: float, simulation_time: float = 10000.0, random_seed: Optional[int] = None ) -> Dict[str, Any]: """ Run M/M/1 queue simulation using SimPy Legacy tool for single-server simulation. For multi-server, use run_mmc_simulation. Args: arrival_rate: λ (customers per time unit) service_rate: μ (customers per time unit) simulation_time: Duration of simulation random_seed: Random seed (None for random) Returns: Simulation and theoretical metrics """ # Handle random seed if random_seed is None: random_seed = int(time.time() * 1000) % 2147483647 # Validate first validation = validate_mm1_config(arrival_rate, service_rate, simulation_time) if not validation["valid"]: return { "error": "Invalid configuration", "details": validation } try: # Run simulation sim_metrics = run_mm1_simulation( arrival_rate=arrival_rate, service_rate=service_rate, simulation_time=simulation_time, random_seed=random_seed ) # Calculate theoretical theo_metrics = calculate_theoretical_metrics(arrival_rate, service_rate) # Compare comparison = compare_simulation_to_theory(sim_metrics, theo_metrics) return { "simulation_metrics": sim_metrics, "theoretical_metrics": theo_metrics, "comparison": comparison, "config": { "arrival_rate": arrival_rate, "service_rate": service_rate, "simulation_time": simulation_time, "random_seed": random_seed } } except Exception as e: return { "error": str(e), "traceback": str(e.__traceback__) } # ============================================================================ # TOOLS: M/M/c Support (NEW) # ============================================================================ @mcp.tool() async def run_mmc_simulation( arrival_rate: float, service_rate: float, num_servers: int = 1, simulation_time: float = 10000.0, random_seed: Optional[int] = None ) -> Dict[str, Any]: """ Run M/M/c queue simulation (supports 1 to n servers) This tool can simulate: - M/M/1: Single server (num_servers=1) - M/M/2: Two servers (num_servers=2) - M/M/3: Three servers (num_servers=3) - M/M/c: Any number of servers Args: arrival_rate: λ (customers per time unit) service_rate: μ (customers per time unit per server) num_servers: c (number of parallel servers) simulation_time: Duration of simulation random_seed: Random seed (None for random) Returns: Simulation and theoretical metrics for M/M/c system Example: # Supermarket with 3 cashiers run_mmc_simulation( arrival_rate=12, # 12 customers/min arrive service_rate=5, # Each cashier serves 5/min num_servers=3 # 3 cashiers ) """ if not MMC_AVAILABLE: return { "error": "M/M/c simulation not available", "suggestion": "Use run_simulation for M/M/1 only" } # Handle random seed if random_seed is None: random_seed = int(time.time() * 1000) % 2147483647 # Validate stability rho = arrival_rate / (num_servers * service_rate) if rho >= 1: return { "error": "System unstable", "details": { "utilization": rho, "message": f"ρ = λ/(c×μ) = {arrival_rate}/({num_servers}×{service_rate}) = {rho:.3f} >= 1", "suggestion": f"Need λ < {num_servers * service_rate} or more servers" } } try: # Create config config = MMcConfig( arrival_rate=arrival_rate, service_rate=service_rate, num_servers=num_servers, simulation_time=simulation_time, random_seed=random_seed ) # Run simulation sim = MMcQueueSimulation(config) sim_results = sim.run() # Calculate theoretical theo_results = calculate_mmc_theoretical( arrival_rate, service_rate, num_servers ) # Compare results if "error" not in sim_results and "error" not in theo_results: comparison = { "utilization_error": abs(sim_results["utilization"] - theo_results["utilization"]) / theo_results["utilization"] * 100, "waiting_time_error": abs(sim_results["avg_waiting_time"] - theo_results["avg_waiting_time"]) / max(theo_results["avg_waiting_time"], 0.001) * 100, "queue_length_error": abs(sim_results["avg_queue_length"] - theo_results["avg_queue_length"]) / max(theo_results["avg_queue_length"], 0.001) * 100 } mape = sum(comparison.values()) / len(comparison) else: comparison = {} mape = None return { "model": f"M/M/{num_servers}", "configuration": { "arrival_rate": arrival_rate, "service_rate": service_rate, "num_servers": num_servers, "utilization": rho, "simulation_time": simulation_time, "random_seed": random_seed }, "simulation_results": sim_results, "theoretical_results": theo_results, "accuracy": { "mape": mape, "errors": comparison } if mape else {} } except Exception as e: return { "error": str(e), "type": type(e).__name__ } @mcp.tool() async def compare_mm1_vs_mmc( arrival_rate: float, service_rate_per_server: float, num_servers: int = 3, simulation_time: float = 10000.0 ) -> Dict[str, Any]: """ Compare M/M/1 × c (separate queues) vs M/M/c (pooled queue) This is the key tool for comparing: - Strategy A: c separate M/M/1 queues (distributed) - Strategy B: Single M/M/c queue (pooled) Perfect for supermarket/bank/coffee shop scenarios! Args: arrival_rate: Total arrival rate λ service_rate_per_server: Service rate μ per server num_servers: Number of servers c simulation_time: Simulation duration Returns: Comparison showing why pooled queue is better Example: # 3 cashiers: separate vs pooled queues compare_mm1_vs_mmc( arrival_rate=12, service_rate_per_server=5, num_servers=3 ) """ if not MMC_AVAILABLE: # Fallback to theoretical comparison only mm1_theo = calculate_theoretical_metrics( arrival_rate / num_servers, service_rate_per_server ) mmc_theo = { "error": "M/M/c simulation not available", "fallback": "Theoretical M/M/c calculation only" } return { "mm1_separate": mm1_theo, "mmc_pooled": mmc_theo, "note": "Install mmc_queue module for full simulation" } seed = int(time.time() * 1000) % 2147483647 # Strategy A: M/M/1 × c (each queue gets λ/c arrivals) mm1_results = [] for i in range(num_servers): config = MMcConfig( arrival_rate=arrival_rate / num_servers, service_rate=service_rate_per_server, num_servers=1, simulation_time=simulation_time, random_seed=seed + i ) sim = MMcQueueSimulation(config) result = sim.run() mm1_results.append(result) # Average across all M/M/1 queues mm1_avg = { "avg_waiting_time": np.mean([r["avg_waiting_time"] for r in mm1_results]), "avg_queue_length": sum(r["avg_queue_length"] for r in mm1_results), "avg_system_time": np.mean([r["avg_system_time"] for r in mm1_results]), "utilization": np.mean([r["utilization"] for r in mm1_results]) } # Strategy B: M/M/c (pooled queue) mmc_config = MMcConfig( arrival_rate=arrival_rate, service_rate=service_rate_per_server, num_servers=num_servers, simulation_time=simulation_time, random_seed=seed ) mmc_sim = MMcQueueSimulation(mmc_config) mmc_results = mmc_sim.run() # Theoretical values mm1_theo = calculate_theoretical_metrics( arrival_rate / num_servers, service_rate_per_server ) mmc_theo = calculate_mmc_theoretical( arrival_rate, service_rate_per_server, num_servers ) # Calculate improvements wait_improvement = (mm1_avg["avg_waiting_time"] - mmc_results["avg_waiting_time"]) / mm1_avg["avg_waiting_time"] * 100 queue_improvement = (mm1_avg["avg_queue_length"] - mmc_results["avg_queue_length"]) / mm1_avg["avg_queue_length"] * 100 return { "scenario": { "total_arrival_rate": arrival_rate, "service_rate_per_server": service_rate_per_server, "num_servers": num_servers, "system_utilization": arrival_rate / (num_servers * service_rate_per_server) }, "strategy_a_separate_queues": { "model": f"M/M/1 × {num_servers}", "simulation": mm1_avg, "theoretical": { "avg_waiting_time": mm1_theo["avg_waiting_time"], "avg_queue_length": mm1_theo["avg_queue_length"] * num_servers, "avg_system_time": mm1_theo["avg_system_time"] } }, "strategy_b_pooled_queue": { "model": f"M/M/{num_servers}", "simulation": { "avg_waiting_time": mmc_results["avg_waiting_time"], "avg_queue_length": mmc_results["avg_queue_length"], "avg_system_time": mmc_results["avg_system_time"], "utilization": mmc_results["utilization"] }, "theoretical": { "avg_waiting_time": mmc_theo["avg_waiting_time"], "avg_queue_length": mmc_theo["avg_queue_length"], "avg_system_time": mmc_theo["avg_system_time"] } }, "pooled_queue_benefits": { "waiting_time_reduction": f"{wait_improvement:.1f}%", "queue_length_reduction": f"{queue_improvement:.1f}%", "better_load_balancing": "Yes - servers share the workload", "fairness": "FIFO guaranteed across all customers", "recommendation": "✅ Use pooled queue (M/M/c)" if wait_improvement > 0 else "Use separate queues" } } @mcp.tool() async def analyze_cashier_problem( arrival_rate: float = 12.0, service_rate: float = 5.0, num_cashiers: int = 3 ) -> Dict[str, Any]: """ Analyze the supermarket cashier problem Compares separate queues vs single pooled queue for cashiers. Args: arrival_rate: New arrivals per minute service_rate: Customers served per minute per cashier num_cashiers: Number of cashiers Returns: Complete analysis with recommendations """ # Run comparison comparison = await compare_mm1_vs_mmc( arrival_rate=arrival_rate, service_rate_per_server=service_rate, num_servers=num_cashiers, simulation_time=10000 ) return { "problem": { "description": f"Supermarket with {num_cashiers} cashiers", "arrival_rate": f"{arrival_rate} customers/minute", "service_rate": f"{service_rate} customers/minute per cashier", "utilization": arrival_rate / (num_cashiers * service_rate) }, "results": comparison, "recommendation": { "best_strategy": "Single pooled queue (M/M/c)", "reasons": [ f"Reduces average wait by {comparison['pooled_queue_benefits']['waiting_time_reduction']}", "Guarantees fairness (FIFO)", "Better server utilization", "No queue selection anxiety for customers" ], "implementation": "Use a single queue with multiple cashiers like banks do" } } @mcp.tool() async def recommend_parameters( target_utilization: float = 0.8, target_waiting_time: float = 1.0, arrival_rate: Optional[float] = None, service_rate: Optional[float] = None, num_servers: Optional[int] = None ) -> Dict[str, Any]: """ Recommend optimal parameters for queue system Given constraints, suggests missing parameters. Args: target_utilization: Desired ρ (default: 0.8) target_waiting_time: Desired Wq in time units arrival_rate: Known λ (optional) service_rate: Known μ (optional) num_servers: Known c (optional) Returns: Recommended configuration with analysis """ recommendations = { "target_utilization": target_utilization, "target_waiting_time": target_waiting_time, "given_parameters": {} } if arrival_rate: recommendations["given_parameters"]["arrival_rate"] = arrival_rate if service_rate: recommendations["given_parameters"]["service_rate"] = service_rate if num_servers: recommendations["given_parameters"]["num_servers"] = num_servers # Calculate recommendations based on what's known if arrival_rate and not service_rate: if num_servers: # Calculate required service rate for M/M/c required_service_rate = arrival_rate / (num_servers * target_utilization) recommendations["recommended"] = { "service_rate": required_service_rate, "reason": f"To achieve ρ={target_utilization} with λ={arrival_rate} and c={num_servers}" } else: # Suggest for M/M/1 required_service_rate = arrival_rate / target_utilization recommendations["recommended"] = { "service_rate": required_service_rate, "num_servers": 1, "reason": f"M/M/1 system with ρ={target_utilization}" } elif service_rate and not arrival_rate: if num_servers: max_arrival_rate = num_servers * service_rate * target_utilization recommendations["recommended"] = { "max_arrival_rate": max_arrival_rate, "reason": f"Maximum λ for stable system with c={num_servers}" } else: max_arrival_rate = service_rate * target_utilization recommendations["recommended"] = { "max_arrival_rate": max_arrival_rate, "num_servers": 1, "reason": f"M/M/1 system" } elif arrival_rate and service_rate and not num_servers: # Calculate required servers min_servers = math.ceil(arrival_rate / (service_rate * target_utilization)) recommendations["recommended"] = { "num_servers": min_servers, "actual_utilization": arrival_rate / (min_servers * service_rate), "reason": f"Minimum servers needed for ρ < {target_utilization}" } # Add simulation time recommendation if arrival_rate: recommendations["simulation_time"] = get_recommended_simulation_time( arrival_rate, service_rate or arrival_rate/target_utilization ) return recommendations # ============================================================================ # TOOLS: Matrix Production System (MPS) v2.0 # Based on KTH Master Thesis (2025) definition: # - Grid layout with Manhattan distance # - Material Handling Vehicles (MHV) with repositioning # - Post-task buffer constraints # - Task-machine eligibility # ============================================================================ @mcp.tool() async def run_mps_sim( config: Optional[dict] = None, layout: Optional[dict] = None, machines: Optional[list] = None, tasks: Optional[list] = None, products: Optional[list] = None, mhv: Optional[dict] = None, grid_distance: float = 15.0, buffer_capacity: int = 2, random_seed: Optional[int] = None, output_format: str = "default" ) -> Dict[str, Any]: """ Run Matrix Production System (MPS) simulation. This implements MPS as defined in the KTH thesis (2025): - Grid layout with machines at fixed positions - MHV (Material Handling Vehicle) transport between machines - Manhattan distance for transport time calculation - Post-task buffer constraints - Task-machine eligibility constraints Args: config: Complete configuration dict (alternative to individual params) layout: Grid layout {"rows": 3, "cols": 2, "machine_positions": {"M1": [0,0], ...}} machines: List of machines [{"id": "M1", "eligible_tasks": ["T1", "T2"]}] tasks: List of tasks [{"id": "T1", "processing_time": 20}] products: List of products [{"id": "A", "task_sequence": ["T1","T3"], "quantity": 1}] mhv: MHV config {"count": 3, "speed": 1.0} grid_distance: Distance between adjacent machines (meters) buffer_capacity: Post-task buffer capacity per machine random_seed: Random seed for reproducibility output_format: Output format - "default" or "schedule" (for CP verification) - "default": Standard format with task_schedule, mhv_schedule lists - "schedule": CP-verification format with jobs grouped by job_id Returns: Simulation results. Format depends on output_format: "default" format: - makespan: Total completion time - task_schedule: List of task execution records - mhv_schedule: List of transport records - machine_utilization: Utilization per machine - mhv_utilization: Utilization per MHV "schedule" format (for CP verification): - jobs: List of jobs with operations grouped by job_id - transports: List of transport records - calculated_makespan: Maximum end time Example (3×2 grid, 6 machines, 5 products): run_mps_sim( layout={"rows": 3, "cols": 2, "machine_positions": { "M1": [0,0], "M2": [0,1], "M3": [1,0], "M4": [1,1], "M5": [2,0], "M6": [2,1] }}, machines=[ {"id": "M1", "eligible_tasks": ["T1", "T2"]}, {"id": "M2", "eligible_tasks": ["T3", "T4"]}, {"id": "M3", "eligible_tasks": ["T5", "T6"]}, {"id": "M4", "eligible_tasks": ["T1", "T3"]}, {"id": "M5", "eligible_tasks": ["T2", "T5"]}, {"id": "M6", "eligible_tasks": ["T4", "T6"]} ], tasks=[ {"id": "T1", "processing_time": 20}, {"id": "T2", "processing_time": 25}, {"id": "T3", "processing_time": 30}, {"id": "T4", "processing_time": 30}, {"id": "T5", "processing_time": 35}, {"id": "T6", "processing_time": 40} ], products=[ {"id": "A", "task_sequence": ["T1","T3","T4","T6"], "quantity": 1}, {"id": "B", "task_sequence": ["T1","T2","T4","T5"], "quantity": 1}, {"id": "C", "task_sequence": ["T2","T3","T5","T6"], "quantity": 1}, {"id": "D", "task_sequence": ["T6","T5","T1","T3"], "quantity": 1}, {"id": "E", "task_sequence": ["T4","T2","T5","T1"], "quantity": 1} ], mhv={"count": 5, "speed": 1.0}, grid_distance=15.0, buffer_capacity=2 ) """ if not MPS_AVAILABLE: return { "error": "MPS simulation not available", "suggestion": "Check MPS module installation" } try: # Build config dict if config is not None: config_dict = config else: config_dict = { "layout": layout or {"rows": 3, "cols": 2, "machine_positions": {}}, "machines": machines or [], "tasks": tasks or [], "products": products or [], "mhv": mhv or {"count": 3, "speed": 1.0}, "grid_distance": grid_distance, "buffer_capacity": buffer_capacity, "random_seed": random_seed } # Run simulation result = run_mps_simulation(config_dict) # Convert to schedule format if requested if output_format == "schedule": result = _convert_to_schedule_format(result) return result except Exception as e: return { "error": str(e), "type": type(e).__name__ } def _convert_to_schedule_format(result: Dict[str, Any]) -> Dict[str, Any]: """ Convert MPS simulation result to CP-verification schedule format. Groups task_schedule by job_id and reformats for CP model verification. """ if "error" in result: return result # Group tasks by job job_tasks = {} for record in result.get("task_schedule", []): job_id = record["job"] if job_id not in job_tasks: job_tasks[job_id] = [] job_tasks[job_id].append({ "task": record["task"], "machine": record["machine"], "start": record["start_time"], "end": record["end_time"] }) # Build jobs list jobs = [] for job_id in sorted(job_tasks.keys()): ops = sorted(job_tasks[job_id], key=lambda x: x["start"]) jobs.append({ "job_id": job_id, "product": job_id[0] if job_id else "", "operations": ops }) # Convert transports transports = [] for record in result.get("mhv_schedule", []): transports.append({ "job": record["job"], "mhv": record["mhv"], "from": record["from_machine"], "to": record["to_machine"], "depart": record["depart_time"], "arrive": record["arrive_time"] }) transports.sort(key=lambda x: x["depart"]) return { "jobs": jobs, "transports": transports, "calculated_makespan": result.get("makespan", 0) } @mcp.tool() async def calculate_mps_metrics( machines: list, tasks: list, products: list, grid_distance: float = 15.0, mhv_speed: float = 1.0 ) -> Dict[str, Any]: """ Calculate theoretical MPS metrics. Estimates manufacturing performance using analytical methods: - Total processing time per product - Minimum transport time (Manhattan distance) - Lower bound on makespan - Machine load analysis Args: machines: List of machines with eligible_tasks tasks: List of tasks with processing_time products: List of products with task_sequence and quantity Returns: Theoretical estimates and analysis Example: calculate_mps_metrics( machines=[ {"id": "M1", "eligible_tasks": ["T1", "T2"], "position": [0,0]}, {"id": "M2", "eligible_tasks": ["T3", "T4"], "position": [0,1]} ], tasks=[ {"id": "T1", "processing_time": 20}, {"id": "T2", "processing_time": 25}, {"id": "T3", "processing_time": 30}, {"id": "T4", "processing_time": 30} ], products=[ {"id": "A", "task_sequence": ["T1", "T3"], "quantity": 1} ] ) """ try: # Build task processing time lookup task_times = {t["id"]: t.get("processing_time", 0) for t in tasks} # Calculate total processing time per product type product_metrics = {} for p in products: total_proc = sum(task_times.get(t, 0) for t in p.get("task_sequence", [])) num_tasks = len(p.get("task_sequence", [])) product_metrics[p["id"]] = { "total_processing_time": total_proc, "num_tasks": num_tasks, "quantity": p.get("quantity", 1), "total_work": total_proc * p.get("quantity", 1) } # Calculate machine load machine_load = {m["id"]: 0 for m in machines} task_to_machines = {} for m in machines: for t in m.get("eligible_tasks", []): if t not in task_to_machines: task_to_machines[t] = [] task_to_machines[t].append(m["id"]) # Distribute load (simplified: split evenly among eligible machines) for p in products: qty = p.get("quantity", 1) for task_id in p.get("task_sequence", []): proc_time = task_times.get(task_id, 0) eligible = task_to_machines.get(task_id, []) if eligible: load_per_machine = (proc_time * qty) / len(eligible) for m_id in eligible: machine_load[m_id] += load_per_machine # Find bottleneck max_load = max(machine_load.values()) if machine_load else 0 bottleneck = [m for m, load in machine_load.items() if load == max_load] # Total work total_processing = sum(pm["total_work"] for pm in product_metrics.values()) total_jobs = sum(p.get("quantity", 1) for p in products) # Minimum transport time estimate min_transport_per_task = grid_distance / mhv_speed # Adjacent machine transport total_task_count = sum(pm["num_tasks"] * pm["quantity"] for p_id, pm in product_metrics.items()) # Each task except first needs transport min_transport = min_transport_per_task * (total_task_count - total_jobs) # Lower bound on makespan (bottleneck determines minimum) makespan_lower_bound = max_load if max_load > 0 else total_processing return { "product_analysis": product_metrics, "machine_load": machine_load, "bottleneck": { "machines": bottleneck, "load": max_load }, "totals": { "total_processing_time": total_processing, "total_jobs": total_jobs, "total_tasks": total_task_count, "estimated_min_transport": min_transport }, "bounds": { "makespan_lower_bound": makespan_lower_bound, "note": "Actual makespan will be higher due to transport, sequencing, and contention" } } except Exception as e: return { "error": str(e), "type": type(e).__name__ } @mcp.tool() async def analyze_mps_schedule( task_schedule: list, mhv_schedule: list, machines: list ) -> Dict[str, Any]: """ Analyze MPS simulation schedule results. Provides detailed analysis of task and transport schedules: - Gantt chart data - Machine utilization breakdown - MHV utilization breakdown - Idle time analysis Args: task_schedule: Task execution records from simulation mhv_schedule: Transport records from simulation machines: Machine definitions Returns: Detailed schedule analysis Example: analyze_mps_schedule( task_schedule=[ {"job": "A0", "task": "T1", "machine": "M1", "start_time": 0, "end_time": 20} ], mhv_schedule=[ {"mhv": "H1", "job": "A0", "from_machine": "M1", "to_machine": "M2", "depart_time": 20, "arrive_time": 35} ], machines=[{"id": "M1"}, {"id": "M2"}] ) """ try: if not task_schedule: return {"error": "No task schedule provided"} # Calculate makespan makespan = max(t["end_time"] for t in task_schedule) # Machine statistics machine_stats = {} for m in machines: m_id = m["id"] m_tasks = [t for t in task_schedule if t["machine"] == m_id] busy_time = sum(t["end_time"] - t["start_time"] for t in m_tasks) machine_stats[m_id] = { "tasks_processed": len(m_tasks), "busy_time": busy_time, "utilization": (busy_time / makespan * 100) if makespan > 0 else 0, "idle_time": makespan - busy_time } # MHV statistics mhv_stats = {} if mhv_schedule: mhv_ids = set(t["mhv"] for t in mhv_schedule) for mhv_id in mhv_ids: mhv_transports = [t for t in mhv_schedule if t["mhv"] == mhv_id] busy_time = sum(t["arrive_time"] - t["depart_time"] for t in mhv_transports) mhv_stats[mhv_id] = { "transports": len(mhv_transports), "busy_time": busy_time, "utilization": (busy_time / makespan * 100) if makespan > 0 else 0, "avg_transport_time": busy_time / len(mhv_transports) if mhv_transports else 0 } # Job completion analysis job_tasks = {} for t in task_schedule: job_id = t["job"] if job_id not in job_tasks: job_tasks[job_id] = [] job_tasks[job_id].append(t) job_stats = {} for job_id, tasks in job_tasks.items(): start = min(t["start_time"] for t in tasks) end = max(t["end_time"] for t in tasks) proc_time = sum(t["end_time"] - t["start_time"] for t in tasks) job_stats[job_id] = { "start_time": start, "completion_time": end, "flow_time": end - start, "processing_time": proc_time, "wait_time": (end - start) - proc_time } return { "makespan": makespan, "machine_statistics": machine_stats, "mhv_statistics": mhv_stats, "job_statistics": job_stats, "summary": { "total_tasks": len(task_schedule), "total_transports": len(mhv_schedule), "avg_machine_utilization": np.mean([s["utilization"] for s in machine_stats.values()]), "avg_mhv_utilization": np.mean([s["utilization"] for s in mhv_stats.values()]) if mhv_stats else 0, "avg_job_flow_time": np.mean([s["flow_time"] for s in job_stats.values()]) } } except Exception as e: return { "error": str(e), "type": type(e).__name__ } @mcp.tool() async def verify_mps_schedule( schedule: dict, config: Optional[dict] = None ) -> Dict[str, Any]: """ Verify a given MPS schedule and calculate its makespan. This tool validates a pre-defined schedule against MPS constraints and calculates the theoretical makespan. Use this to verify if LLM-generated code produces correct results. The schedule format includes: - jobs: List of job schedules with operations (task, machine, start, end) - transports: Optional list of MHV transport events - calculated_makespan: Expected makespan (ground truth) Constraints verified: 1. Task precedence (operations follow correct order) 2. Machine eligibility (task runs on valid machine) 3. Processing time (duration matches task definition) 4. Disjunctive (no machine overlap) Args: schedule: Schedule dictionary with jobs and operations config: Optional MPS configuration. Uses default if not provided. Returns: Dictionary with: - calculated_makespan: Makespan computed from schedule - expected_makespan: Expected makespan from schedule data - error_percent: Difference percentage - matches: True if matches (within 0.1% tolerance) - constraint_valid: True if all constraints satisfied - constraint_violations: List of any violations Example: verify_mps_schedule( schedule={ "jobs": [ { "job_id": "A0", "operations": [ {"task": "T1", "machine": "M1", "start": 0, "end": 20}, {"task": "T3", "machine": "M2", "start": 35, "end": 65}, {"task": "T4", "machine": "M2", "start": 65, "end": 95}, {"task": "T6", "machine": "M3", "start": 110, "end": 150} ] } ], "calculated_makespan": 150.0 } ) """ if not MPS_AVAILABLE: return { "error": "MPS simulation not available", "suggestion": "Check MPS module installation" } try: # Use the verification function from mps_simulation result = _verify_mps_schedule(schedule, config) return result except Exception as e: return { "error": str(e), "type": type(e).__name__ } @mcp.tool() async def get_mps_default_config() -> Dict[str, Any]: """ Get default MPS configuration (Base Setup from KTH thesis). Returns the standard 3×2 grid configuration with: - 6 machines (M1-M6) - 6 tasks (T1-T6) - 5 products (A-E) - 5 MHVs This matches the experimental setup from the KTH Master Thesis (2025). Returns: Complete MPS configuration dictionary """ return { "layout": { "rows": 3, "cols": 2, "machine_positions": { "M1": [0, 0], "M2": [0, 1], "M3": [1, 0], "M4": [1, 1], "M5": [2, 0], "M6": [2, 1] } }, "machines": [ {"id": "M1", "eligible_tasks": ["T1", "T2"]}, {"id": "M2", "eligible_tasks": ["T3", "T4"]}, {"id": "M3", "eligible_tasks": ["T5", "T6"]}, {"id": "M4", "eligible_tasks": ["T1", "T3"]}, {"id": "M5", "eligible_tasks": ["T2", "T5"]}, {"id": "M6", "eligible_tasks": ["T4", "T6"]} ], "tasks": [ {"id": "T1", "processing_time": 20}, {"id": "T2", "processing_time": 25}, {"id": "T3", "processing_time": 30}, {"id": "T4", "processing_time": 30}, {"id": "T5", "processing_time": 35}, {"id": "T6", "processing_time": 40} ], "products": [ {"id": "A", "task_sequence": ["T1", "T3", "T4", "T6"], "quantity": 1}, {"id": "B", "task_sequence": ["T1", "T2", "T4", "T5"], "quantity": 1}, {"id": "C", "task_sequence": ["T2", "T3", "T5", "T6"], "quantity": 1}, {"id": "D", "task_sequence": ["T6", "T5", "T1", "T3"], "quantity": 1}, {"id": "E", "task_sequence": ["T4", "T2", "T5", "T1"], "quantity": 1} ], "mhv": {"count": 5, "speed": 1.0}, "grid_distance": 15.0, "buffer_capacity": 2, "description": "Base MPS setup from KTH Master Thesis (2025)", "reference": "Tiruvallur Devendar, R. (2025). Feasibility Study of Scheduling in Matrix Production Systems Using Large Language Models" } # ============================================================================ # PROMPTS: Context Templates # ============================================================================ @mcp.prompt() async def generate_simulation_code( arrival_rate: float, service_rate: float, num_servers: int = 1, simulation_time: float = 10000.0, random_seed: int = 42 ) -> str: """ Generate complete SimPy simulation code Creates production-ready Python code for M/M/1 or M/M/c simulation. """ model_type = f"M/M/{num_servers}" if num_servers > 1 else "M/M/1" return f""" Generate complete, production-ready SimPy code for {model_type} queue simulation. Requirements: - Arrival rate: {arrival_rate} customers per time unit - Service rate: {service_rate} customers per time unit per server - Number of servers: {num_servers} - Simulation duration: {simulation_time} time units - Random seed: {random_seed} for reproducibility The code must: 1. Import necessary libraries (simpy, numpy, random) 2. Define a simulation class with proper initialization 3. Implement customer arrival process (Poisson) 4. Implement service process (exponential) 5. {"Handle multiple servers with shared queue" if num_servers > 1 else "Handle single server queue"} 6. Collect metrics: waiting times, queue lengths, utilization 7. Calculate and return performance metrics 8. Include proper error handling 9. Add comprehensive docstrings Output format: Complete Python code ready to execute. """ @mcp.prompt() async def explain_mm1_theory() -> str: """ Educational prompt about M/M/1 queuing theory """ return """ You are an expert in queuing theory. Explain M/M/1 queuing systems: 1. What does M/M/1 mean? 2. Key assumptions and when they apply 3. Important formulas with examples 4. Real-world applications 5. Limitations and when to use M/M/c instead Provide clear, educational content with numerical examples. """ @mcp.prompt() async def explain_mmc_theory() -> str: """ Educational prompt about M/M/c queuing theory """ return """ You are an expert in queuing theory. Explain M/M/c queuing systems: 1. **What is M/M/c?** - First M: Markovian (Poisson) arrivals with rate λ - Second M: Markovian (exponential) service with rate μ per server - c: Number of identical parallel servers 2. **Key Formulas:** - Utilization: ρ = λ/(c×μ) must be < 1 for stability - Erlang C formula for P(wait) - Average wait time: Wq 3. **M/M/1 vs M/M/c Comparison:** - M/M/1 × c: Separate queues, customers choose - M/M/c: Single queue, next customer to next free server - M/M/c is ALWAYS better (lower wait times, fair, balanced) 4. **Real-world Examples:** - Banks: Single queue → multiple tellers (M/M/c) - Supermarkets: Often separate queues (suboptimal) - Call centers: Automatic distribution (M/M/c) 5. **Why Pooled Queue Wins:** - Statistical multiplexing gain - No idle servers while customers wait elsewhere - Automatic load balancing - Fairness guaranteed Provide numerical examples showing the dramatic improvement. """ @mcp.prompt() async def analyze_results( simulation_metrics: Dict[str, Any], theoretical_metrics: Dict[str, Any] ) -> str: """ Analyze and interpret simulation results """ return f""" Analyze these M/M/1 simulation results: Simulation Metrics: {json.dumps(simulation_metrics, indent=2)} Theoretical Metrics: {json.dumps(theoretical_metrics, indent=2)} Provide: 1. Accuracy assessment (how close to theory?) 2. Key insights from the results 3. Any concerning deviations 4. Recommendations for improvement 5. Business implications Focus on practical interpretation. """ @mcp.prompt() async def debug_simulation(error_message: str) -> str: """ Debug simulation issues """ return f""" Help debug this M/M/1 simulation error: Error: {error_message} Common issues to check: 1. Arrival rate >= service rate (unstable) 2. Division by zero in metrics 3. Negative time values 4. Memory issues with long simulations 5. Random seed problems Provide specific debugging steps and solutions. """ # ============================================================================ # Main Entry Point # ============================================================================ def main(): """Run the MCP server""" import sys print("=" * 60) print("Simulation MCP Server v0.3.1") print("M/M/1 | M/M/c | MPS (Manufacturing)") print("=" * 60) print("\n📊 Available Tools:") print(" Queuing:") print(" - validate_config, calculate_metrics, run_simulation") print(" - run_mmc_simulation, compare_mm1_vs_mmc") print(" - analyze_cashier_problem, recommend_parameters") print(" Manufacturing (MPS v2.0 - KTH Thesis Definition):") print(" - run_mps_sim: Execute MPS simulation with MHV transport") print(" - calculate_mps_metrics: Machine load & bottleneck analysis") print(" - analyze_mps_schedule: Schedule analysis & utilization") print(" - get_mps_default_config: Default 3x2 grid configuration") print("\n📚 Available Resources:") print(" M/M/1: mm1://schema, mm1://parameters, mm1://metrics") print(" M/M/c: mmc://schema, mmc://formulas, mmc://comparison") print(" MPS: mps://schema, mps://parameters, mps://formulas") print(" mps://metrics, mps://examples, mps://guidelines") print("\n🚀 Server ready for queuing and manufacturing simulations!") print("=" * 60) # Check for HTTP mode (for Smithery) if "--http" in sys.argv or "MCP_TRANSPORT" in os.environ and os.environ["MCP_TRANSPORT"] == "http": # HTTP mode for Smithery import uvicorn from starlette.middleware.cors import CORSMiddleware # Get the Starlette app from FastMCP try: app = mcp.streamable_http_app() except AttributeError: # Fallback for older FastMCP versions app = mcp.app # Add CORS middleware for Smithery app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["GET", "POST", "OPTIONS"], allow_headers=["*"], ) host = os.environ.get("HOST", "0.0.0.0") port = int(os.environ.get("PORT", "8081")) print(f"\nStarting MCP server with HTTP transport on {host}:{port}") uvicorn.run(app, host=host, port=port) else: # Standard stdio mode print("\nStarting MCP server with stdio transport") mcp.run() if __name__ == "__main__": import os main()