Chore: rename api->camera-management-api and web->management-dashboard-web-app; update compose, ignore, README references
This commit is contained in:
Alireza Vaezi
267
camera-management-api/tests/recording/test_auto_recording.py
Normal file
267
camera-management-api/tests/recording/test_auto_recording.py
Normal file
@@ -0,0 +1,267 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Test script for auto-recording functionality.
|
||||
|
||||
This script tests the auto-recording feature by simulating MQTT state changes
|
||||
and verifying that cameras start and stop recording automatically.
|
||||
"""
|
||||
|
||||
import sys
|
||||
import os
|
||||
import time
|
||||
import json
|
||||
import requests
|
||||
from datetime import datetime
|
||||
|
||||
# Add the parent directory to Python path
|
||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
|
||||
from usda_vision_system.core.config import Config
|
||||
from usda_vision_system.core.state_manager import StateManager
|
||||
from usda_vision_system.core.events import EventSystem, publish_machine_state_changed
|
||||
|
||||
|
||||
class AutoRecordingTester:
|
||||
"""Test class for auto-recording functionality"""
|
||||
|
||||
def __init__(self):
|
||||
self.api_base_url = "http://localhost:8000"
|
||||
self.config = Config("config.json")
|
||||
self.state_manager = StateManager()
|
||||
self.event_system = EventSystem()
|
||||
|
||||
# Test results
|
||||
self.test_results = []
|
||||
|
||||
def log_test(self, test_name: str, success: bool, message: str = ""):
|
||||
"""Log a test result"""
|
||||
status = "✅ PASS" if success else "❌ FAIL"
|
||||
timestamp = datetime.now().strftime("%H:%M:%S")
|
||||
result = f"[{timestamp}] {status} {test_name}"
|
||||
if message:
|
||||
result += f" - {message}"
|
||||
print(result)
|
||||
|
||||
self.test_results.append({
|
||||
"test_name": test_name,
|
||||
"success": success,
|
||||
"message": message,
|
||||
"timestamp": timestamp
|
||||
})
|
||||
|
||||
def check_api_available(self) -> bool:
|
||||
"""Check if the API server is available"""
|
||||
try:
|
||||
response = requests.get(f"{self.api_base_url}/cameras", timeout=5)
|
||||
return response.status_code == 200
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
def get_camera_status(self, camera_name: str) -> dict:
|
||||
"""Get camera status from API"""
|
||||
try:
|
||||
response = requests.get(f"{self.api_base_url}/cameras", timeout=5)
|
||||
if response.status_code == 200:
|
||||
cameras = response.json()
|
||||
return cameras.get(camera_name, {})
|
||||
except Exception as e:
|
||||
print(f"Error getting camera status: {e}")
|
||||
return {}
|
||||
|
||||
def get_auto_recording_status(self) -> dict:
|
||||
"""Get auto-recording manager status"""
|
||||
try:
|
||||
response = requests.get(f"{self.api_base_url}/auto-recording/status", timeout=5)
|
||||
if response.status_code == 200:
|
||||
return response.json()
|
||||
except Exception as e:
|
||||
print(f"Error getting auto-recording status: {e}")
|
||||
return {}
|
||||
|
||||
def enable_auto_recording(self, camera_name: str) -> bool:
|
||||
"""Enable auto-recording for a camera"""
|
||||
try:
|
||||
response = requests.post(f"{self.api_base_url}/cameras/{camera_name}/auto-recording/enable", timeout=5)
|
||||
return response.status_code == 200
|
||||
except Exception as e:
|
||||
print(f"Error enabling auto-recording: {e}")
|
||||
return False
|
||||
|
||||
def disable_auto_recording(self, camera_name: str) -> bool:
|
||||
"""Disable auto-recording for a camera"""
|
||||
try:
|
||||
response = requests.post(f"{self.api_base_url}/cameras/{camera_name}/auto-recording/disable", timeout=5)
|
||||
return response.status_code == 200
|
||||
except Exception as e:
|
||||
print(f"Error disabling auto-recording: {e}")
|
||||
return False
|
||||
|
||||
def simulate_machine_state_change(self, machine_name: str, state: str):
|
||||
"""Simulate a machine state change via event system"""
|
||||
print(f"🔄 Simulating machine state change: {machine_name} -> {state}")
|
||||
publish_machine_state_changed(machine_name, state, "test_script")
|
||||
|
||||
def test_api_connectivity(self) -> bool:
|
||||
"""Test API connectivity"""
|
||||
available = self.check_api_available()
|
||||
self.log_test("API Connectivity", available,
|
||||
"API server is reachable" if available else "API server is not reachable")
|
||||
return available
|
||||
|
||||
def test_auto_recording_status(self) -> bool:
|
||||
"""Test auto-recording status endpoint"""
|
||||
status = self.get_auto_recording_status()
|
||||
success = bool(status and "running" in status)
|
||||
self.log_test("Auto-Recording Status API", success,
|
||||
f"Status: {status}" if success else "Failed to get status")
|
||||
return success
|
||||
|
||||
def test_camera_auto_recording_config(self) -> bool:
|
||||
"""Test camera auto-recording configuration"""
|
||||
success = True
|
||||
|
||||
# Test enabling auto-recording for camera1
|
||||
enabled = self.enable_auto_recording("camera1")
|
||||
if enabled:
|
||||
self.log_test("Enable Auto-Recording (camera1)", True, "Successfully enabled")
|
||||
else:
|
||||
self.log_test("Enable Auto-Recording (camera1)", False, "Failed to enable")
|
||||
success = False
|
||||
|
||||
# Check camera status
|
||||
time.sleep(1)
|
||||
camera_status = self.get_camera_status("camera1")
|
||||
auto_enabled = camera_status.get("auto_recording_enabled", False)
|
||||
self.log_test("Auto-Recording Status Check", auto_enabled,
|
||||
f"Camera1 auto-recording enabled: {auto_enabled}")
|
||||
|
||||
if not auto_enabled:
|
||||
success = False
|
||||
|
||||
return success
|
||||
|
||||
def test_machine_state_simulation(self) -> bool:
|
||||
"""Test machine state change simulation"""
|
||||
try:
|
||||
# Test vibratory conveyor (camera1)
|
||||
self.simulate_machine_state_change("vibratory_conveyor", "on")
|
||||
time.sleep(2)
|
||||
|
||||
camera_status = self.get_camera_status("camera1")
|
||||
is_recording = camera_status.get("is_recording", False)
|
||||
auto_active = camera_status.get("auto_recording_active", False)
|
||||
|
||||
self.log_test("Machine ON -> Recording Start", is_recording,
|
||||
f"Camera1 recording: {is_recording}, auto-active: {auto_active}")
|
||||
|
||||
# Test turning machine off
|
||||
time.sleep(3)
|
||||
self.simulate_machine_state_change("vibratory_conveyor", "off")
|
||||
time.sleep(2)
|
||||
|
||||
camera_status = self.get_camera_status("camera1")
|
||||
is_recording_after = camera_status.get("is_recording", False)
|
||||
auto_active_after = camera_status.get("auto_recording_active", False)
|
||||
|
||||
self.log_test("Machine OFF -> Recording Stop", not is_recording_after,
|
||||
f"Camera1 recording: {is_recording_after}, auto-active: {auto_active_after}")
|
||||
|
||||
return is_recording and not is_recording_after
|
||||
|
||||
except Exception as e:
|
||||
self.log_test("Machine State Simulation", False, f"Error: {e}")
|
||||
return False
|
||||
|
||||
def test_retry_mechanism(self) -> bool:
|
||||
"""Test retry mechanism for failed recording attempts"""
|
||||
# This test would require simulating camera failures
|
||||
# For now, we'll just check if the retry queue is accessible
|
||||
try:
|
||||
status = self.get_auto_recording_status()
|
||||
retry_queue = status.get("retry_queue", {})
|
||||
|
||||
self.log_test("Retry Queue Access", True,
|
||||
f"Retry queue accessible, current items: {len(retry_queue)}")
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
self.log_test("Retry Queue Access", False, f"Error: {e}")
|
||||
return False
|
||||
|
||||
def run_all_tests(self):
|
||||
"""Run all auto-recording tests"""
|
||||
print("🧪 Starting Auto-Recording Tests")
|
||||
print("=" * 50)
|
||||
|
||||
# Check if system is running
|
||||
if not self.test_api_connectivity():
|
||||
print("\n❌ Cannot run tests - API server is not available")
|
||||
print("Please start the USDA Vision System first:")
|
||||
print(" python main.py")
|
||||
return False
|
||||
|
||||
# Run tests
|
||||
tests = [
|
||||
self.test_auto_recording_status,
|
||||
self.test_camera_auto_recording_config,
|
||||
self.test_machine_state_simulation,
|
||||
self.test_retry_mechanism,
|
||||
]
|
||||
|
||||
passed = 0
|
||||
total = len(tests)
|
||||
|
||||
for test in tests:
|
||||
try:
|
||||
if test():
|
||||
passed += 1
|
||||
time.sleep(1) # Brief pause between tests
|
||||
except Exception as e:
|
||||
self.log_test(test.__name__, False, f"Exception: {e}")
|
||||
|
||||
# Print summary
|
||||
print("\n" + "=" * 50)
|
||||
print(f"📊 Test Summary: {passed}/{total} tests passed")
|
||||
|
||||
if passed == total:
|
||||
print("🎉 All auto-recording tests passed!")
|
||||
return True
|
||||
else:
|
||||
print(f"⚠️ {total - passed} test(s) failed")
|
||||
return False
|
||||
|
||||
def cleanup(self):
|
||||
"""Cleanup after tests"""
|
||||
print("\n🧹 Cleaning up...")
|
||||
|
||||
# Disable auto-recording for test cameras
|
||||
self.disable_auto_recording("camera1")
|
||||
self.disable_auto_recording("camera2")
|
||||
|
||||
# Turn off machines
|
||||
self.simulate_machine_state_change("vibratory_conveyor", "off")
|
||||
self.simulate_machine_state_change("blower_separator", "off")
|
||||
|
||||
print("✅ Cleanup completed")
|
||||
|
||||
|
||||
def main():
|
||||
"""Main test function"""
|
||||
tester = AutoRecordingTester()
|
||||
|
||||
try:
|
||||
success = tester.run_all_tests()
|
||||
return 0 if success else 1
|
||||
except KeyboardInterrupt:
|
||||
print("\n⚠️ Tests interrupted by user")
|
||||
return 1
|
||||
except Exception as e:
|
||||
print(f"\n❌ Test execution failed: {e}")
|
||||
return 1
|
||||
finally:
|
||||
tester.cleanup()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
exit_code = main()
|
||||
sys.exit(exit_code)
|
||||
@@ -0,0 +1,193 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Test script to verify auto-recording functionality with simulated MQTT messages.
|
||||
|
||||
This script tests that:
|
||||
1. Auto recording manager properly handles machine state changes
|
||||
2. Recording starts when machine turns "on"
|
||||
3. Recording stops when machine turns "off"
|
||||
4. Camera configuration from config.json is used
|
||||
"""
|
||||
|
||||
import sys
|
||||
import os
|
||||
import time
|
||||
import logging
|
||||
from datetime import datetime
|
||||
|
||||
# Add the current directory to Python path
|
||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))
|
||||
|
||||
|
||||
def setup_logging():
|
||||
"""Setup logging for the test"""
|
||||
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s")
|
||||
|
||||
|
||||
def test_auto_recording_with_mqtt():
|
||||
"""Test auto recording functionality with simulated MQTT messages"""
|
||||
print("🧪 Testing Auto Recording with MQTT Messages")
|
||||
print("=" * 50)
|
||||
|
||||
try:
|
||||
# Import required modules
|
||||
from usda_vision_system.core.config import Config
|
||||
from usda_vision_system.core.state_manager import StateManager
|
||||
from usda_vision_system.core.events import EventSystem, EventType
|
||||
from usda_vision_system.recording.auto_manager import AutoRecordingManager
|
||||
|
||||
print("✅ Modules imported successfully")
|
||||
|
||||
# Create system components
|
||||
config = Config("config.json")
|
||||
state_manager = StateManager()
|
||||
event_system = EventSystem()
|
||||
|
||||
# Create a mock camera manager for testing
|
||||
class MockCameraManager:
|
||||
def __init__(self):
|
||||
self.recording_calls = []
|
||||
self.stop_calls = []
|
||||
|
||||
def manual_start_recording(self, camera_name, filename, exposure_ms=None, gain=None, fps=None):
|
||||
call_info = {"camera_name": camera_name, "filename": filename, "exposure_ms": exposure_ms, "gain": gain, "fps": fps, "timestamp": datetime.now()}
|
||||
self.recording_calls.append(call_info)
|
||||
print(f"📹 MOCK: Starting recording for {camera_name}")
|
||||
print(f" - Filename: {filename}")
|
||||
print(f" - Settings: exposure={exposure_ms}ms, gain={gain}, fps={fps}")
|
||||
return True
|
||||
|
||||
def manual_stop_recording(self, camera_name):
|
||||
call_info = {"camera_name": camera_name, "timestamp": datetime.now()}
|
||||
self.stop_calls.append(call_info)
|
||||
print(f"⏹️ MOCK: Stopping recording for {camera_name}")
|
||||
return True
|
||||
|
||||
mock_camera_manager = MockCameraManager()
|
||||
|
||||
# Create auto recording manager
|
||||
auto_manager = AutoRecordingManager(config, state_manager, event_system, mock_camera_manager)
|
||||
|
||||
print("✅ Auto recording manager created")
|
||||
|
||||
# Start the auto recording manager
|
||||
if not auto_manager.start():
|
||||
print("❌ Failed to start auto recording manager")
|
||||
return False
|
||||
|
||||
print("✅ Auto recording manager started")
|
||||
|
||||
# Test 1: Simulate blower_separator turning ON (should trigger camera1)
|
||||
print("\n🔄 Test 1: Blower separator turns ON")
|
||||
print("📡 Publishing machine state change event...")
|
||||
# Use the same event system instance that the auto manager is subscribed to
|
||||
event_system.publish(EventType.MACHINE_STATE_CHANGED, "test_script", {"machine_name": "blower_separator", "state": "on", "previous_state": None})
|
||||
time.sleep(1.0) # Give more time for event processing
|
||||
|
||||
print(f"📊 Total recording calls so far: {len(mock_camera_manager.recording_calls)}")
|
||||
for call in mock_camera_manager.recording_calls:
|
||||
print(f" - {call['camera_name']}: {call['filename']}")
|
||||
|
||||
# Check if recording was started for camera1
|
||||
camera1_calls = [call for call in mock_camera_manager.recording_calls if call["camera_name"] == "camera1"]
|
||||
if camera1_calls:
|
||||
call = camera1_calls[-1]
|
||||
print(f"✅ Camera1 recording started with config:")
|
||||
print(f" - Exposure: {call['exposure_ms']}ms (expected: 0.3ms)")
|
||||
print(f" - Gain: {call['gain']} (expected: 4.0)")
|
||||
print(f" - FPS: {call['fps']} (expected: 0)")
|
||||
|
||||
# Verify settings match config.json
|
||||
if call["exposure_ms"] == 0.3 and call["gain"] == 4.0 and call["fps"] == 0:
|
||||
print("✅ Camera settings match config.json")
|
||||
else:
|
||||
print("❌ Camera settings don't match config.json")
|
||||
return False
|
||||
else:
|
||||
print("❌ Camera1 recording was not started")
|
||||
return False
|
||||
|
||||
# Test 2: Simulate vibratory_conveyor turning ON (should trigger camera2)
|
||||
print("\n🔄 Test 2: Vibratory conveyor turns ON")
|
||||
event_system.publish(EventType.MACHINE_STATE_CHANGED, "test_script", {"machine_name": "vibratory_conveyor", "state": "on", "previous_state": None})
|
||||
time.sleep(0.5)
|
||||
|
||||
# Check if recording was started for camera2
|
||||
camera2_calls = [call for call in mock_camera_manager.recording_calls if call["camera_name"] == "camera2"]
|
||||
if camera2_calls:
|
||||
call = camera2_calls[-1]
|
||||
print(f"✅ Camera2 recording started with config:")
|
||||
print(f" - Exposure: {call['exposure_ms']}ms (expected: 0.2ms)")
|
||||
print(f" - Gain: {call['gain']} (expected: 2.0)")
|
||||
print(f" - FPS: {call['fps']} (expected: 0)")
|
||||
|
||||
# Verify settings match config.json
|
||||
if call["exposure_ms"] == 0.2 and call["gain"] == 2.0 and call["fps"] == 0:
|
||||
print("✅ Camera settings match config.json")
|
||||
else:
|
||||
print("❌ Camera settings don't match config.json")
|
||||
return False
|
||||
else:
|
||||
print("❌ Camera2 recording was not started")
|
||||
return False
|
||||
|
||||
# Test 3: Simulate machines turning OFF
|
||||
print("\n🔄 Test 3: Machines turn OFF")
|
||||
event_system.publish(EventType.MACHINE_STATE_CHANGED, "test_script", {"machine_name": "blower_separator", "state": "off", "previous_state": None})
|
||||
event_system.publish(EventType.MACHINE_STATE_CHANGED, "test_script", {"machine_name": "vibratory_conveyor", "state": "off", "previous_state": None})
|
||||
time.sleep(0.5)
|
||||
|
||||
# Check if recordings were stopped
|
||||
camera1_stops = [call for call in mock_camera_manager.stop_calls if call["camera_name"] == "camera1"]
|
||||
camera2_stops = [call for call in mock_camera_manager.stop_calls if call["camera_name"] == "camera2"]
|
||||
|
||||
if camera1_stops and camera2_stops:
|
||||
print("✅ Both cameras stopped recording when machines turned OFF")
|
||||
else:
|
||||
print(f"❌ Recording stop failed - Camera1 stops: {len(camera1_stops)}, Camera2 stops: {len(camera2_stops)}")
|
||||
return False
|
||||
|
||||
# Stop the auto recording manager
|
||||
auto_manager.stop()
|
||||
print("✅ Auto recording manager stopped")
|
||||
|
||||
print("\n🎉 All auto recording tests passed!")
|
||||
print("\n📊 Summary:")
|
||||
print(f" - Total recording starts: {len(mock_camera_manager.recording_calls)}")
|
||||
print(f" - Total recording stops: {len(mock_camera_manager.stop_calls)}")
|
||||
print(f" - Camera1 starts: {len([c for c in mock_camera_manager.recording_calls if c['camera_name'] == 'camera1'])}")
|
||||
print(f" - Camera2 starts: {len([c for c in mock_camera_manager.recording_calls if c['camera_name'] == 'camera2'])}")
|
||||
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ Test failed with error: {e}")
|
||||
import traceback
|
||||
|
||||
traceback.print_exc()
|
||||
return False
|
||||
|
||||
|
||||
def main():
|
||||
"""Run the auto recording test"""
|
||||
setup_logging()
|
||||
|
||||
success = test_auto_recording_with_mqtt()
|
||||
|
||||
if success:
|
||||
print("\n✅ Auto recording functionality is working correctly!")
|
||||
print("\n📝 The system should now properly:")
|
||||
print(" 1. Start recording when machines turn ON")
|
||||
print(" 2. Stop recording when machines turn OFF")
|
||||
print(" 3. Use camera settings from config.json")
|
||||
print(" 4. Generate appropriate filenames with timestamps")
|
||||
else:
|
||||
print("\n❌ Auto recording test failed!")
|
||||
print("Please check the implementation and try again.")
|
||||
|
||||
return success
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
success = main()
|
||||
sys.exit(0 if success else 1)
|
||||
@@ -0,0 +1,214 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Simple test script for auto-recording functionality.
|
||||
|
||||
This script performs basic checks to verify that the auto-recording feature
|
||||
is properly integrated and configured.
|
||||
"""
|
||||
|
||||
import sys
|
||||
import os
|
||||
import json
|
||||
import time
|
||||
|
||||
# Add the current directory to Python path
|
||||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||||
|
||||
|
||||
def test_config_structure():
|
||||
"""Test that config.json has the required auto-recording fields"""
|
||||
print("🔍 Testing configuration structure...")
|
||||
|
||||
try:
|
||||
with open("config.json", "r") as f:
|
||||
config = json.load(f)
|
||||
|
||||
# Check system-level auto-recording setting
|
||||
system_config = config.get("system", {})
|
||||
if "auto_recording_enabled" not in system_config:
|
||||
print("❌ Missing 'auto_recording_enabled' in system config")
|
||||
return False
|
||||
|
||||
print(f"✅ System auto-recording enabled: {system_config['auto_recording_enabled']}")
|
||||
|
||||
# Check camera-level auto-recording settings
|
||||
cameras = config.get("cameras", [])
|
||||
if not cameras:
|
||||
print("❌ No cameras found in config")
|
||||
return False
|
||||
|
||||
for camera in cameras:
|
||||
camera_name = camera.get("name", "unknown")
|
||||
required_fields = ["auto_start_recording_enabled", "auto_recording_max_retries", "auto_recording_retry_delay_seconds"]
|
||||
|
||||
missing_fields = [field for field in required_fields if field not in camera]
|
||||
if missing_fields:
|
||||
print(f"❌ Camera {camera_name} missing fields: {missing_fields}")
|
||||
return False
|
||||
|
||||
print(f"✅ Camera {camera_name} auto-recording config:")
|
||||
print(f" - Enabled: {camera['auto_start_recording_enabled']}")
|
||||
print(f" - Max retries: {camera['auto_recording_max_retries']}")
|
||||
print(f" - Retry delay: {camera['auto_recording_retry_delay_seconds']}s")
|
||||
print(f" - Machine topic: {camera.get('machine_topic', 'unknown')}")
|
||||
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ Error reading config: {e}")
|
||||
return False
|
||||
|
||||
|
||||
def test_module_imports():
|
||||
"""Test that all required modules can be imported"""
|
||||
print("\n🔍 Testing module imports...")
|
||||
|
||||
try:
|
||||
from usda_vision_system.recording.auto_manager import AutoRecordingManager
|
||||
|
||||
print("✅ AutoRecordingManager imported successfully")
|
||||
|
||||
from usda_vision_system.core.config import Config
|
||||
|
||||
config = Config("config.json")
|
||||
print("✅ Config loaded successfully")
|
||||
|
||||
from usda_vision_system.core.state_manager import StateManager
|
||||
|
||||
state_manager = StateManager()
|
||||
print("✅ StateManager created successfully")
|
||||
|
||||
from usda_vision_system.core.events import EventSystem
|
||||
|
||||
event_system = EventSystem()
|
||||
print("✅ EventSystem created successfully")
|
||||
|
||||
# Test creating AutoRecordingManager (without camera_manager for now)
|
||||
auto_manager = AutoRecordingManager(config, state_manager, event_system, None)
|
||||
print("✅ AutoRecordingManager created successfully")
|
||||
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ Import error: {e}")
|
||||
return False
|
||||
|
||||
|
||||
def test_camera_mapping():
|
||||
"""Test camera to machine topic mapping"""
|
||||
print("\n🔍 Testing camera to machine mapping...")
|
||||
|
||||
try:
|
||||
with open("config.json", "r") as f:
|
||||
config = json.load(f)
|
||||
|
||||
cameras = config.get("cameras", [])
|
||||
expected_mappings = {"camera1": "blower_separator", "camera2": "vibratory_conveyor"} # Blower separator # Conveyor/cracker cam
|
||||
|
||||
for camera in cameras:
|
||||
camera_name = camera.get("name")
|
||||
machine_topic = camera.get("machine_topic")
|
||||
|
||||
if camera_name in expected_mappings:
|
||||
expected_topic = expected_mappings[camera_name]
|
||||
if machine_topic == expected_topic:
|
||||
print(f"✅ {camera_name} correctly mapped to {machine_topic}")
|
||||
else:
|
||||
print(f"❌ {camera_name} mapped to {machine_topic}, expected {expected_topic}")
|
||||
return False
|
||||
else:
|
||||
print(f"⚠️ Unknown camera: {camera_name}")
|
||||
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ Error checking mappings: {e}")
|
||||
return False
|
||||
|
||||
|
||||
def test_api_models():
|
||||
"""Test that API models include auto-recording fields"""
|
||||
print("\n🔍 Testing API models...")
|
||||
|
||||
try:
|
||||
from usda_vision_system.api.models import CameraStatusResponse, CameraConfigResponse, AutoRecordingConfigRequest, AutoRecordingConfigResponse, AutoRecordingStatusResponse
|
||||
|
||||
# Check CameraStatusResponse has auto-recording fields
|
||||
camera_response = CameraStatusResponse(name="test", status="available", is_recording=False, last_checked="2024-01-01T00:00:00", auto_recording_enabled=True, auto_recording_active=False, auto_recording_failure_count=0)
|
||||
print("✅ CameraStatusResponse includes auto-recording fields")
|
||||
|
||||
# Check CameraConfigResponse has auto-recording fields
|
||||
config_response = CameraConfigResponse(
|
||||
name="test",
|
||||
machine_topic="test_topic",
|
||||
storage_path="/test",
|
||||
enabled=True,
|
||||
auto_start_recording_enabled=True,
|
||||
auto_recording_max_retries=3,
|
||||
auto_recording_retry_delay_seconds=5,
|
||||
exposure_ms=1.0,
|
||||
gain=1.0,
|
||||
target_fps=30.0,
|
||||
sharpness=100,
|
||||
contrast=100,
|
||||
saturation=100,
|
||||
gamma=100,
|
||||
noise_filter_enabled=False,
|
||||
denoise_3d_enabled=False,
|
||||
auto_white_balance=True,
|
||||
color_temperature_preset=0,
|
||||
wb_red_gain=1.0,
|
||||
wb_green_gain=1.0,
|
||||
wb_blue_gain=1.0,
|
||||
anti_flicker_enabled=False,
|
||||
light_frequency=1,
|
||||
bit_depth=8,
|
||||
hdr_enabled=False,
|
||||
hdr_gain_mode=0,
|
||||
)
|
||||
print("✅ CameraConfigResponse includes auto-recording fields")
|
||||
|
||||
print("✅ All auto-recording API models available")
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ API model error: {e}")
|
||||
return False
|
||||
|
||||
|
||||
def main():
|
||||
"""Run all basic tests"""
|
||||
print("🧪 Auto-Recording Integration Test")
|
||||
print("=" * 40)
|
||||
|
||||
tests = [test_config_structure, test_module_imports, test_camera_mapping, test_api_models]
|
||||
|
||||
passed = 0
|
||||
total = len(tests)
|
||||
|
||||
for test in tests:
|
||||
try:
|
||||
if test():
|
||||
passed += 1
|
||||
except Exception as e:
|
||||
print(f"❌ Test {test.__name__} failed with exception: {e}")
|
||||
|
||||
print("\n" + "=" * 40)
|
||||
print(f"📊 Results: {passed}/{total} tests passed")
|
||||
|
||||
if passed == total:
|
||||
print("🎉 All integration tests passed!")
|
||||
print("\n📝 Next steps:")
|
||||
print("1. Start the system: python main.py")
|
||||
print("2. Run full tests: python tests/test_auto_recording.py")
|
||||
print("3. Test with MQTT messages to trigger auto-recording")
|
||||
return True
|
||||
else:
|
||||
print(f"⚠️ {total - passed} test(s) failed")
|
||||
print("Please fix the issues before running the full system")
|
||||
return False
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
success = main()
|
||||
sys.exit(0 if success else 1)
|
||||
Reference in New Issue
Block a user