usda-vision/docs/REFACTORING_PLAN.md

Quick Start: Code Quality Refactoring Plan

🎯 Priority Order (Safe → Risky)

✅ Task 1: Extract Duplicate Code (30 min, zero risk)

File: Create camera-management-api/usda_vision_system/camera/utils.py

Move suppress_camera_errors() from 3 files into one shared location.

Files to update:

• camera/recorder.py
• camera/streamer.py
• camera/monitor.py

Benefit: Single source of truth, easier to fix bugs

---

✅ Task 2: Extract Constants (30 min, zero risk)

File: Create camera-management-api/usda_vision_system/camera/constants.py

Extract magic numbers:

• Timeouts (200ms, 1000ms, etc.)
• Queue sizes (5, 10, 30)
• Sleep intervals (0.1s, etc.)
• FPS defaults (10.0, 15.0, 30.0)

Benefit: Self-documenting, easy to tune

---

✅ Task 3: Split API Routes (1-2 hours, low risk)

Create:

• api/routes/__init__.py
• api/routes/camera_routes.py
• api/routes/recording_routes.py
• api/routes/system_routes.py
• api/routes/mqtt_routes.py
• api/routes/storage_routes.py

Move routes from api/server.py to appropriate modules.

Benefit: Much easier to navigate 800+ line file

---

✅ Task 4: Add Type Hints (ongoing, zero risk)

Add type hints as you touch code:

• Start with public methods
• Use Optional[...] for nullable values
• Use Dict[str, ...] for dictionaries

Benefit: Better IDE support, catch errors early

---

📝 Detailed Steps for Task 1 (Start Here!)

Step 1: Create utils file

# camera-management-api/usda_vision_system/camera/utils.py
"""Shared utilities for camera operations"""
import contextlib
import os

@contextlib.contextmanager
def suppress_camera_errors():
    """Context manager to temporarily suppress camera SDK error output"""
    original_stderr = os.dup(2)
    original_stdout = os.dup(1)

    try:
        devnull = os.open(os.devnull, os.O_WRONLY)
        os.dup2(devnull, 2)  # stderr
        os.dup2(devnull, 1)  # stdout
        os.close(devnull)
        yield
    finally:
        os.dup2(original_stderr, 2)
        os.dup2(original_stdout, 1)
        os.close(original_stderr)
        os.close(original_stdout)

Step 2: Update imports in recorder.py

# camera-management-api/usda_vision_system/camera/recorder.py
# Remove the duplicate function
# Change import to:
from .utils import suppress_camera_errors

Step 3: Repeat for streamer.py and monitor.py

Step 4: Test

docker compose restart api
# Verify everything still works

That's it! Single source of truth for camera error suppression.

---

📝 Detailed Steps for Task 2 (Constants)

Step 1: Create constants file

# camera-management-api/usda_vision_system/camera/constants.py
"""Constants for camera operations"""

# Timeouts (milliseconds)
CAMERA_GET_BUFFER_TIMEOUT = 200
CAMERA_INIT_TIMEOUT = 1000
CAMERA_TEST_CAPTURE_TIMEOUT = 1000

# Frame queue sizes
MJPEG_QUEUE_MAXSIZE = 5
RTSP_QUEUE_MAXSIZE = 10
RECORDING_QUEUE_MAXSIZE = 30

# Frame rates
PREVIEW_FPS = 10.0
RTSP_FPS = 15.0
DEFAULT_VIDEO_FPS = 30.0

# Sleep intervals (seconds)
STREAMING_LOOP_SLEEP = 0.1
FRAME_RATE_CONTROL_SLEEP_BASE = 0.01

# JPEG quality
PREVIEW_JPEG_QUALITY = 70

Step 2: Update files to use constants

# Before
pRawData, FrameHead = mvsdk.CameraGetImageBuffer(self.hCamera, 200)

# After
from .constants import CAMERA_GET_BUFFER_TIMEOUT
pRawData, FrameHead = mvsdk.CameraGetImageBuffer(self.hCamera, CAMERA_GET_BUFFER_TIMEOUT)

---

📝 Detailed Steps for Task 3 (Route Splitting)

Step 1: Create routes directory structure

api/
├── __init__.py
├── server.py
├── models.py
└── routes/
    ├── __init__.py
    ├── camera_routes.py
    ├── recording_routes.py
    ├── system_routes.py
    ├── mqtt_routes.py
    └── storage_routes.py

Step 2: Example - camera_routes.py

# api/routes/camera_routes.py
from fastapi import APIRouter, HTTPException
from typing import Dict
from ..models import CameraStatusResponse

def register_camera_routes(app, camera_manager, logger):
    """Register camera-related routes"""
    
    @app.get("/cameras", response_model=Dict[str, CameraStatusResponse])
    async def get_cameras():
        """Get all camera statuses"""
        try:
            return camera_manager.get_all_camera_status()
        except Exception as e:
            logger.error(f"Error getting cameras: {e}")
            raise HTTPException(status_code=500, detail=str(e))
    
    @app.get("/cameras/{camera_name}/status", response_model=CameraStatusResponse)
    async def get_camera_status(camera_name: str):
        """Get specific camera status"""
        # ... implementation

Step 3: Update server.py

# api/server.py
def _setup_routes(self):
    from .routes import camera_routes, recording_routes, system_routes
    
    # Register route groups
    camera_routes.register_camera_routes(self.app, self.camera_manager, self.logger)
    recording_routes.register_recording_routes(self.app, self.camera_manager, self.logger)
    system_routes.register_system_routes(self.app, self.state_manager, self.logger)
    # ... etc

---

✅ Testing After Each Refactoring

# 1. Restart API
docker compose restart api

# 2. Test key endpoints
curl http://localhost:8000/health
curl http://localhost:8000/system/status
curl http://localhost:8000/cameras

# 3. Test camera operations (if cameras connected)
curl http://localhost:8000/cameras/camera1/status

# 4. Check logs
docker compose logs api --tail 50

---

📊 Progress Tracking

• Task 1: Extract duplicate code (utils.py)
• Task 2: Extract constants (constants.py)
• Task 3: Split API routes (routes/ directory)
• Task 4: Add type hints (ongoing)

Estimated Time: 2-3 hours total for first 3 tasks

Risk Level: Very Low - All are structural changes with no behavior modification

---

Start with Task 1 - it's the easiest and gives immediate benefit!