Add crop manager for staging and training image lifecycle
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
48
vigilar/detection/crop_manager.py
Normal file
48
vigilar/detection/crop_manager.py
Normal file
@@ -0,0 +1,48 @@
|
||||
"""Manage detection crop images for training and staging."""
|
||||
|
||||
import logging
|
||||
import shutil
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
import cv2
|
||||
import numpy as np
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class CropManager:
|
||||
def __init__(self, staging_dir: str, training_dir: str):
|
||||
self._staging_dir = Path(staging_dir)
|
||||
self._training_dir = Path(training_dir)
|
||||
|
||||
def save_staging_crop(self, crop: np.ndarray, species: str, camera_id: str) -> str:
|
||||
self._staging_dir.mkdir(parents=True, exist_ok=True)
|
||||
timestamp = int(time.time() * 1000)
|
||||
filename = f"{species}_{camera_id}_{timestamp}.jpg"
|
||||
filepath = self._staging_dir / filename
|
||||
cv2.imwrite(str(filepath), crop)
|
||||
return str(filepath)
|
||||
|
||||
def promote_to_training(self, staging_path: str, pet_name: str) -> str:
|
||||
pet_dir = self._training_dir / pet_name.lower()
|
||||
pet_dir.mkdir(parents=True, exist_ok=True)
|
||||
src = Path(staging_path)
|
||||
dst = pet_dir / src.name
|
||||
shutil.move(str(src), str(dst))
|
||||
return str(dst)
|
||||
|
||||
def cleanup_expired(self, retention_days: int = 7) -> int:
|
||||
if not self._staging_dir.exists():
|
||||
return 0
|
||||
|
||||
cutoff = time.time() - retention_days * 86400
|
||||
deleted = 0
|
||||
for filepath in self._staging_dir.iterdir():
|
||||
if filepath.is_file() and filepath.stat().st_mtime < cutoff:
|
||||
filepath.unlink()
|
||||
deleted += 1
|
||||
|
||||
if deleted:
|
||||
log.info("Cleaned up %d expired staging crops", deleted)
|
||||
return deleted
|
||||
Reference in New Issue
Block a user