🚀
بصيـر API
مستقبل القيادة الذاتية بين يديك
{status_text}
{active_sessions_count}
الجلسات النشطة
"""
return HTMLResponse(content=html_content)
@app.post("/start_session", summary="Start a new driving session", tags=["Session Management"])
def start_session():
session_id = str(uuid.uuid4())
# 1. الحصول على الإعدادات الكاملة من المصدر الوحيد
config = get_master_config()
# 2. استخراج الإعدادات المطلوبة لكل مكون بشكل صريح
grid_conf = config['grid_conf']
controller_params = config['controller_params']
simulation_freq = config['simulation']['frequency']
# 3. تهيئة المتتبع (Tracker) بمعلماته المحددة
tracker = Tracker(
grid_conf=grid_conf,
match_threshold=controller_params.get('tracker_match_thresh', 2.5),
prune_age=controller_params.get('tracker_prune_age', 5)
)
# 4. تهيئة المتحكم (Controller)
controller = InterfuserController({
'controller_params': controller_params,
'grid_conf': grid_conf,
'frequency': simulation_freq
})
# 5. إنشاء الجلسة بالكائنات المهيأة
SESSIONS[session_id] = {
'tracker': tracker,
'controller': controller,
'frame_num': 0
}
logging.info(f"New session started: {session_id}")
return {"session_id": session_id}
@app.post("/run_step", response_model=RunStepResponse, summary="Process a single simulation step", tags=["Core"])
@torch.no_grad()
def run_step(request: RunStepRequest):
if MODEL is None:
raise HTTPException(status_code=503, detail="Model is not available.")
session = SESSIONS.get(request.session_id)
if not session:
raise HTTPException(status_code=404, detail="Session ID not found.")
# --- 1. الإدراك (Perception) ---
image = b64_to_cv2(request.image_b64)
model_input = prepare_model_input(image, request.measurements)
traffic, waypoints, junc, light, stop, _ = MODEL(model_input)
# --- 2. معالجة مخرجات النموذج ---
traffic_processed = torch.cat([torch.sigmoid(traffic[0][:, 0:1]), traffic[0][:, 1:]], dim=1)
traffic_np = traffic_processed.cpu().numpy().reshape(20, 20, -1)
waypoints_np = waypoints[0].cpu().numpy()
junction_prob = torch.softmax(junc, dim=1)[0, 1].item()
light_prob = torch.softmax(light, dim=1)[0, 1].item()
stop_prob = torch.softmax(stop, dim=1)[0, 1].item()
# --- 3. التتبع والتحكم ---
ego_pos = np.array(request.measurements.pos_global)
ego_theta = request.measurements.theta
frame_num = session['frame_num']
active_tracks = session['tracker'].process_frame(traffic_np, ego_pos, ego_theta, frame_num)
steer, throttle, brake, ctrl_info = session['controller'].run_step(
speed=request.measurements.speed, waypoints=torch.from_numpy(waypoints_np),
junction=junction_prob, traffic_light=light_prob, stop_sign=stop_prob,
bev_map=traffic_np, ego_pos=ego_pos, ego_theta=ego_theta, frame_num=frame_num
)
# --- 4. إنشاء الواجهة المرئية ---
display_iface = DisplayInterface(DisplayConfig(width=1600, height=900))
bev_maps = render_bev(active_tracks, waypoints_np, ego_pos, ego_theta)
display_data = {
'camera_view': image, 'map_t0': bev_maps['t0'], 'map_t1': bev_maps['t1'], 'map_t2': bev_maps['t2'],
'frame_num': frame_num, 'speed': request.measurements.speed * 3.6,
'target_speed': ctrl_info.get('target_speed', 0) * 3.6,
'steer': steer, 'throttle': throttle, 'brake': brake,
'light_prob': light_prob, 'stop_prob': stop_prob,
'object_counts': {'car': len(active_tracks)}
}
dashboard = display_iface.run_interface(display_data)
# --- 5. تحديث الجلسة وإرجاع الرد ---
session['frame_num'] += 1
return RunStepResponse(
control_commands=ControlCommands(steer=steer, throttle=throttle, brake=brake),
scene_analysis=SceneAnalysis(is_junction=junction_prob, traffic_light_state=light_prob, stop_sign=stop_prob),
predicted_waypoints=[tuple(wp) for wp in waypoints_np.tolist()],
dashboard_b64=cv2_to_b64(dashboard),
reason=ctrl_info.get('brake_reason', "Cruising")
)
@app.post("/end_session", summary="End and clean up a session", tags=["Session Management"])
def end_session(session_id: str):
if session_id in SESSIONS:
del SESSIONS[session_id]
logging.info(f"Session ended: {session_id}")
return {"message": f"Session {session_id} ended."}
raise HTTPException(status_code=404, detail="Session not found.")
# ================== تشغيل الخادم ==================
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=7860)