MODEL/usda-vision
5
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add comprehensive test suite for USDA Vision Camera System

Browse Source 
- Implemented main test script to verify system components and functionality.
- Added individual test scripts for camera exposure settings, API changes, camera recovery, maximum FPS, MQTT events, logging, and timezone functionality.
- Created service file for system management and automatic startup.
- Included detailed logging and error handling in test scripts for better diagnostics.
- Ensured compatibility with existing camera SDK and API endpoints.
This commit is contained in:
Alireza Vaezi
2025-07-28 17:33:49 -04:00
parent 9cb043ef5f
commit 7bc8138f24
72 changed files with 419 additions and 118 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
usda_vision_system/camera/__init__.py
View File

@@ -2,7 +2,7 @@
Camera module for the USDA Vision Camera System.
This module handles GigE camera discovery, management, monitoring, and recording
using the python demo library (mvsdk).
using the camera SDK library (mvsdk).
"""
from .manager import CameraManager

BIN
usda_vision_system/camera/__pycache__/__init__.cpython-311.pyc
View File

Binary file not shown.

BIN
usda_vision_system/camera/__pycache__/manager.cpython-311.pyc
View File

Binary file not shown.

BIN
usda_vision_system/camera/__pycache__/monitor.cpython-311.pyc
View File

Binary file not shown.

BIN
usda_vision_system/camera/__pycache__/recorder.cpython-311.pyc
View File

Binary file not shown.

4
usda_vision_system/camera/manager.py
View File

@@ -12,8 +12,8 @@ import logging
from typing import Dict, List, Optional, Tuple, Any
from datetime import datetime
# Add python demo to path
sys.path.append(os.path.join(os.path.dirname(__file__), "..", "..", "python demo"))
# Add camera SDK to path
sys.path.append(os.path.join(os.path.dirname(__file__), "..", "..", "camera_sdk"))
import mvsdk
from ..core.config import Config, CameraConfig

4
usda_vision_system/camera/monitor.py
View File

@@ -12,8 +12,8 @@ import logging
import contextlib
from typing import Dict, List, Optional, Any
# Add python demo to path
sys.path.append(os.path.join(os.path.dirname(__file__), "..", "..", "python demo"))
# Add camera SDK to path
sys.path.append(os.path.join(os.path.dirname(__file__), "..", "..", "camera_sdk"))
import mvsdk
from ..core.config import Config

6
usda_vision_system/camera/recorder.py
View File

@@ -1,7 +1,7 @@
"""
Camera Recorder for the USDA Vision Camera System.
This module handles video recording from GigE cameras using the python demo library (mvsdk).
This module handles video recording from GigE cameras using the camera SDK library (mvsdk).
"""
import sys
@@ -16,8 +16,8 @@ from typing import Optional, Dict, Any
from datetime import datetime
from pathlib import Path
# Add python demo to path
sys.path.append(os.path.join(os.path.dirname(__file__), "..", "..", "python demo"))
# Add camera SDK to path
sys.path.append(os.path.join(os.path.dirname(__file__), "..", "..", "camera_sdk"))
import mvsdk
from ..core.config import CameraConfig

28
usda_vision_system/camera/sdk_config.py
View File

@@ -8,8 +8,8 @@ import sys
import os
import logging
# Add python demo to path
sys.path.append(os.path.join(os.path.dirname(__file__), "..", "..", "python demo"))
# Add camera SDK to path
sys.path.append(os.path.join(os.path.dirname(__file__), "..", "..", "camera_sdk"))
import mvsdk
logger = logging.getLogger(__name__)
@@ -21,62 +21,62 @@ _sdk_initialized = False
def initialize_sdk_with_suppression():
"""Initialize the camera SDK with error suppression"""
global _sdk_initialized
if _sdk_initialized:
return True
try:
# Initialize SDK with English language
result = mvsdk.CameraSdkInit(1)
if result == 0:
logger.info("Camera SDK initialized successfully")
# Try to set system options to suppress logging
try:
# These are common options that might control logging
# We'll try them and ignore failures since they might not be supported
# Try to disable debug output
try:
mvsdk.CameraSetSysOption("DebugLevel", "0")
except:
pass
# Try to disable console output
try:
mvsdk.CameraSetSysOption("ConsoleOutput", "0")
except:
pass
# Try to disable error logging
try:
mvsdk.CameraSetSysOption("ErrorLog", "0")
except:
pass
# Try to set log level to none
try:
mvsdk.CameraSetSysOption("LogLevel", "0")
except:
pass
# Try to disable verbose mode
try:
mvsdk.CameraSetSysOption("Verbose", "0")
except:
pass
logger.debug("Attempted to configure SDK logging options")
except Exception as e:
logger.debug(f"Could not configure SDK logging options: {e}")
_sdk_initialized = True
return True
else:
logger.error(f"SDK initialization failed with code: {result}")
return False
except Exception as e:
logger.error(f"SDK initialization failed: {e}")
return False

BIN
usda_vision_system/core/__pycache__/config.cpython-311.pyc
View File

Binary file not shown.

131
usda_vision_system/core/config.py
View File

@@ -16,23 +16,22 @@ from pathlib import Path
@dataclass
class MQTTConfig:
"""MQTT broker configuration"""
broker_host: str = "192.168.1.110"
broker_port: int = 1883
username: Optional[str] = None
password: Optional[str] = None
topics: Dict[str, str] = None
topics: Optional[Dict[str, str]] = None
def __post_init__(self):
if self.topics is None:
self.topics = {
"vibratory_conveyor": "vision/vibratory_conveyor/state",
"blower_separator": "vision/blower_separator/state"
}
self.topics = {"vibratory_conveyor": "vision/vibratory_conveyor/state", "blower_separator": "vision/blower_separator/state"}
@dataclass
class CameraConfig:
"""Individual camera configuration"""
name: str
machine_topic: str # Which MQTT topic triggers this camera
storage_path: str
@@ -43,42 +42,44 @@ class CameraConfig:
# Image Quality Settings
sharpness: int = 100 # 0-200, default 100 (no sharpening)
contrast: int = 100 # 0-200, default 100 (normal contrast)
saturation: int = 100 # 0-200, default 100 (normal saturation, color cameras only)
gamma: int = 100 # 0-300, default 100 (normal gamma)
contrast: int = 100 # 0-200, default 100 (normal contrast)
saturation: int = 100 # 0-200, default 100 (normal saturation, color cameras only)
gamma: int = 100 # 0-300, default 100 (normal gamma)
# Noise Reduction
noise_filter_enabled: bool = True # Enable basic noise filtering
denoise_3d_enabled: bool = False # Enable advanced 3D denoising (may reduce FPS)
denoise_3d_enabled: bool = False # Enable advanced 3D denoising (may reduce FPS)
# Color Settings (for color cameras)
auto_white_balance: bool = True # Enable automatic white balance
auto_white_balance: bool = True # Enable automatic white balance
color_temperature_preset: int = 0 # 0=auto, 1=daylight, 2=fluorescent, etc.
# Advanced Settings
anti_flicker_enabled: bool = True # Reduce artificial lighting flicker
light_frequency: int = 1 # 0=50Hz, 1=60Hz (match local power frequency)
light_frequency: int = 1 # 0=50Hz, 1=60Hz (match local power frequency)
# Bit Depth & Format
bit_depth: int = 8 # 8, 10, 12, or 16 bits per channel
bit_depth: int = 8 # 8, 10, 12, or 16 bits per channel
# HDR Settings
hdr_enabled: bool = False # Enable High Dynamic Range
hdr_gain_mode: int = 0 # HDR processing mode
hdr_enabled: bool = False # Enable High Dynamic Range
hdr_gain_mode: int = 0 # HDR processing mode
@dataclass
class StorageConfig:
"""Storage configuration"""
base_path: str = "/storage"
max_file_size_mb: int = 1000 # Max size per video file
max_recording_duration_minutes: int = 60 # Max recording duration
cleanup_older_than_days: int = 30 # Auto cleanup old files
@dataclass
class SystemConfig:
"""System-wide configuration"""
camera_check_interval_seconds: int = 2
log_level: str = "INFO"
log_file: str = "usda_vision_system.log"
@@ -90,60 +91,57 @@ class SystemConfig:
class Config:
"""Main configuration manager"""
def __init__(self, config_file: Optional[str] = None):
self.config_file = config_file or "config.json"
self.logger = logging.getLogger(__name__)
# Default configurations
self.mqtt = MQTTConfig()
self.storage = StorageConfig()
self.system = SystemConfig()
# Camera configurations - will be populated from config file or defaults
self.cameras: List[CameraConfig] = []
# Load configuration
self.load_config()
# Ensure storage directories exist
self._ensure_storage_directories()
def load_config(self) -> None:
"""Load configuration from file"""
config_path = Path(self.config_file)
if config_path.exists():
try:
with open(config_path, 'r') as f:
with open(config_path, "r") as f:
config_data = json.load(f)
# Load MQTT config
if 'mqtt' in config_data:
mqtt_data = config_data['mqtt']
if "mqtt" in config_data:
mqtt_data = config_data["mqtt"]
self.mqtt = MQTTConfig(**mqtt_data)
# Load storage config
if 'storage' in config_data:
storage_data = config_data['storage']
if "storage" in config_data:
storage_data = config_data["storage"]
self.storage = StorageConfig(**storage_data)
# Load system config
if 'system' in config_data:
system_data = config_data['system']
if "system" in config_data:
system_data = config_data["system"]
self.system = SystemConfig(**system_data)
# Load camera configs
if 'cameras' in config_data:
self.cameras = [
CameraConfig(**cam_data)
for cam_data in config_data['cameras']
]
if "cameras" in config_data:
self.cameras = [CameraConfig(**cam_data) for cam_data in config_data["cameras"]]
else:
self._create_default_camera_configs()
self.logger.info(f"Configuration loaded from {config_path}")
except Exception as e:
self.logger.error(f"Error loading config from {config_path}: {e}")
self._create_default_camera_configs()
@@ -151,66 +149,50 @@ class Config:
self.logger.info(f"Config file {config_path} not found, using defaults")
self._create_default_camera_configs()
self.save_config() # Save default config
def _create_default_camera_configs(self) -> None:
"""Create default camera configurations"""
self.cameras = [
CameraConfig(
name="camera1",
machine_topic="vibratory_conveyor",
storage_path=os.path.join(self.storage.base_path, "camera1")
),
CameraConfig(
name="camera2",
machine_topic="blower_separator",
storage_path=os.path.join(self.storage.base_path, "camera2")
)
]
self.cameras = [CameraConfig(name="camera1", machine_topic="vibratory_conveyor", storage_path=os.path.join(self.storage.base_path, "camera1")), CameraConfig(name="camera2", machine_topic="blower_separator", storage_path=os.path.join(self.storage.base_path, "camera2"))]
def save_config(self) -> None:
"""Save current configuration to file"""
config_data = {
'mqtt': asdict(self.mqtt),
'storage': asdict(self.storage),
'system': asdict(self.system),
'cameras': [asdict(cam) for cam in self.cameras]
}
config_data = {"mqtt": asdict(self.mqtt), "storage": asdict(self.storage), "system": asdict(self.system), "cameras": [asdict(cam) for cam in self.cameras]}
try:
with open(self.config_file, 'w') as f:
with open(self.config_file, "w") as f:
json.dump(config_data, f, indent=2)
self.logger.info(f"Configuration saved to {self.config_file}")
except Exception as e:
self.logger.error(f"Error saving config to {self.config_file}: {e}")
def _ensure_storage_directories(self) -> None:
"""Ensure all storage directories exist"""
try:
# Create base storage directory
Path(self.storage.base_path).mkdir(parents=True, exist_ok=True)
# Create camera-specific directories
for camera in self.cameras:
Path(camera.storage_path).mkdir(parents=True, exist_ok=True)
self.logger.info("Storage directories verified/created")
except Exception as e:
self.logger.error(f"Error creating storage directories: {e}")
def get_camera_by_topic(self, topic: str) -> Optional[CameraConfig]:
"""Get camera configuration by MQTT topic"""
for camera in self.cameras:
if camera.machine_topic == topic:
return camera
return None
def get_camera_by_name(self, name: str) -> Optional[CameraConfig]:
"""Get camera configuration by name"""
for camera in self.cameras:
if camera.name == name:
return camera
return None
def update_camera_config(self, name: str, **kwargs) -> bool:
"""Update camera configuration"""
camera = self.get_camera_by_name(name)
@@ -221,12 +203,7 @@ class Config:
self.save_config()
return True
return False
def to_dict(self) -> Dict[str, Any]:
"""Convert configuration to dictionary"""
return {
'mqtt': asdict(self.mqtt),
'storage': asdict(self.storage),
'system': asdict(self.system),
'cameras': [asdict(cam) for cam in self.cameras]
}
return {"mqtt": asdict(self.mqtt), "storage": asdict(self.storage), "system": asdict(self.system), "cameras": [asdict(cam) for cam in self.cameras]}