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

Add 'api/' from commit '14ac229098e65aa643f84e8e17e0c5f1aaf8d639'

Browse Source 
git-subtree-dir: api
git-subtree-mainline: 4743f19aef
git-subtree-split: 14ac229098
This commit is contained in:
Alireza Vaezi
2025-08-07 20:57:34 -04:00
parent 4743f19aef 14ac229098
commit 7dbb36d619
146 changed files with 31249 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

233
api/usda_vision_system/core/config.py Normal file
View File

@@ -0,0 +1,233 @@
"""
Configuration management for the USDA Vision Camera System.
This module handles all configuration settings including MQTT broker settings,
camera configurations, storage paths, and system parameters.
"""
import os
import json
import logging
from typing import Dict, List, Optional, Any
from dataclasses import dataclass, asdict
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: 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"}
@dataclass
class CameraConfig:
"""Individual camera configuration"""
name: str
machine_topic: str # Which MQTT topic triggers this camera
storage_path: str
exposure_ms: float = 1.0
gain: float = 3.5
target_fps: float = 3.0
enabled: bool = True
# Video recording settings
video_format: str = "mp4" # Video file format (mp4, avi)
video_codec: str = "mp4v" # Video codec (mp4v for MP4, XVID for AVI)
video_quality: int = 95 # Video quality (0-100, higher is better)
# Auto-recording settings
auto_start_recording_enabled: bool = False # Enable automatic recording when machine turns on
auto_recording_max_retries: int = 3 # Maximum retry attempts for failed auto-recording starts
auto_recording_retry_delay_seconds: int = 5 # Delay between retry attempts
# 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)
# Noise Reduction
noise_filter_enabled: bool = True # Enable basic noise filtering
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
color_temperature_preset: int = 0 # 0=auto, 1=daylight, 2=fluorescent, etc.
# Manual White Balance RGB Gains (for manual white balance mode)
wb_red_gain: float = 1.0 # Red channel gain (0.0-3.99, default 1.0)
wb_green_gain: float = 1.0 # Green channel gain (0.0-3.99, default 1.0)
wb_blue_gain: float = 1.0 # Blue channel gain (0.0-3.99, default 1.0)
# Advanced Settings
anti_flicker_enabled: bool = True # Reduce artificial lighting flicker
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
# HDR Settings
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"
api_host: str = "0.0.0.0"
api_port: int = 8000
enable_api: bool = True
timezone: str = "America/New_York"
# Auto-recording system settings
auto_recording_enabled: bool = True # Global enable/disable for auto-recording feature # Atlanta, Georgia timezone
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:
config_data = json.load(f)
# Load MQTT config
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"]
self.storage = StorageConfig(**storage_data)
# Load system config
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 = []
for cam_data in config_data["cameras"]:
# Set defaults for new video format fields if not present
cam_data.setdefault("video_format", "mp4")
cam_data.setdefault("video_codec", "mp4v")
cam_data.setdefault("video_quality", 95)
self.cameras.append(CameraConfig(**cam_data))
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()
else:
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"))]
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]}
try:
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)
if camera:
for key, value in kwargs.items():
if hasattr(camera, key):
setattr(camera, key, value)
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]}