dsh/dual_camera_system/test_ptz.py

#!/usr/bin/env python3
"""
PTZ 控制测试脚本
用于验证球机 PTZ 控制是否正常工作
"""

import os
import sys
import time

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

from config import PTZ_CAMERA, SDK_PATH, PTZ_CONFIG
from dahua_sdk import DahuaSDK, PTZCommand
from ptz_camera import PTZCamera


def test_ptz_basic_control(ptz: PTZCamera):
    """测试基本 PTZ 控制"""
    print("\n" + "=" * 50)
    print("测试基本 PTZ 控制")
    print("=" * 50)
    
    # 测试向左移动
    print("\n[测试1] 向左移动 2 秒...")
    ptz.move_left(speed=4)
    time.sleep(2)
    ptz.stop_move()
    print("停止移动")
    time.sleep(1)
    
    # 测试向右移动
    print("\n[测试2] 向右移动 2 秒...")
    ptz.move_right(speed=4)
    time.sleep(2)
    ptz.stop_move()
    print("停止移动")
    time.sleep(1)
    
    # 测试放大
    print("\n[测试3] 放大 2 秒...")
    ptz.zoom_in(speed=4)
    time.sleep(2)
    ptz.stop_move()
    print("停止变倍")
    time.sleep(1)
    
    # 测试缩小
    print("\n[测试4] 缩小 2 秒...")
    ptz.zoom_out(speed=4)
    time.sleep(2)
    ptz.stop_move()
    print("停止变倍")


def test_ptz_exact_goto(ptz: PTZCamera):
    """测试精确定位"""
    print("\n" + "=" * 50)
    print("测试精确定位 (EXACTGOTO)")
    print("=" * 50)
    
    # 测试不同的 pan 角度
    test_points = [
        (0, 0, 1),      # pan=0°
        (45, 0, 1),     # pan=45°
        (90, 0, 1),     # pan=90°
        (135, 0, 1),    # pan=135°
        (180, 0, 1),    # pan=180°
        (270, 0, 1),    # pan=270°
        (0, 0, 1),      # 返回 pan=0°
    ]
    
    for i, (pan, tilt, zoom) in enumerate(test_points):
        print(f"\n[测试点 {i+1}/{len(test_points)}] 移动到 pan={pan}°, tilt={tilt}°, zoom={zoom}")
        result = ptz.goto_exact_position(pan, tilt, zoom)
        if result:
            print("  ✓ 命令成功")
        else:
            print("  ✗ 命令失败")
        time.sleep(3)  # 等待球机移动


def test_ptz_fast_goto(ptz: PTZCamera, sdk: DahuaSDK):
    """测试快速定位 (FASTGOTO)"""
    print("\n" + "=" * 50)
    print("测试快速定位 (FASTGOTO)")
    print("=" * 50)
    print("FASTGOTO 参数格式: 水平(-8191~8191), 垂直(-8191~8191), 变倍(-16~16)")
    
    # FASTGOTO 使用不同的参数格式
    # 参数范围: 水平/垂直 -8191~8191, 变倍 -16~16
    test_points = [
        (0, 0, 0),       # 中心
        (-4000, 0, 0),   # 向左
        (4000, 0, 0),    # 向右
        (0, -2000, 0),   # 向上
        (0, 2000, 0),    # 向下
        (0, 0, 0),       # 返回中心
    ]
    
    for i, (p1, p2, p3) in enumerate(test_points):
        print(f"\n[测试点 {i+1}/{len(test_points)}] FASTGOTO: p1={p1}, p2={p2}, p3={p3}")
        result = sdk.ptz_control(
            ptz.login_handle,
            ptz.config['channel'],
            PTZCommand.FASTGOTO,
            p1, p2, p3, False
        )
        if result:
            print("  ✓ 命令成功")
        else:
            print("  ✗ 命令失败")
        time.sleep(3)


