usda-vision/docs/CODE_QUALITY_IMPROVEMENTS.md

Code Quality Improvements - Simple & Safe Refactorings

📊 Current Codebase Analysis

Largest Files (Focus Areas)

• recorder.py: 1,122 lines ⚠️
• server.py: 842 lines ⚠️
• streamer.py: 745 lines ⚠️
• manager.py: 614 lines

Key Observations

1. Large classes with multiple responsibilities
2. API routes all in one method (_setup_routes - 700+ lines)
3. Duplicate code (camera error suppression, initialization)
4. Long methods (recording loop, streaming loop)
5. Mixed concerns (camera hardware + business logic)

---

🎯 Simple, Safe Refactorings (No Behavior Changes)

1. Extract Duplicate Code ⭐ Easy Win

Problem: suppress_camera_errors() appears in 3+ files

Solution: Move to shared utility

# Before: Duplicate in recorder.py, streamer.py, monitor.py
@contextlib.contextmanager
def suppress_camera_errors():
    # ... 20 lines duplicated

# After: Single source in camera/utils.py
# camera/utils.py
@contextlib.contextmanager
def suppress_camera_errors():
    """Suppress camera SDK error output"""
    # ... implementation

# Then import everywhere:
from .utils import suppress_camera_errors

Impact:

• ✅ Reduces duplication
• ✅ Single place to fix bugs
• ✅ No behavior change
• ⚡ 5 minutes, zero risk

---

2. Split API Routes into Modules ⭐ High Impact

Problem: _setup_routes() has 700+ lines, hard to navigate

Current Structure:

# api/server.py
def _setup_routes(self):
    @self.app.get("/cameras/...")
    async def get_cameras(): ...
    
    @self.app.post("/cameras/...")
    async def start_recording(): ...
    
    # ... 30+ more routes

Solution: Group routes by domain

# api/server.py
def _setup_routes(self):
    from .routes import camera_routes, recording_routes, system_routes
    camera_routes.register(self.app, self.camera_manager)
    recording_routes.register(self.app, self.camera_manager)
    system_routes.register(self.app, self.state_manager, ...)

# api/routes/camera_routes.py
def register(app, camera_manager):
    @app.get("/cameras")
    async def get_cameras():
        return camera_manager.get_all_camera_status()
    
    # ... all camera routes

Impact:

• ✅ Much easier to find/edit routes
• ✅ Clearer organization
• ✅ Can split across files
• ⚡ 30 minutes, low risk

---

3. Extract Camera Initialization ⭐ Medium Impact

Problem: Camera initialization code duplicated in recorder.py, streamer.py, monitor.py

Solution: Create CameraInitializer class

# camera/initializer.py
class CameraInitializer:
    """Handles camera initialization with consistent configuration"""
    
    def __init__(self, camera_config: CameraConfig, device_info):
        self.camera_config = camera_config
        self.device_info = device_info
    
    def initialize(self) -> tuple[int, Any, bool]:
        """Returns (hCamera, cap, monoCamera)"""
        # Common initialization logic
        # ...
        return hCamera, cap, monoCamera
    
    def configure_settings(self, hCamera):
        """Configure camera settings"""
        # Common configuration
        # ...

# Then use:
initializer = CameraInitializer(config, device_info)
self.hCamera, self.cap, self.monoCamera = initializer.initialize()
initializer.configure_settings(self.hCamera)

Impact:

• ✅ Removes ~200 lines of duplication
• ✅ Consistent initialization
• ✅ Easier to test
• ⚡ 1 hour, medium risk (test carefully)

---

4. Break Down Large Methods ⭐ Medium Impact

Problem: _recording_loop() is 100+ lines, _streaming_loop() is 80+ lines

Solution: Extract frame processing logic

# Before: recorder.py
def _recording_loop(self, use_streamer_frames: bool):
    while not self._stop_recording_event.is_set():
        # Get frame (30 lines)
        if use_streamer_frames:
            # ... complex logic
        else:
            # ... complex logic
        
        # Write frame (10 lines)
        # Rate control (5 lines)

# After: recorder.py
def _recording_loop(self, use_streamer_frames: bool):
    frame_source = self._create_frame_source(use_streamer_frames)
    
    while not self._stop_recording_event.is_set():
        frame = frame_source.get_frame()
        if frame:
            self._write_frame(frame)
        self._control_frame_rate()

def _create_frame_source(self, use_streamer_frames):
    """Create appropriate frame source"""
    if use_streamer_frames:
        return StreamerFrameSource(self.streamer)
    return DirectCameraSource(self.hCamera, self.frame_buffer)

def _write_frame(self, frame):
    """Write frame to video"""
    if self.video_writer:
        self.video_writer.write(frame)
        self.frame_count += 1

Impact:

• ✅ Methods are shorter, clearer
• ✅ Easier to test individual pieces
• ✅ Better readability
• ⚡ 2 hours, medium risk

---

5. Type Hints & Documentation ⭐ Easy, Incremental

Problem: Some methods lack type hints, unclear parameter meanings

Solution: Add type hints gradually (no behavior change)

# Before
def start_recording(self, filename):
    # ...

# After
def start_recording(self, filename: str) -> bool:
    """
    Start video recording for the camera.
    
    Args:
        filename: Output filename (will be prefixed with timestamp)
        
    Returns:
        True if recording started successfully, False otherwise
        
    Raises:
        CameraException: If camera initialization fails
    """

