create_simulation_from_usecase
Generate and run simulations based on specified use cases and uploaded files to analyze human and social behavior in digital spaces using Fujitsu Social Digital Twin MCP Server.
Instructions
指定されたユースケースとアップロードされたファイルからシミュレーションを作成・実行します。
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| ctx | No | ||
| simulation_params | Yes | ||
| uploaded_files | Yes | ||
| usecase | Yes |
Implementation Reference
- src/fujitsu_sdt_mcp/server.py:683-807 (handler)The main asynchronous handler function for the 'create_simulation_from_usecase' tool. It validates inputs, builds simulation configuration based on usecase and parameters, handles required files from uploads, and uses FujitsuSocialDigitalTwinClient to create and start the simulation.async def create_simulation_from_usecase(usecase: str, uploaded_files: Dict[str, str], simulation_params: Dict[str, Any], ctx: Optional[Context] = None) -> Dict[str, Any]: """指定されたユースケースとアップロードされたファイルからシミュレーションを作成・実行します。""" try: required_files = _get_required_files_for_usecase(usecase) missing_files = [f for f in required_files if f not in uploaded_files] if missing_files: return { "success": False, "error": f"必要なファイルが不足しています: {', '.join(missing_files)}" } # パラメータのバリデーション if "region" not in simulation_params: return {"success": False, "error": "地域名(region)が指定されていません"} if "start" not in simulation_params: return {"success": False, "error": "開始時間(start)が指定されていません"} if "end" not in simulation_params: return {"success": False, "error": "終了時間(end)が指定されていません"} # ユースケース固有のパラメータを設定 config = { "name": f"{usecase}_{simulation_params['region']}_{datetime.now().strftime('%Y%m%d_%H%M%S')}", "usecase": _convert_usecase_to_api_format(usecase), "region": simulation_params["region"], "start": int(simulation_params["start"]), "end": int(simulation_params["end"]), "vtypes": { "car": True, "pedestrian": True, "escooter": "e-scooter初期配備" == usecase, "bicycle": True, "pt": True } } # ユースケース固有のパラメータを追加 if usecase == "e-scooter初期配備": if "escooter_count" not in simulation_params: return {"success": False, "error": "eスクーター台数(escooter_count)が指定されていません"} config["escooter_count"] = int(simulation_params["escooter_count"]) if "escooter_fee" in simulation_params: config["escooter_fee"] = int(simulation_params["escooter_fee"]) elif usecase == "ダイナミックディスカウント": if "discount_percentage" not in simulation_params: return {"success": False, "error": "割引率(discount_percentage)が指定されていません"} config["discount_percentage"] = int(simulation_params["discount_percentage"]) if "discount_hours" in simulation_params: if isinstance(simulation_params["discount_hours"], list): config["discount_hours"] = simulation_params["discount_hours"] else: config["discount_hours"] = [int(h) for h in str(simulation_params["discount_hours"]).split(",")] elif usecase == "パーク&ライド": if "parking_capacity" not in simulation_params: return {"success": False, "error": "駐車場容量(parking_capacity)が指定されていません"} if "parking_fee" not in simulation_params: return {"success": False, "error": "駐車料金(parking_fee)が指定されていません"} config["parking"] = [ { "id": "parking1", "total": int(simulation_params["parking_capacity"]), "charge": int(simulation_params["parking_fee"]), "type": "car" } ] if "parking_locations" in simulation_params and isinstance(simulation_params["parking_locations"], list): for i, location in enumerate(simulation_params["parking_locations"]): if i >= len(config["parking"]): config["parking"].append({ "id": f"parking{i+1}", "total": int(simulation_params["parking_capacity"]), "charge": int(simulation_params["parking_fee"]), "type": "car" }) if isinstance(location, list) and len(location) == 2: config["parking"][i]["pos"] = location # ファイルパスをパラメータに追加 file_paths = {} for file_key, file_path in uploaded_files.items(): file_paths[file_key] = file_path config["input_files"] = file_paths # シミュレーションデータを作成 async with await get_http_client() as client: api_client = FujitsuSocialDigitalTwinClient(client) simdata_result = await api_client.create_simdata(config) if not simdata_result.get("success"): return simdata_result simdata_id = simdata_result.get("data", {}).get("id") # シミュレーションを開始 result = await api_client.start_simulation(simdata_id) if result.get("success"): return { "success": True, "usecase": usecase, "simulation_id": result.get("data", {}).get("id"), "simdata_id": simdata_id, "status": "started", "region": simulation_params["region"], "time_range": f"{simulation_params['start']}時-{simulation_params['end']}時", "config_summary": _get_config_summary(config) } else: return result except Exception as e: logger.error(f"Error creating simulation from usecase: {e}") return format_api_error(500, str(e))
- Helper function that returns the list of required input files specific to the given usecase.def _get_required_files_for_usecase(usecase: str) -> list[str]: """ユースケースごとに必要なファイルのリストを返す""" required_files = { "e-scooter初期配備": ["map.net.xml.gz", "station.csv", "od.csv.gz", "precondition.json"], "ダイナミックディスカウント": ["map.net.xml.gz", "station.csv", "od.csv.gz", "precondition.json"], "道路閉鎖": ["map.net.xml.gz", "station.csv", "od.csv.gz", "road-closure.csv", "precondition.json"], "ロードプライシング": ["map.net.xml.gz", "station.csv", "od.csv.gz", "road-closure.csv", "road-pricing.csv", "precondition.json"], "パーク&ライド": ["map.net.xml.gz", "station.csv", "od.csv.gz", "parking.csv", "precondition.json"] } return required_files.get(usecase, [])
- Helper function that maps the usecase name to the corresponding API format string.def _convert_usecase_to_api_format(usecase: str) -> str: """ユースケース名をAPI形式に変換""" mapping = { "e-scooter初期配備": "escooter-placement", "ダイナミックディスカウント": "dynamic-discount", "道路閉鎖": "road-closure", "ロードプライシング": "road-pricing", "パーク&ライド": "park-and-ride" } return mapping.get(usecase, "generic")
- Helper function that generates a textual summary of the simulation configuration for the response.def _get_config_summary(config: Dict[str, Any]) -> str: """設定の要約を生成""" summary = [] if "name" in config: summary.append(f"名前: {config['name']}") if "usecase" in config: summary.append(f"ユースケース: {config['usecase']}") if "region" in config: summary.append(f"地域: {config['region']}") if "start" in config and "end" in config: summary.append(f"時間範囲: {config['start']}時-{config['end']}時") if "escooter_count" in config: summary.append(f"eスクーター台数: {config['escooter_count']}台") if "escooter_fee" in config: summary.append(f"eスクーター料金: {config['escooter_fee']}円") if "discount_percentage" in config: summary.append(f"割引率: {config['discount_percentage']}%") if "discount_hours" in config: summary.append(f"割引時間帯: {', '.join(map(str, config['discount_hours']))}時") if "parking" in config and config["parking"]: parking = config["parking"][0] summary.append(f"駐車場容量: {parking.get('total', 0)}台") summary.append(f"駐車料金: {parking.get('charge', 0)}円") return "\n".join(summary)
- src/fujitsu_sdt_mcp/server.py:682-682 (registration)The @mcp.tool() decorator registers this function as an MCP tool.@mcp.tool()