File size: 1,590 Bytes
ad1357a |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 |
# backend_api.py
# 后端API交互相关
import requests
import uuid
import json
from typing import Optional
from config import API_ENDPOINTS
def submit_to_backend(scene: str, prompt: str, mode: str, model_type: str, user: str = "Gradio-user") -> dict:
job_id = str(uuid.uuid4())
data = {
"model_type": model_type,
"instruction": prompt,
"episode_type": scene,
"mode": mode,
}
payload = {
"user": user,
"task": "robot_navigation",
"job_id": job_id,
"data": json.dumps(data)
}
try:
headers = {"Content-Type": "application/json"}
response = requests.post(
API_ENDPOINTS["submit_task"],
json=payload,
headers=headers,
timeout=200
)
return response.json()
except Exception as e:
return {"status": "error", "message": str(e)}
def get_task_status(task_id: str) -> dict:
try:
response = requests.get(f"{API_ENDPOINTS['query_status']}/{task_id}", timeout=5)
try:
return response.json()
except json.JSONDecodeError:
return {"status": "error", "message": response.text}
except Exception as e:
return {"status": "error", "message": str(e)}
def get_task_result(task_id: str) -> Optional[dict]:
try:
response = requests.get(
f"{API_ENDPOINTS['get_result']}/{task_id}",
timeout=5
)
return response.json()
except Exception as e:
return None
|