145 lines
4.1 KiB
Python
145 lines
4.1 KiB
Python
from flask import Flask, render_template, request, jsonify
|
|
from flask_socketio import SocketIO, emit
|
|
import logging
|
|
import math
|
|
from gimbal import gimbal_ctrl
|
|
|
|
# 配置日志
|
|
logging.basicConfig(level=logging.INFO)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Flask 应用初始化
|
|
app = Flask(__name__)
|
|
app.config['SECRET_KEY'] = 'h2Ms4pw9GzvwiFHyNPhH' # 请更换为安全的密钥
|
|
# socketio = SocketIO(app, cors_allowed_origins="*")
|
|
socketio = SocketIO(app)
|
|
|
|
# 全局变量存储摇杆数据
|
|
joystick_data = {}
|
|
|
|
ptz = gimbal_ctrl("192.168.17.114", 49125)
|
|
|
|
@app.route('/')
|
|
def index():
|
|
return render_template('index.html')
|
|
|
|
@socketio.on('joystick_data')
|
|
def handle_joystick_data(data):
|
|
global joystick_data
|
|
joystick_data = data
|
|
# logger.info(f"收到摇杆数据：{data}")
|
|
|
|
force = data.get('force', 0)
|
|
force = force if force <= 1 else 1
|
|
|
|
degree = data.get('degree', 0)
|
|
|
|
x = math.cos(math.radians(degree)) * force
|
|
y = math.sin(math.radians(degree)) * force
|
|
|
|
# print(f"force: {force} x: {x}, y: {y}")
|
|
ptz.set_pitch_speed(int(y * 100))
|
|
ptz.set_yaw_speed(int(x * -100))
|
|
|
|
# direction = data.get('direction', '')
|
|
# logger.info(f"收到摇杆数据：{direction}")
|
|
# if direction:
|
|
# handle_robot_movement(direction, data.get('force', 0))
|
|
|
|
# emit('data_received', {'status': 'success', 'timestamp': data.get('timestamp')})
|
|
|
|
# def handle_robot_movement(direction, force):
|
|
# """
|
|
# 根据摇杆方向和力度控制机器人
|
|
# 这里是示例实现，您可以根据实际需求修改
|
|
# """
|
|
# # 示例：根据方向执行不同的动作
|
|
# actions = {
|
|
# 'n': '向前移动',
|
|
# 's': '向后移动',
|
|
# 'w': '向左移动',
|
|
# 'e': '向右移动',
|
|
# 'nw': '左前移动',
|
|
# 'ne': '右前移动',
|
|
# 'sw': '左后移动',
|
|
# 'se': '右后移动'
|
|
# }
|
|
|
|
# action = actions.get(direction, '停止移动')
|
|
# speed = int(force * 100) if force else 0
|
|
|
|
# logger.info(f"执行动作：{action}, 速度：{speed}%")
|
|
|
|
@socketio.on('connect')
|
|
def handle_connect():
|
|
"""处理客户端连接"""
|
|
logger.info(f"客户端已连接：{request.sid}")
|
|
emit('connection_response', {'status': 'connected', 'message': 'WebSocket 连接成功'})
|
|
|
|
@socketio.on('disconnect')
|
|
def handle_disconnect():
|
|
"""处理客户端断开连接"""
|
|
logger.info(f"客户端已断开连接：{request.sid}")
|
|
|
|
@socketio.on('ping')
|
|
def handle_ping():
|
|
"""处理心跳检测"""
|
|
emit('pong')
|
|
|
|
@socketio.on('gimbal_center')
|
|
def handle_gimbal_center():
|
|
"""处理云台中心"""
|
|
ptz.center()
|
|
logger.info(f"云台回中")
|
|
|
|
@socketio.on('gimbal_pip')
|
|
def handle_gimbal_pip(data):
|
|
"""处理云台PIP"""
|
|
mode = int(data)
|
|
if mode >= 0 and mode <= 3:
|
|
ptz.set_pip_mode(mode)
|
|
logger.info(f"云台PIP {mode}")
|
|
|
|
@socketio.on('gimbal_ir_mode')
|
|
def handle_gimbal_ir_mode(data):
|
|
"""处理云台IR"""
|
|
mode = int(data)
|
|
if mode >= 0 and mode <= 9:
|
|
ptz.set_ir_mode(mode)
|
|
logger.info(f"云台IR {mode}")
|
|
|
|
@socketio.on('gimbal_zoom_in')
|
|
def handle_gimbal_zoom_in():
|
|
"""处理云台缩放"""
|
|
ptz.zoom_in()
|
|
|
|
@socketio.on('gimbal_zoom_out')
|
|
def handle_gimbal_zoom_out():
|
|
"""处理云台缩放"""
|
|
ptz.zoom_out()
|
|
|
|
|
|
# @app.route('/get_joystick_data')
|
|
# def get_joystick_data():
|
|
# """API 端点：获取当前摇杆数据"""
|
|
# from flask import jsonify
|
|
# return jsonify(joystick_data)
|
|
|
|
# @app.route('/control/<action>')
|
|
# def robot_control(action):
|
|
# """
|
|
# 示例控制端点
|
|
# 可用于直接控制机器人动作
|
|
# """
|
|
# from flask import jsonify
|
|
# valid_actions = ['forward', 'backward', 'left', 'right', 'stop']
|
|
|
|
# if action in valid_actions:
|
|
# logger.info(f"执行机器人控制：{action}")
|
|
# # 在这里添加实际的机器人控制逻辑
|
|
# return jsonify({'status': 'success', 'action': action})
|
|
# else:
|
|
# return jsonify({'status': 'error', 'message': '无效的动作'}), 400
|
|
|
|
if __name__ == '__main__':
|
|
socketio.run(app, host='10.21.31.250', port=5000, debug=True) |