- Deleted unused API test files, RTSP diagnostic scripts, and development utility scripts to reduce clutter. - Removed outdated database schema and modularization proposal documents to maintain focus on current architecture. - Cleaned up configuration files and logging scripts that are no longer in use, enhancing project maintainability.
438 lines
10 KiB
Markdown
438 lines
10 KiB
Markdown
# 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
|
|
|
|
```python
|
|
# Before: Duplicate in recorder.py, streamer.py, monitor.py
|
|
@contextlib.contextmanager
|
|
def suppress_camera_errors():
|
|
# ... 20 lines duplicated
|
|
```
|
|
|
|
```python
|
|
# 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**:
|
|
```python
|
|
# 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
|
|
|
|
```python
|
|
# 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
|
|
|
|
```python
|
|
# 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
|
|
|
|
```python
|
|
# 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)
|
|
```
|
|
|
|
```python
|
|
# 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)
|
|
|
|
```python
|
|
# 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**:
|
|
```python
|
|
self.hCamera = ...
|
|
self.cap = ...
|
|
self.monoCamera = ...
|
|
self.frame_buffer = ...
|
|
self.frame_buffer_size = ...
|
|
```
|
|
|
|
**Better**:
|
|
```python
|
|
# 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
|
|
|
|
```python
|
|
# 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
|
|
```bash
|
|
# 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!
|
|
|