Impact:

• ✅ Better IDE support
• ✅ Self-documenting code
• ✅ Catch errors earlier
• ⚡ Can do incrementally, zero risk

---

6. Create Value Objects ⭐ Medium Impact

Problem: Camera properties scattered across instance variables

Current:

self.hCamera = ...
self.cap = ...
self.monoCamera = ...
self.frame_buffer = ...
self.frame_buffer_size = ...

Better:

# camera/domain/camera_handle.py
@dataclass
class CameraHandle:
    """Represents a camera hardware connection"""
    handle: int
    capabilities: Any  # Camera capability struct
    is_monochrome: bool
    frame_buffer: Any
    frame_buffer_size: int
    
    def is_valid(self) -> bool:
        return self.handle is not None

# Usage
self.camera = CameraHandle(...)
if self.camera.is_valid():
    # ...

Impact:

• ✅ Groups related data
• ✅ Easier to pass around
• ✅ Better encapsulation
• ⚡ 2 hours, low risk

---

7. Extract Constants ⭐ Easy Win

Problem: Magic numbers scattered throughout

# Before
time.sleep(0.1)
timeout=200
maxsize=5

# After: camera/constants.py
STREAMING_LOOP_SLEEP = 0.1  # seconds
CAMERA_GET_BUFFER_TIMEOUT = 200  # milliseconds
FRAME_QUEUE_MAXSIZE = 5

Impact:

• ✅ Self-documenting
• ✅ Easy to tune
• ✅ No behavior change
• ⚡ 30 minutes, zero risk

---

🚀 Recommended Refactoring Order (Safe → Risky)

Phase 1: Quick Wins (1-2 hours, zero risk)

1. ✅ Extract suppress_camera_errors() to shared utils
2. ✅ Extract constants to constants.py
3. ✅ Add type hints to public methods

Phase 2: Organization (3-4 hours, low risk)

4. ✅ Split API routes into modules (routes/camera_routes.py, etc.)
5. ✅ Group related functions into utility modules
6. ✅ Improve docstrings

Phase 3: Structure (6-8 hours, medium risk)

7. ✅ Extract CameraInitializer class
8. ✅ Break down large methods (_recording_loop, _streaming_loop)
9. ✅ Create value objects for camera handles

Phase 4: Advanced (optional, higher risk)

10. ⚠️ Extract frame source abstractions
11. ⚠️ Create repository pattern for camera access
12. ⚠️ Dependency injection container

---

📋 Specific File Improvements

recorder.py (1,122 lines)

Quick wins:

• Extract suppress_camera_errors → utils
• Extract constants (timeouts, buffer sizes)
• Split _initialize_video_writer (100+ lines) into smaller methods

Medium refactoring:

• Extract _recording_loop frame source logic
• Create VideoWriterManager class
• Extract camera configuration to separate class

server.py (842 lines)

Quick wins:

• Split _setup_routes into route modules
• Extract WebSocket logic to separate file
• Group related endpoints

streamer.py (745 lines)

Quick wins:

• Extract suppress_camera_errors → utils
• Extract RTSP FFmpeg command building
• Extract frame queue management

Medium refactoring:

• Extract _streaming_loop frame processing
• Create FrameQueueManager class

manager.py (614 lines)

Quick wins:

• Extract camera discovery to separate class
• Split initialization methods
• Extract status reporting

---

🎯 "Good Enough" Approach

Don't over-engineer! Focus on:

1. Readability: Can a new developer understand it?
2. Editability: Can you change one thing without breaking others?
3. Testability: Can you test pieces in isolation?

Avoid:

• ❌ Premature abstraction
• ❌ Design patterns "just because"
• ❌ Perfect code (good enough > perfect)
• ❌ Breaking working code

---

💡 Minimal Viable Refactoring

If you only do 3 things:

1. Extract duplicate code (suppress_camera_errors, constants)

   • 30 minutes, huge improvement

2. Split API routes (into route modules)

   • 1 hour, makes API much more manageable

3. Add type hints (gradually, as you touch code)

   • Ongoing, improves IDE support

Result: Much more maintainable code with minimal effort!

---

🔧 Tools to Help

Code Quality Tools

# Linting
pip install ruff black mypy
ruff check .
black --check .
mypy camera-management-api/

# Complexity
pip install radon
radon cc camera-management-api/usda_vision_system -a

Refactoring Safely

• ✅ Write tests first (if not present)
• ✅ Refactor in small steps
• ✅ Test after each change
• ✅ Use git branches
• ✅ Review diffs carefully

---

📊 Success Metrics

After refactoring, you should see:

• ✅ Lower cyclomatic complexity
• ✅ Smaller average method length
• ✅ Less duplicate code
• ✅ More consistent patterns
• ✅ Easier to add features
• ✅ Fewer bugs when making changes

---

🎓 Best Practices

1. One refactoring per commit - easier to review/rollback
2. Don't refactor while adding features - separate PRs
3. Measure before/after - use code metrics
4. Document decisions - why this structure?
5. Keep it simple - don't add complexity "for the future"

---

Bottom Line: Start with #1, #2, and #3 (duplicate extraction, route splitting, type hints). These give you 80% of the benefit with 20% of the effort, and they're completely safe!