157 lines
5.7 KiB
Python
157 lines
5.7 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Скрипт для отправки тестового JPEG-кадра на сервер.
|
||
Автоматически создает комнату и отправляет видеокадры через WebSocket.
|
||
"""
|
||
|
||
import asyncio
|
||
import websockets
|
||
import json
|
||
import sys
|
||
import cv2
|
||
import numpy as np
|
||
import requests
|
||
from pathlib import Path
|
||
from dotenv import load_dotenv
|
||
import os
|
||
|
||
# Загружаем переменные окружения из .env
|
||
load_dotenv()
|
||
|
||
# Параметры подключения
|
||
HOST = os.getenv("PUBLIC_HOST", "192.168.0.112")
|
||
PORT = int(os.getenv("PORT", 8000))
|
||
FRAME_FILE = "test_frame.jpg"
|
||
FRAMES_TO_SEND = 300 # 10 сек при 30 FPS
|
||
|
||
# Учетные данные админа (из server.py)
|
||
ADMIN_USERNAME = "admin"
|
||
ADMIN_PASSWORD = "admin123"
|
||
|
||
async def send_frames():
|
||
"""Подключиться к WebSocket и отправить кадры"""
|
||
|
||
# 0. Логинимся как админ чтобы получить сессию
|
||
print(f"[Test] Logging in as admin...")
|
||
session = requests.Session()
|
||
try:
|
||
login_response = session.post(
|
||
f"http://{HOST}:{PORT}/login",
|
||
data={
|
||
"username": ADMIN_USERNAME,
|
||
"password": ADMIN_PASSWORD
|
||
}
|
||
)
|
||
if login_response.status_code != 200:
|
||
print(f"[Test] ❌ Failed to login: {login_response.status_code}")
|
||
print(f"[Test] Response: {login_response.text}")
|
||
return
|
||
print(f"[Test] ✓ Admin logged in")
|
||
except Exception as e:
|
||
print(f"[Test] ❌ Error during login: {e}")
|
||
return
|
||
|
||
# 1. Создаём новую комнату на сервере
|
||
print(f"[Test] Creating room on {HOST}:{PORT}...")
|
||
try:
|
||
response = session.post(
|
||
f"http://{HOST}:{PORT}/api/create-room",
|
||
data={
|
||
"name": "Test Room",
|
||
"password": "testpass",
|
||
"max_connections": "10"
|
||
}
|
||
)
|
||
if response.status_code != 200:
|
||
print(f"[Test] ❌ Failed to create room: {response.status_code}")
|
||
print(f"[Test] Response: {response.text}")
|
||
return
|
||
|
||
room_data = response.json()
|
||
if not room_data.get("success"):
|
||
print(f"[Test] ❌ Room creation failed: {room_data.get('error')}")
|
||
return
|
||
|
||
room_id = room_data.get("room", {}).get("id")
|
||
password = room_data.get("room", {}).get("password", "testpass")
|
||
print(f"[Test] ✓ Room created: {room_id}")
|
||
except Exception as e:
|
||
print(f"[Test] ❌ Error creating room: {e}")
|
||
return
|
||
|
||
# 2. Подключаемся к WebSocket с этой комнатой
|
||
uri = f"ws://{HOST}:{PORT}/ws/client/{room_id}/{password}"
|
||
print(f"[Test] Connecting to: {uri}")
|
||
|
||
try:
|
||
async with websockets.connect(uri) as ws:
|
||
# Получаем ответ сервера
|
||
response = await asyncio.wait_for(ws.recv(), timeout=5)
|
||
print(f"[Test] ✓ Server response: {response}")
|
||
|
||
# Читаем тестовый кадр
|
||
if not Path(FRAME_FILE).exists():
|
||
print(f"[Test] File not found: {FRAME_FILE}")
|
||
print(f"[Test] Creating a proper JPEG test file with OpenCV...")
|
||
if not create_minimal_jpeg(FRAME_FILE):
|
||
print(f"[Test] ❌ Failed to create test JPEG")
|
||
return
|
||
|
||
with open(FRAME_FILE, "rb") as f:
|
||
frame_data = f.read()
|
||
|
||
print(f"[Test] Frame size: {len(frame_data)} bytes")
|
||
|
||
# Отправляем кадры в цикле (30 FPS, ~10 секунд)
|
||
print(f"[Test] Sending frames (30 FPS for ~{FRAMES_TO_SEND // 30} seconds)...")
|
||
for i in range(FRAMES_TO_SEND): # ~10 sec * 30 FPS
|
||
await ws.send(frame_data)
|
||
if (i + 1) % 30 == 0:
|
||
print(f"[Test] ✓ Sent {i + 1} frames")
|
||
await asyncio.sleep(1 / 30) # ~30 FPS
|
||
|
||
print(f"[Test] ✓ Done! Sent {FRAMES_TO_SEND} frames")
|
||
await asyncio.sleep(2) # Пауза перед отключением
|
||
|
||
except asyncio.TimeoutError:
|
||
print(f"[Test] ❌ Timeout waiting for server response")
|
||
except ConnectionRefusedError:
|
||
print(f"[Test] ❌ Connection refused. Check HOST/PORT and firewall")
|
||
except Exception as e:
|
||
print(f"[Test] ❌ Error: {type(e).__name__}: {e}")
|
||
|
||
def create_minimal_jpeg(filename):
|
||
"""Создаёт корректный JPEG-файл (640x480 с синим фоном) для теста"""
|
||
# Создаём пустой образ 640x480 (синий)
|
||
image = np.zeros((480, 640, 3), dtype=np.uint8)
|
||
image[:, :] = (255, 0, 0) # BGR: синий цвет
|
||
|
||
# Добавляем текст
|
||
cv2.putText(image, "Test Frame", (260, 240),
|
||
cv2.FONT_HERSHEY_SIMPLEX, 1, (255, 255, 255), 2)
|
||
|
||
# Кодируем в JPEG
|
||
success, jpeg_data = cv2.imencode('.jpg', image)
|
||
if not success:
|
||
print(f"[Test] ❌ Failed to create JPEG")
|
||
return False
|
||
|
||
# Сохраняем в файл
|
||
with open(filename, "wb") as f:
|
||
f.write(jpeg_data.tobytes())
|
||
|
||
print(f"[Test] ✓ Created test JPEG: {filename} ({len(jpeg_data.tobytes())} bytes)")
|
||
return True
|
||
|
||
if __name__ == "__main__":
|
||
print("=" * 60)
|
||
print("WebSocket Frame Sender Test")
|
||
print("=" * 60)
|
||
print(f"Server: http://{HOST}:{PORT}")
|
||
print()
|
||
|
||
try:
|
||
asyncio.run(send_frames())
|
||
except KeyboardInterrupt:
|
||
print("\n[Test] Interrupted by user")
|