def interactive_test(ptz: PTZCamera, sdk: DahuaSDK):
    """交互式测试"""
    print("\n" + "=" * 50)
    print("交互式测试模式")
    print("=" * 50)
    print("命令:")
    print("  left/right/up/down - 基本移动")
    print("  zoomin/zoomout - 变倍")
    print("  stop - 停止")
    print("  goto <pan> <tilt> [zoom] - 精确定位 (角度)")
    print("  fast <p1> <p2> <p3> - 快速定位")
    print("  q - 退出")
    print("=" * 50)
    
    while True:
        try:
            cmd = input("\n> ").strip().lower()
            if not cmd:
                continue
            
            if cmd == 'q':
                break
            
            elif cmd == 'left':
                ptz.move_left(speed=4)
                print("向左移动中... (输入 'stop' 停止)")
            
            elif cmd == 'right':
                ptz.move_right(speed=4)
                print("向右移动中... (输入 'stop' 停止)")
            
            elif cmd == 'up':
                ptz.move_up(speed=4)
                print("向上移动中... (输入 'stop' 停止)")
            
            elif cmd == 'down':
                ptz.move_down(speed=4)
                print("向下移动中... (输入 'stop' 停止)")
            
            elif cmd == 'zoomin':
                ptz.zoom_in(speed=4)
                print("放大中... (输入 'stop' 停止)")
            
            elif cmd == 'zoomout':
                ptz.zoom_out(speed=4)
                print("缩小中... (输入 'stop' 停止)")
            
            elif cmd == 'stop':
                ptz.stop_move()
                print("已停止")
            
            elif cmd.startswith('goto'):
                parts = cmd.split()
                if len(parts) >= 3:
                    pan = float(parts[1])
                    tilt = float(parts[2])
                    zoom = int(parts[3]) if len(parts) > 3 else 1
                    print(f"移动到 pan={pan}°, tilt={tilt}°, zoom={zoom}")
                    ptz.goto_exact_position(pan, tilt, zoom)
                else:
                    print("用法: goto <pan> <tilt> [zoom]")
            
            elif cmd.startswith('fast'):
                parts = cmd.split()
                if len(parts) >= 4:
                    p1 = int(parts[1])
                    p2 = int(parts[2])
                    p3 = int(parts[3])
                    print(f"FASTGOTO: p1={p1}, p2={p2}, p3={p3}")
                    sdk.ptz_control(
                        ptz.login_handle,
                        ptz.config['channel'],
                        PTZCommand.FASTGOTO,
                        p1, p2, p3, False
                    )
                else:
                    print("用法: fast <p1> <p2> <p3>")
            
            else:
                print("未知命令")
                
        except KeyboardInterrupt:
            break
        except Exception as e:
            print(f"错误: {e}")


def test_channels(sdk: DahuaSDK):
    """测试不同通道号的 PTZ 控制"""
    print("\n" + "=" * 50)
    print("测试不同通道号")
    print("=" * 50)
    
    config = PTZ_CAMERA.copy()
    
    # 连接设备
    login_handle, error = sdk.login(
        config['ip'], config['port'],
        config['username'], config['password']
    )
    
    if login_handle is None:
        print(f"连接失败: 错误码={error}")
        return
    
    print(f"连接成功: handle={login_handle}")
    
    # 测试通道 0-3
    for channel in range(4):
        print(f"\n--- 测试通道 {channel} ---")
        
        # 测试 LEFT 命令
        print(f"  测试 LEFT 命令...")
        result = sdk.ptz_control(login_handle, channel, PTZCommand.LEFT, 0, 4, 0, False)
        if result:
            print(f"  通道 {channel}: LEFT 成功 ✓")
            time.sleep(1)
            # 发送停止
            sdk.ptz_control(login_handle, channel, PTZCommand.LEFT, 0, 0, 0, True)
        else:
            print(f"  通道 {channel}: LEFT 失败 ✗")
        
        time.sleep(0.5)
        
        # 测试 EXACTGOTO
        print(f"  测试 EXACTGOTO...")
        result = sdk.ptz_control(login_handle, channel, PTZCommand.EXACTGOTO, 900, 0, 1, False)
        if result:
            print(f"  通道 {channel}: EXACTGOTO 成功 ✓")
        else:
            print(f"  通道 {channel}: EXACTGOTO 失败 ✗")
        
        time.sleep(1)
    
    sdk.logout(login_handle)
    print("\n通道测试完成")


def main():
    print("PTZ 控制测试脚本")
    print("=" * 50)
    
    # 初始化 SDK
    sdk_path = os.path.join(SDK_PATH['lib_path'], SDK_PATH['netsdk'])
    print(f"加载 SDK: {sdk_path}")
    
    sdk = DahuaSDK(sdk_path)
    if not sdk.init():
        print("SDK 初始化失败!")
        return 1
    
    print("SDK 初始化成功")
    
    # 询问是否先测试通道
    print("\n是否先测试不同通道号? (y/n): ", end="")
    if input().strip().lower() == 'y':
        test_channels(sdk)
        sdk.cleanup()
        return 0
    
    # 连接球机
    ptz = PTZCamera(sdk, PTZ_CAMERA)
    if not ptz.connect():
        print("连接球机失败!")
        sdk.cleanup()
        return 1
    
    print("连接球机成功")
    
    try:
        # 运行测试
        print("\n选择测试模式:")
        print("  1 - 基本控制测试 (左/右/上/下/变倍)")
        print("  2 - 精确定位测试 (EXACTGOTO)")
        print("  3 - 快速定位测试 (FASTGOTO)")
        print("  4 - 交互式测试")
        print("  5 - 全部测试")
        
        choice = input("\n请选择 (1-5): ").strip()
        
        if choice == '1':
            test_ptz_basic_control(ptz)
        elif choice == '2':
            test_ptz_exact_goto(ptz)
        elif choice == '3':
            test_ptz_fast_goto(ptz, sdk)
        elif choice == '4':
            interactive_test(ptz, sdk)
        elif choice == '5':
            test_ptz_basic_control(ptz)
            test_ptz_exact_goto(ptz)
            test_ptz_fast_goto(ptz, sdk)
        else:
            print("无效选择")
        
    finally:
        print("\n断开连接...")
        ptz.disconnect()
        sdk.cleanup()
    
    return 0


if __name__ == '__main__':
    sys.exit(main() or 0)