Ryan/VideoDetect
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Cleanup

Browse Source
This commit is contained in:
Ryan Shpeherd
2026-09-09 15:01:42 -04:00
parent 9f098d4b1f
commit 458d8862cc
16 changed files with 0 additions and 3187 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/batcher.py
-68
View File
@@ -1,68 +0,0 @@
"""Dynamic batching utilities for face detection inference."""
import logging
import time
from collections import deque
from typing import Callable, Deque, Generic, List, Optional, TypeVar
from gpu_manager import GPUMemoryManager
logger = logging.getLogger(__name__)
T = TypeVar("T")
R = TypeVar("R")
class DynamicBatcher(Generic[T, R]):
"""Accumulate items into batches and flush based on size or timeout."""
def __init__(
self,
process_fn: Callable[[List[T]], List[R]],
gpu_manager: Optional[GPUMemoryManager] = None,
max_batch_size: int = 16,
batch_timeout_ms: float = 100.0,
min_batch_size: int = 1,
):
self.process_fn = process_fn
self.gpu_manager = gpu_manager
self.max_batch_size = max_batch_size
self.batch_timeout_ms = batch_timeout_ms
self.min_batch_size = min_batch_size
self._queue: Deque[T] = deque()
self._last_flush = time.monotonic()
def add(self, item: T) -> List[R]:
"""Add an item and return any results if a batch was flushed."""
self._queue.append(item)
if len(self._queue) >= self._current_batch_size():
return self.flush()
elapsed_ms = (time.monotonic() - self._last_flush) * 1000
if elapsed_ms >= self.batch_timeout_ms and len(self._queue) >= self.min_batch_size:
return self.flush()
return []
def flush(self) -> List[R]:
"""Flush all queued items through the processor."""
if not self._queue:
return []
batch = [self._queue.popleft() for _ in range(min(len(self._queue), self._current_batch_size()))]
results = self.process_fn(batch)
self._last_flush = time.monotonic()
if self.gpu_manager is not None:
self.gpu_manager.adjust_batch_size()
self.gpu_manager.empty_cache()
return results
def _current_batch_size(self) -> int:
if self.gpu_manager is not None:
return min(self.gpu_manager.current_batch_size, self.max_batch_size)
return self.max_batch_size
@property
def queued_count(self) -> int:
return len(self._queue)
src/data_export.py
-115
View File
@@ -1,115 +0,0 @@
"""Batch Parquet/JSONL export of video processing results."""
import json
import logging
import os
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Dict, List, Optional
logger = logging.getLogger(__name__)
class DataExporter:
"""Buffer result records and flush to Parquet or JSONL when the batch is full."""
def __init__(
self,
output_path: str,
model_version: str = "v0.0.0",
export_format: str = "parquet",
compression: str = "snappy",
batch_size: int = 100,
include_frame_confidences: bool = True,
):
self.output_path = Path(output_path) / model_version
self.model_version = model_version
self.export_format = export_format.lower()
self.compression = compression
self.batch_size = batch_size
self.include_frame_confidences = include_frame_confidences
self._buffer: List[Dict[str, Any]] = []
self.output_path.mkdir(parents=True, exist_ok=True)
def add(self, record: Dict[str, Any]) -> None:
"""Buffer one result record; flush automatically when batch is full."""
self._buffer.append(record)
if len(self._buffer) >= self.batch_size:
self.flush()
def flush(self) -> Optional[str]:
"""Write buffered records to disk; returns the output file path or None."""
if not self._buffer:
return None
records = self._buffer[:]
self._buffer.clear()
if not self.include_frame_confidences:
for r in records:
r.pop("confidence_scores", None)
batch_id = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S")
stem = f"{self.model_version}_{batch_id}"
if self.export_format in ("parquet", "both"):
path = self._write_parquet(records, stem)
if self.export_format in ("jsonl", "both"):
path = self._write_jsonl(records, stem)
if self.export_format not in ("parquet", "jsonl", "both"):
logger.warning("Unknown export format '%s'; defaulting to jsonl", self.export_format)
path = self._write_jsonl(records, stem)
return str(path)
def _write_parquet(self, records: List[Dict[str, Any]], stem: str) -> Path:
out = self.output_path / f"{stem}.parquet"
try:
import pyarrow as pa
import pyarrow.parquet as pq
schema = pa.schema([
pa.field("video_id", pa.int64()),
pa.field("file_path", pa.string()),
pa.field("model_version", pa.string()),
pa.field("sample_count", pa.int32()),
pa.field("confidence_scores", pa.list_(pa.float64())),
pa.field("video_confidence", pa.float64()),
pa.field("routing", pa.string()),
pa.field("processed_at", pa.string()),
])
table = pa.table(
{
"video_id": [r.get("video_id") for r in records],
"file_path": [r.get("file_path", "") for r in records],
"model_version": [r.get("model_version", self.model_version) for r in records],
"sample_count": [r.get("sample_count", 0) for r in records],
"confidence_scores": [r.get("confidence_scores", []) for r in records],
"video_confidence": [float(r.get("video_confidence", 0.0)) for r in records],
"routing": [r.get("routing", "SKIP") for r in records],
"processed_at": [r.get("processed_at", "") for r in records],
},
schema=schema,
)
pq.write_table(table, out, compression=self.compression)
logger.info("Exported %d records to %s", len(records), out)
except ImportError:
logger.warning("pyarrow not available; falling back to JSONL")
out = self._write_jsonl(records, stem.replace(".parquet", ""))
return out
def _write_jsonl(self, records: List[Dict[str, Any]], stem: str) -> Path:
out = self.output_path / f"{stem}.jsonl"
with open(out, "w", encoding="utf-8") as fh:
for record in records:
fh.write(json.dumps(record, default=str) + "\n")
logger.info("Exported %d records to %s", len(records), out)
return out
def __del__(self):
if self._buffer:
try:
self.flush()
except Exception:
pass
src/main.py
-122
View File
@@ -1,122 +0,0 @@
"""
VideoDetect - Video Classification System
Main entry point for the worker service.
Initializes all components and starts the processing pipeline.
"""
import logging
import signal
import sys
import threading
import time
from pathlib import Path
# Add src to path
sys.path.insert(0, str(Path(__file__).parent))
from config_loader import get_config
from db_connector import DBConnector
from logging_config import setup_logging
from orchestrator import WorkerPool
from scanner import DirectoryScanner
logger = logging.getLogger(__name__)
def graceful_shutdown(signum, frame):
"""Handle shutdown signals gracefully."""
logger.info("Received signal %d, initiating graceful shutdown...", signum)
sys.exit(0)
def main():
"""Initialize and start the VideoDetect worker."""
# Register signal handlers
signal.signal(signal.SIGTERM, graceful_shutdown)
signal.signal(signal.SIGINT, graceful_shutdown)
# Load configuration
config = get_config()
log_config = config.get_section("logging")
setup_logging(
level=log_config.get("level", "INFO"),
log_format=log_config.get("format", "json"),
rotation_max_bytes=log_config.get("rotation_max_bytes", 104857600),
rotation_backup_count=log_config.get("rotation_backup_count", 10),
)
logger.info("VideoDetect Worker starting...")
logger.info("Config: %s", config)
# Initialize database connection
db_config = config.get_section("database")
db = DBConnector(
host=db_config.get("host", "mariadb"),
port=db_config.get("port", 3306),
database=db_config.get("name", "videodetect"),
user=db_config.get("user", "videodetect"),
password=db_config.get("password", "videodetect123"),
pool_size=db_config.get("pool_size", 20),
pool_min=db_config.get("pool_min", 5),
pool_recycle=db_config.get("pool_recycle", 3600),
)
# Verify database connectivity
if not db.health_check():
logger.error("Cannot connect to database. Exiting.")
sys.exit(1)
logger.info("Database connection established.")
# Initialize schema if needed
schema_path = Path(__file__).parent.parent / "db" / "schema.sql"
if schema_path.exists():
db.initialize_schema(str(schema_path))
logger.info("Schema initialized.")
# Verify GPU availability
try:
import torch
if torch.cuda.is_available():
gpu_count = torch.cuda.device_count()
gpu_name = torch.cuda.get_device_name(0)
logger.info("GPU available: %d GPUs, primary: %s", gpu_count, gpu_name)
else:
logger.warning("CUDA is not available! Processing will be slow.")
except ImportError:
logger.warning("PyTorch not installed. GPU features disabled.")
# Initialize the directory scanner.
# It scans the permanent storage location in place (no staging/copy step),
# periodically discovering new/removed video files and queueing any that
# have not been scanned yet as PENDING for the worker pool to pick up.
storage_config = config.get_section("storage")
scanner_config = config.get_section("scanner")
scanner = DirectoryScanner(
db_connector=db,
config=config.data,
input_path=storage_config.get("input_path", "/data/input"),
scan_interval=int(scanner_config.get("scan_interval_seconds", 60)),
walker_threads=int(scanner_config.get("walker_threads", 8)),
)
logger.info("Worker initialization complete. Starting scanner and processing loop...")
# Run the scanner in a background thread (its start() is a blocking loop).
scanner_thread = threading.Thread(
target=scanner.start, name="directory-scanner", daemon=True
)
scanner_thread.start()
pool = WorkerPool(db, config.data, max_workers=1)
try:
pool.start()
except KeyboardInterrupt:
logger.info("Worker shutting down.")
pool.stop()
scanner.stop()
if __name__ == "__main__":
main()
src/orchestrator.py
-441
View File
@@ -1,441 +0,0 @@
"""
Batch Orchestration Skeleton
Manages job queue, worker pool, state transitions, and crash recovery.
Ensures atomic state transitions and idempotent processing.
"""
import logging
import time
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime, timezone
from enum import Enum
from pathlib import Path
from typing import Dict, List, Optional
from batcher import DynamicBatcher
from classifier import FaceClassifier
from data_export import DataExporter
from face_detector import Detection, FaceDetector
from frame_sampler import FrameSampler
from gpu_manager import GPUMemoryManager
from prober import VideoProber
from result_updater import ResultUpdater
from scratch_manager import ScratchManager
import aggregator
import router
logger = logging.getLogger(__name__)
class JobStatus(Enum):
"""Job status states."""
PENDING = "PENDING"
PROCESSING = "PROCESSING"
COMPLETED = "COMPLETED"
FAILED = "ERROR"
SKIPPED = "UNSCANNABLE"
class Job:
"""Represents a single video processing job."""
def __init__(
self,
video_id: int,
file_path: str,
priority: float = 0.0,
sampling_interval_seconds: Optional[int] = None,
):
self.video_id = video_id
self.file_path = file_path
self.priority = priority # Higher = more urgent (based on modification time)
self.sampling_interval_seconds = sampling_interval_seconds
self.status = JobStatus.PENDING
self.created_at = datetime.now(timezone.utc)
self.started_at: Optional[datetime] = None
self.completed_at: Optional[datetime] = None
self.attempts = 0
self.max_retries = 3
self.error_message: Optional[str] = None
def to_dict(self) -> dict:
return {
"video_id": self.video_id,
"file_path": self.file_path,
"priority": self.priority,
"status": self.status.value,
"created_at": self.created_at.isoformat(),
"started_at": self.started_at.isoformat() if self.started_at else None,
"completed_at": self.completed_at.isoformat() if self.completed_at else None,
"attempts": self.attempts,
"error_message": self.error_message,
}
def __repr__(self):
return f"Job(id={self.video_id}, path={self.file_path}, status={self.status.value})"
class WorkerPool:
"""Manage a pool of worker processes for parallel processing."""
def __init__(self, db_connector, config: dict, max_workers: int = 2):
self.db = db_connector
self.config = config
self.max_workers = max_workers
self._executor = ThreadPoolExecutor(max_workers=max_workers)
self._running = False
self._jobs_processed = 0
self._jobs_failed = 0
self._sampling_config = (config or {}).get("sampling", {})
self._storage_config = (config or {}).get("storage", {})
self._face_detection_config = (config or {}).get("face_detection", {})
self._batching_config = (config or {}).get("batching", {})
self._gpu_manager = GPUMemoryManager(
max_memory_gb=config.get("gpu", {}).get("max_memory_gb", 18.0),
reduce_threshold_gb=self._batching_config.get("vram_reduce_threshold_gb", 16.0),
increase_threshold_gb=self._batching_config.get("vram_increase_threshold_gb", 10.0),
initial_batch_size=self._batching_config.get("max_batch_size", 16),
)
self._face_detector = FaceDetector(
engine_path=self._face_detection_config.get("model_path", "/models/face_detector/face_detector.trt"),
input_size=int(self._face_detection_config.get("input_size", 640)),
confidence_threshold=float(self._face_detection_config.get("confidence_threshold", 0.25)),
iou_threshold=float(self._face_detection_config.get("iou_threshold", 0.45)),
max_faces_per_frame=int(self._face_detection_config.get("max_faces_per_frame", 10)),
max_faces_per_video=int(self._face_detection_config.get("max_faces_per_video", 100)),
)
classifier_config = (config or {}).get("classifier", {})
self._classifier = FaceClassifier(
engine_path=classifier_config.get("model_path", "/models/classifier/classifier.trt"),
temperature=float(classifier_config.get("temperature", 1.0)),
input_size=int(classifier_config.get("input_size", 224)),
)
self._aggregation_config = (config or {}).get("aggregation", {})
self._routing_config = (config or {}).get("routing", {})
model_version = self._get_active_model_version()
export_config = (config or {}).get("export", {})
self._result_updater = ResultUpdater(db_connector, model_version=model_version)
self._exporter = DataExporter(
output_path=export_config.get("output_path", "/data/output"),
model_version=model_version,
export_format=export_config.get("format", "parquet"),
compression=export_config.get("compression", "snappy"),
batch_size=int(export_config.get("batch_size", 100)),
include_frame_confidences=bool(export_config.get("include_frame_confidences", True)),
)
def _get_active_model_version(self) -> str:
"""Return the currently active model version from the DB."""
try:
row = self.db.fetchone("SELECT version FROM models WHERE status = 'ACTIVE' LIMIT 1")
if row:
return row["version"]
except Exception as exc:
logger.debug("Could not fetch active model version: %s", exc)
return "v0.0.0-placeholder"
def start(self):
"""Start the worker pool."""
self._running = True
logger.info("Worker pool starting with %d workers", self.max_workers)
while self._running:
# Get pending jobs
jobs = self._get_pending_jobs()
if jobs:
# Submit jobs to executor
for job in jobs:
future = self._executor.submit(self._process_job, job)
future.add_done_callback(self._on_job_complete)
else:
time.sleep(5) # No jobs, wait
def stop(self):
"""Stop the worker pool gracefully."""
self._running = False
logger.info("Worker pool stopping. Processed: %d, Failed: %d",
self._jobs_processed, self._jobs_failed)
self._executor.shutdown(wait=True)
def _get_pending_jobs(self) -> List[Job]:
"""Get pending jobs from DB with atomic locking."""
now = datetime.now(timezone.utc).replace(tzinfo=None)
# Atomic lock: update status from PENDING to PROCESSING
result = self.db.execute(
"""UPDATE videos SET status = 'PROCESSING', last_processed_time = %s,
updated_at = %s
WHERE id IN (
SELECT id FROM (
SELECT id FROM videos
WHERE status = 'PENDING'
ORDER BY last_scan_time DESC
LIMIT %s
) AS sub
)""",
(now, now, self.max_workers),
transaction=True,
)
if result == 0:
return []
# Get the locked jobs
locked = self.db.fetchall(
"SELECT id, file_path, last_scan_time FROM videos WHERE status = 'PROCESSING' AND last_processed_time = %s",
(now,),
)
return [
Job(
video_id=row["id"],
file_path=row["file_path"],
priority=row["last_scan_time"].timestamp(),
)
for row in locked
]
def _process_job(self, job: Job) -> bool:
"""Process a single job (placeholder - actual processing in later stories)."""
job.status = JobStatus.PROCESSING
job.started_at = datetime.now(timezone.utc)
job.attempts += 1
logger.info("Processing job: %s (attempt %d)", job, job.attempts)
try:
sampling_interval = job.sampling_interval_seconds or self._sampling_config.get("interval_seconds", 30)
scratch_base_path = self._storage_config.get("scratch_path", "/scratch")
scratch_manager = ScratchManager(
base_path=scratch_base_path,
video_id=str(job.video_id),
auto_cleanup=True,
)
frame_dir = scratch_manager.ensure_frame_dir()
prober = VideoProber(timeout=10)
metadata = prober.probe(job.file_path)
if metadata.is_unscannable or metadata.duration is None:
raise RuntimeError(metadata.error_message or "Video metadata could not be determined")
resolution = None
if metadata.resolution_w and metadata.resolution_h:
resolution = (metadata.resolution_w, metadata.resolution_h)
sampler = FrameSampler(
interval_seconds=int(sampling_interval),
quality=int(self._sampling_config.get("quality", 2)),
output_format=self._sampling_config.get("format", "jpeg"),
)
extracted_frames = sampler.extract_frames(
video_path=job.file_path,
output_dir=str(frame_dir),
duration=metadata.duration,
interval_seconds=int(sampling_interval),
resolution=resolution,
timeout_seconds=30,
)
if not extracted_frames:
scratch_manager.cleanup()
raise RuntimeError("No frames were extracted")
# Face detection on sampled frames
batch_size = self._gpu_manager.current_batch_size
detections_per_frame = self._face_detector.detect_faces(extracted_frames, batch_size=batch_size)
# Flatten and cap total faces per video
all_detections: List[Detection] = []
for frame_dets in detections_per_frame:
all_detections.extend(frame_dets)
all_detections = sorted(all_detections, key=lambda d: d.confidence, reverse=True)
all_detections = all_detections[: self._face_detection_config.get("max_faces_per_video", 100)]
# Extract face crops
crop_dir = scratch_manager.frame_dir.parent / "crops"
cropped_detections = self._face_detector.extract_crops(
all_detections,
output_dir=str(crop_dir),
crop_size=(224, 224),
)
if not cropped_detections:
logger.info("No faces detected for video %s; routing to SKIP", job.video_id)
routing_decision = router.SKIP
video_confidence = 0.0
frame_confidences: List[float] = []
else:
crop_paths = [d.crop_path for d in cropped_detections if d.crop_path]
frame_confidences = self._classifier.classify(
crop_paths, batch_size=self._gpu_manager.current_batch_size
)
video_confidence = aggregator.aggregate(
frame_confidences,
strategy=self._aggregation_config.get("strategy", "max"),
alpha=float(self._aggregation_config.get("alpha", 1.0)),
beta=float(self._aggregation_config.get("beta", 0.1)),
top_k=int(self._aggregation_config.get("top_k", 3)),
)
routing_decision = router.route(
video_confidence,
t_high=float(self._routing_config.get("T_high", 0.75)),
t_low=float(self._routing_config.get("T_low", 0.45)),
)
logger.info(
"Video %s: C=%.4f routing=%s faces=%d frames=%d",
job.video_id, video_confidence, routing_decision,
len(cropped_detections), len(extracted_frames),
)
# Persist results atomically (video update + processing log) before cleanup
persisted = self._result_updater.persist(
video_id=job.video_id,
frame_count=len(extracted_frames),
confidence=video_confidence,
routing=routing_decision,
frame_confidences=frame_confidences,
)
if persisted:
job.status = JobStatus.COMPLETED
job.completed_at = datetime.now(timezone.utc)
self._exporter.add({
"video_id": job.video_id,
"file_path": job.file_path,
"model_version": self._result_updater.model_version,
"sample_count": len(extracted_frames),
"confidence_scores": frame_confidences,
"video_confidence": video_confidence,
"routing": routing_decision,
"processed_at": datetime.now(timezone.utc).isoformat(),
})
# Scratch cleanup only after successful persistence
scratch_manager.cleanup_all()
self._jobs_processed += 1
return True
except Exception as e:
logger.error("Job failed: %s - %s", job, e, exc_info=True)
job.error_message = str(e)
if job.attempts < job.max_retries:
# Retry
self._retry_job(job)
return False
else:
# Max retries exceeded
self._fail_job(job)
self._jobs_failed += 1
return False
def _complete_job(
self,
job: Job,
frame_count: Optional[int] = None,
confidence: Optional[float] = None,
routing: Optional[str] = None,
):
"""Mark a job as completed, persisting confidence and routing decision."""
now = datetime.now(timezone.utc).replace(tzinfo=None)
job.status = JobStatus.COMPLETED
job.completed_at = datetime.now(timezone.utc)
self.db.execute(
"""UPDATE videos
SET status = 'COMPLETED',
frame_count = %s,
confidence_score = %s,
routing_decision = %s,
updated_at = %s
WHERE id = %s""",
(frame_count, confidence, routing, now, job.video_id),
transaction=True,
)
logger.info("Job completed: %s", job)
def _fail_job(self, job: Job):
"""Mark a job as failed."""
now = datetime.now(timezone.utc).replace(tzinfo=None)
job.status = JobStatus.FAILED
job.completed_at = datetime.now(timezone.utc)
self.db.execute(
"""UPDATE videos SET status = 'ERROR', error_message = %s, updated_at = %s
WHERE id = %s""",
(job.error_message, now, job.video_id),
transaction=True,
)
logger.error("Job failed permanently: %s", job)
def _retry_job(self, job: Job):
"""Re-queue a job for retry."""
now = datetime.now(timezone.utc).replace(tzinfo=None)
self.db.execute(
"""UPDATE videos SET status = 'PENDING', updated_at = %s
WHERE id = %s""",
(now, job.video_id),
transaction=True,
)
logger.info("Job re-queued for retry: %s (attempt %d/%d)",
job, job.attempts, job.max_retries)
def _on_job_complete(self, future):
"""Callback when a job completes."""
try:
future.result()
except Exception as e:
logger.error("Unhandled job error: %s", e)
def get_stats(self) -> dict:
"""Get worker pool statistics."""
return {
"max_workers": self.max_workers,
"jobs_processed": self._jobs_processed,
"jobs_failed": self._jobs_failed,
"is_running": self._running,
"active_workers": self._executor._work_queue.qsize(),
}
class CrashRecovery:
"""Handle crash recovery and stale job detection."""
def __init__(self, db_connector, lock_timeout_minutes: int = 5):
self.db = db_connector
self.lock_timeout = lock_timeout_minutes
def recover_stale_jobs(self) -> int:
"""Re-queue jobs stuck in PROCESSING beyond the lock timeout."""
now = datetime.now(timezone.utc).replace(tzinfo=None)
timeout = now.replace(minute=now.minute - self.lock_timeout)
result = self.db.execute(
"""UPDATE videos SET status = 'PENDING', updated_at = %s
WHERE status = 'PROCESSING' AND last_processed_time < %s""",
(now, timeout),
transaction=True,
)
if result > 0:
logger.info("Recovered %d stale jobs", result)
return result
def check_health(self) -> dict:
"""Check system health for crash recovery purposes."""
processing_count = self.db.fetchone(
"SELECT COUNT(*) as count FROM videos WHERE status = 'PROCESSING'"
)
pending_count = self.db.fetchone(
"SELECT COUNT(*) as count FROM videos WHERE status = 'PENDING'"
)
return {
"processing_count": processing_count["count"] if processing_count else 0,
"pending_count": pending_count["count"] if pending_count else 0,
"stale_jobs_recovered": self.recover_stale_jobs(),
}
src/processing_logger.py
-39
View File
@@ -1,39 +0,0 @@
"""Insert processing audit log entries within an existing DB transaction."""
import json
import logging
from datetime import datetime, timezone
from typing import List, Optional
logger = logging.getLogger(__name__)
def insert_log(
cursor,
video_id: int,
model_version: str,
frame_count: int,
confidence_score: Optional[float],
routing_decision: str,
frame_confidences: Optional[List[float]] = None,
) -> None:
"""Insert one row into processing_logs; must be called inside an open transaction."""
now = datetime.now(timezone.utc).replace(tzinfo=None)
confidence_scores_json = json.dumps(frame_confidences) if frame_confidences is not None else None
cursor.execute(
"""INSERT INTO processing_logs
(video_id, model_version, frame_count, confidence_score,
confidence_scores, routing_decision, processed_at)
VALUES (%s, %s, %s, %s, %s, %s, %s)""",
(
video_id,
model_version,
frame_count or 0,
confidence_score,
confidence_scores_json,
routing_decision,
now,
),
)
logger.debug("Inserted processing log for video %s (routing=%s)", video_id, routing_decision)
src/result_updater.py
-75
View File
@@ -1,75 +0,0 @@
"""Atomic result persistence: update videos + insert processing_logs in one transaction."""
import logging
from datetime import datetime, timezone
from typing import List, Optional
import processing_logger
logger = logging.getLogger(__name__)
class ResultUpdater:
def __init__(self, db_connector, model_version: str = "v0.0.0-placeholder"):
self.db = db_connector
self.model_version = model_version
def persist(
self,
video_id: int,
frame_count: int,
confidence: float,
routing: str,
frame_confidences: Optional[List[float]] = None,
) -> bool:
"""Atomically update videos and insert a processing log.
Returns False without raising if the state guard prevents the update
(video is no longer in PROCESSING state).
"""
now = datetime.now(timezone.utc).replace(tzinfo=None)
try:
with self.db.transaction() as conn:
cursor = conn.cursor()
try:
cursor.execute(
"""UPDATE videos
SET status = 'COMPLETED',
last_processed_time = %s,
confidence_score = %s,
routing_decision = %s,
model_version = %s,
frame_count = %s,
updated_at = %s
WHERE id = %s AND status = 'PROCESSING'""",
(now, confidence, routing, self.model_version, frame_count, now, video_id),
)
if cursor.rowcount == 0:
logger.warning(
"State guard: video %s is not PROCESSING; skipping update", video_id
)
return False
processing_logger.insert_log(
cursor,
video_id=video_id,
model_version=self.model_version,
frame_count=frame_count,
confidence_score=confidence,
routing_decision=routing,
frame_confidences=frame_confidences,
)
finally:
cursor.close()
logger.info(
"Persisted results for video %s: C=%.4f routing=%s model=%s",
video_id, confidence, routing, self.model_version,
)
return True
except Exception as exc:
logger.error("Failed to persist results for video %s: %s", video_id, exc)
raise
src/retry.py
-114
View File
@@ -1,114 +0,0 @@
"""
Retry decorator with exponential backoff for transient failures.
Usage:
@retry(max_attempts=3, step="extract")
def extract_frames(...): ...
Non-retryable error types are re-raised immediately without consuming attempts.
"""
import functools
import logging
import time
from typing import Callable, Optional, Tuple, Type
logger = logging.getLogger(__name__)
# Error class names that should never be retried
_NON_RETRYABLE_NAMES = frozenset({
"CodecUnsupportedError",
"FileCorruptError",
"InvalidPathError",
"FileNotFoundError",
"PermissionError",
"IsADirectoryError",
})
class RetryExhaustedError(Exception):
"""Raised when all retry attempts have been exhausted."""
def retry(
max_attempts: int = 3,
initial_delay: float = 1.0,
backoff_factor: float = 2.0,
exceptions: Tuple[Type[Exception], ...] = (Exception,),
step: str = "unknown",
):
"""
Decorator: retry ``func`` up to ``max_attempts`` times on retryable exceptions.
Non-retryable exceptions (see _NON_RETRYABLE_NAMES) propagate immediately.
Each retry waits ``initial_delay * backoff_factor ** attempt`` seconds.
"""
def decorator(func: Callable) -> Callable:
@functools.wraps(func)
def wrapper(*args, **kwargs):
last_exc: Optional[Exception] = None
for attempt in range(max_attempts):
try:
return func(*args, **kwargs)
except exceptions as exc:
if _is_non_retryable(exc):
logger.error(
"Non-retryable error in step=%s (attempt %d/%d): %s: %s",
step, attempt + 1, max_attempts,
type(exc).__name__, exc,
)
raise
last_exc = exc
_increment_retry_counter(step)
if attempt < max_attempts - 1:
delay = initial_delay * (backoff_factor ** attempt)
logger.warning(
"Transient error in step=%s (attempt %d/%d), "
"retrying in %.1fs: %s: %s",
step, attempt + 1, max_attempts, delay,
type(exc).__name__, exc,
)
time.sleep(delay)
else:
logger.error(
"All %d attempts exhausted in step=%s: %s: %s",
max_attempts, step, type(exc).__name__, exc,
)
raise RetryExhaustedError(
f"step={step} failed after {max_attempts} attempts"
) from last_exc
return wrapper
return decorator
def retry_from_config(config, step: str = "unknown"):
"""Build a ``@retry`` decorator from config.yaml monitoring.retry settings."""
mon = config.get_section("monitoring")
cfg = mon.get("retry", {})
return retry(
max_attempts=int(cfg.get("max_attempts", 3)),
initial_delay=float(cfg.get("initial_delay", 1.0)),
backoff_factor=float(cfg.get("backoff_factor", 2.0)),
step=step,
)
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _is_non_retryable(exc: Exception) -> bool:
return type(exc).__name__ in _NON_RETRYABLE_NAMES
def _increment_retry_counter(step: str):
try:
from metrics import retry_attempts_total
retry_attempts_total.labels(step=step).inc()
except Exception:
pass # metrics not available; don't break the retry logic
src/review_export.py
-114
View File
@@ -1,114 +0,0 @@
"""
Review queue export for backend scripts.
Provides CSV and JSON export of annotated review data, with optional
filtering by date range, model version, annotation status, and ground truth.
Can be called standalone or imported from other src/ modules.
"""
import csv
import io
import json
import logging
from datetime import datetime
from pathlib import Path
from typing import Any, Dict, List, Optional
logger = logging.getLogger(__name__)
def fetch_annotated(
db_connector,
annotated_only: bool = True,
model_version: Optional[str] = None,
date_from: Optional[str] = None,
date_to: Optional[str] = None,
ground_truth: Optional[bool] = None,
) -> List[Dict[str, Any]]:
"""Query the review_queue and return matching records as plain dicts."""
clauses: List[str] = []
params: List[Any] = []
if annotated_only:
clauses.append("rq.annotated = TRUE")
if model_version:
clauses.append("v.model_version = %s")
params.append(model_version)
if date_from:
clauses.append("rq.annotated_at >= %s")
params.append(date_from)
if date_to:
clauses.append("rq.annotated_at < %s")
params.append(date_to)
if ground_truth is not None:
clauses.append("rq.ground_truth = %s")
params.append(bool(ground_truth))
where = ("WHERE " + " AND ".join(clauses)) if clauses else ""
rows = db_connector.fetchall(
f"""SELECT rq.video_id, v.file_path, rq.confidence_score, rq.routing_decision,
v.model_version, rq.ground_truth, rq.annotated_at, rq.notes,
pl.confidence_scores
FROM review_queue rq
JOIN videos v ON v.id = rq.video_id
LEFT JOIN (
SELECT video_id, confidence_scores,
ROW_NUMBER() OVER (PARTITION BY video_id ORDER BY processed_at DESC) rn
FROM processing_logs
) pl ON pl.video_id = rq.video_id AND pl.rn = 1
{where}
ORDER BY rq.annotated_at DESC""",
params if params else None,
)
return [_normalise(row) for row in (rows or [])]
def _normalise(row: dict) -> Dict[str, Any]:
try:
scores = json.loads(row["confidence_scores"]) if row.get("confidence_scores") else []
except (json.JSONDecodeError, TypeError):
scores = []
at = row.get("annotated_at")
return {
"video_id": row["video_id"],
"file_path": row["file_path"],
"confidence_score": row["confidence_score"],
"routing_decision": row["routing_decision"],
"model_version": row.get("model_version"),
"ground_truth": bool(row["ground_truth"]) if row["ground_truth"] is not None else None,
"annotated_at": at.isoformat() if isinstance(at, datetime) else at,
"notes": row.get("notes"),
"contributing_frames": scores,
}
def export_json(records: List[Dict[str, Any]], output_path: str) -> str:
"""Write records to a UTF-8 JSON file; returns the path."""
path = Path(output_path)
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(records, default=str, indent=2), encoding="utf-8")
logger.info("Exported %d records to %s", len(records), path)
return str(path)
def export_csv(records: List[Dict[str, Any]], output_path: str) -> str:
"""Write records to a UTF-8 CSV file; returns the path."""
path = Path(output_path)
path.parent.mkdir(parents=True, exist_ok=True)
if not records:
path.write_text("", encoding="utf-8")
return str(path)
buf = io.StringIO()
writer = csv.DictWriter(buf, fieldnames=list(records[0].keys()))
writer.writeheader()
for r in records:
row = dict(r)
row["contributing_frames"] = json.dumps(row["contributing_frames"])
writer.writerow(row)
path.write_text(buf.getvalue(), encoding="utf-8")
logger.info("Exported %d records to %s", len(records), path)
return str(path)
src/scanner.py
-537
View File
@@ -1,537 +0,0 @@
"""
Directory Scanner Service
Walks /data/input to discover new/modified files, computes hashes,
probes video metadata, validates codecs, and syncs state to MariaDB.
Supports incremental scanning for efficient 30TB corpus handling.
"""
import hashlib
import json
import logging
import os
import socket
import threading
import time
import uuid
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Dict, List, Optional, Tuple
from codec_validator import CodecValidator
from prober import VideoMetadata, VideoProber
logger = logging.getLogger(__name__)
class DirectoryScanner:
"""Scan directories for new/modified video files and sync to DB."""
VIDEO_EXTENSIONS = {'.mp4', '.mov', '.avi', '.mkv', '.webm', '.flv', '.wmv'}
def __init__(
self,
db_connector,
config: dict,
input_path: str = "/data/input",
scan_interval: int = 60,
walker_threads: int = 8,
):
self.db = db_connector
self.config = config
self.input_path = Path(input_path)
self.scan_interval = scan_interval
self.walker_threads = walker_threads
self.prober = VideoProber(
timeout=config.get("scanner", {}).get("ffprobe_timeout_seconds", 10)
)
self.codec_validator = CodecValidator(
whitelist=set(config.get("codec", {}).get("whitelist", [])),
default_status_on_error=config.get("codec", {}).get("default_status_on_error", "UNSCANNABLE"),
)
self._running = False
self._scan_count = 0
self._total_files_discovered = 0
self._total_files_new = 0
self._total_files_modified = 0
self._total_files_unscannable = 0
# Single-instance guard: prevents overlapping scans both within this
# process (threading lock) and across replicas (DB lock with a lease).
self._instance_id = f"{socket.gethostname()}-{os.getpid()}-{uuid.uuid4().hex[:8]}"
self._lock_name = "directory_scanner"
self._lock_lease_seconds = int(
config.get("scanner", {}).get("lock_lease_seconds", 21600)
)
self._heartbeat_every = int(
config.get("scanner", {}).get("heartbeat_interval_files", 500)
)
self._scan_lock = threading.Lock()
self._lock_table_ensured = False
def start(self):
"""Start the scanner loop.
Each cycle is guarded so that at most one scan runs at a time:
- an in-process threading lock prevents re-entrant scans, and
- a DB lock (with a lease) prevents overlapping scans across replicas.
A long-running scan keeps its lease alive via heartbeats, so the next
scheduled tick (or another replica) waits instead of starting a second
parallel scan.
"""
self._running = True
logger.info(
"Scanner starting: input_path=%s interval=%ds threads=%d instance=%s",
self.input_path, self.scan_interval, self.walker_threads, self._instance_id,
)
while self._running:
# In-process re-entrancy guard: never run two scans at once.
if not self._scan_lock.acquire(blocking=False):
logger.warning("A scan is already in progress; skipping this cycle.")
self._sleep_interval()
continue
try:
if self.acquire_lock():
try:
self._run_scan()
finally:
self.release_lock()
else:
logger.info(
"Scanner lock held by another instance; skipping this cycle."
)
except Exception as e:
logger.error("Scanner error: %s", e, exc_info=True)
finally:
self._scan_lock.release()
# Sleep until next scan
self._sleep_interval()
def _sleep_interval(self):
"""Sleep for the scan interval, waking early if stopped."""
for _ in range(self.scan_interval):
if not self._running:
break
time.sleep(1)
def stop(self):
"""Stop the scanner."""
self._running = False
logger.info("Scanner stopping. Total scans: %d", self._scan_count)
# ------------------------------------------------------------------
# Single-instance lock (cross-process / cross-replica guard)
# ------------------------------------------------------------------
def _ensure_lock_table(self):
"""Create the scanner_lock table if it does not already exist."""
if self._lock_table_ensured:
return
self.db.execute(
"""CREATE TABLE IF NOT EXISTS scanner_lock (
lock_name VARCHAR(64) PRIMARY KEY,
owner VARCHAR(128) NOT NULL,
locked_at DATETIME NOT NULL,
lease_seconds INT NOT NULL DEFAULT 21600
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci"""
)
self._lock_table_ensured = True
def acquire_lock(self) -> bool:
"""Attempt to acquire the cross-process scanner lock.
Returns True if this instance now owns the lock, False otherwise.
A stale lock (held longer than the lease) is taken over so a crashed
scanner does not block scanning forever.
"""
try:
self._ensure_lock_table()
except Exception as e:
logger.warning("Could not ensure scanner_lock table: %s", e)
return True # fail-open: keep scanning rather than block entirely
now = datetime.now(timezone.utc).replace(tzinfo=None)
# 1) Try to insert a fresh lock row.
try:
self.db.execute(
"""INSERT INTO scanner_lock (lock_name, owner, locked_at, lease_seconds)
VALUES (%s, %s, %s, %s)""",
(self._lock_name, self._instance_id, now, self._lock_lease_seconds),
transaction=True,
)
logger.info("Acquired scanner lock (fresh). owner=%s", self._instance_id)
return True
except Exception:
# Row already exists -> fall through to steal-if-stale.
pass
# 2) Take over the lock if it is stale or already ours.
stale_before = now - timedelta(seconds=self._lock_lease_seconds)
try:
affected = self.db.execute(
"""UPDATE scanner_lock
SET owner = %s, locked_at = %s
WHERE lock_name = %s
AND (owner = %s OR locked_at < %s)""",
(self._instance_id, now, self._lock_name, self._instance_id, stale_before),
transaction=True,
)
if affected and affected > 0:
logger.info("Acquired scanner lock (stale takeover). owner=%s",
self._instance_id)
return True
except Exception as e:
logger.warning("Failed to check scanner lock: %s", e)
return True # fail-open
logger.info("Scanner lock held by another instance; not acquiring.")
return False
def release_lock(self):
"""Release the scanner lock if we own it."""
try:
self.db.execute(
"""DELETE FROM scanner_lock WHERE lock_name = %s AND owner = %s""",
(self._lock_name, self._instance_id),
transaction=True,
)
except Exception as e:
logger.warning("Failed to release scanner lock: %s", e)
def _heartbeat(self):
"""Refresh the lock lease so a long-running scan is not stolen."""
try:
now = datetime.now(timezone.utc).replace(tzinfo=None)
self.db.execute(
"""UPDATE scanner_lock SET locked_at = %s
WHERE lock_name = %s AND owner = %s""",
(now, self._lock_name, self._instance_id),
transaction=True,
)
except Exception as e:
logger.debug("Scanner lock heartbeat failed: %s", e)
def _run_scan(self):
"""Execute a single scan cycle."""
scan_start = time.time()
self._scan_count += 1
logger.info("Scan #%d starting...", self._scan_count)
# Get last scan time from DB
last_scan = self._get_last_scan_time()
# Discover files
files_to_process, files_removed = self._discover_files(last_scan)
# Process files in parallel
results = self._process_files_batch(files_to_process)
# Update DB
scan_result = self._update_db(results, files_removed)
# Record scan history
duration = time.time() - scan_start
self._record_scan_history(scan_result, duration)
logger.info(
"Scan #%d complete: discovered=%d new=%d modified=%d unscannable=%d removed=%d duration=%.1fs",
self._scan_count,
scan_result["discovered"],
scan_result["new"],
scan_result["modified"],
scan_result["unscannable"],
len(files_removed),
duration,
)
def _get_last_scan_time(self) -> Optional[datetime]:
"""Get the last scan time from DB."""
result = self.db.fetchone(
"SELECT scan_end FROM scan_history ORDER BY id DESC LIMIT 1"
)
if result and result.get("scan_end"):
return result["scan_end"]
return None
def _discover_files(self, last_scan: Optional[datetime]) -> Tuple[List[Path], List[str]]:
"""Discover new and modified files in the input directory."""
files_to_process = []
files_removed = []
if not self.input_path.exists():
logger.warning("Input path does not exist: %s", self.input_path)
return files_to_process, files_removed
# Get files from DB for comparison
if last_scan:
# Only check files modified after last scan
db_files = self.db.fetchall(
"SELECT file_path, last_scan_time FROM videos WHERE last_scan_time > %s",
(last_scan,),
)
db_paths = {row["file_path"] for row in db_files}
# Check for modified files
for row in db_files:
file_path = Path(row["file_path"])
if file_path.exists() and file_path.is_file():
try:
mtime = datetime.fromtimestamp(
file_path.stat().st_mtime, tz=timezone.utc
)
if mtime > row["last_scan_time"]:
files_to_process.append(file_path)
except OSError:
pass
else:
# Full scan - walk the directory
logger.info("Full scan (no previous scan found). Walking %s...", self.input_path)
for root, dirs, filenames in os.walk(self.input_path):
for filename in filenames:
ext = Path(filename).suffix.lower()
if ext in self.VIDEO_EXTENSIONS:
file_path = Path(root) / filename
files_to_process.append(file_path)
# Check for removed files (if we have a last scan)
if last_scan:
all_db_files = self.db.fetchall(
"SELECT file_path FROM videos WHERE status NOT IN ('REMOVED', 'ERROR')",
)
current_paths = {str(f) for f in files_to_process}
for row in all_db_files:
fp = row["file_path"]
if fp not in current_paths and Path(fp).exists():
# File still exists but not in current walk - might be in a new directory
pass
elif fp not in current_paths and not Path(fp).exists():
files_removed.append(fp)
logger.info("Discovered %d files to process, %d files removed",
len(files_to_process), len(files_removed))
return files_to_process, files_removed
def _process_files_batch(self, files: List[Path]) -> List[dict]:
"""Process a batch of files in parallel."""
results = []
processed = 0
with ThreadPoolExecutor(max_workers=self.walker_threads) as executor:
future_to_file = {
executor.submit(self._process_single_file, f): f
for f in files
}
for future in as_completed(future_to_file):
file_path = future_to_file[future]
try:
result = future.result()
results.append(result)
except Exception as e:
logger.error("Error processing %s: %s", file_path, e, exc_info=True)
results.append({
"file_path": str(file_path),
"status": "ERROR",
"error_message": str(e),
})
processed += 1
# Keep the single-instance lock alive during long scans.
if self._heartbeat_every and processed % self._heartbeat_every == 0:
self._heartbeat()
return results
def _process_single_file(self, file_path: Path) -> dict:
"""Process a single file: hash, probe, validate, determine status."""
file_path_str = str(file_path)
result = {"file_path": file_path_str}
try:
# Compute hash
file_hash = self._compute_hash(file_path_str)
result["file_hash"] = file_hash
# Check for duplicate
existing = self.db.fetchone(
"SELECT id, status FROM videos WHERE file_hash = %s",
(file_hash,),
)
if existing:
result["status"] = existing["status"]
result["video_id"] = existing["id"]
result["action"] = "duplicate"
logger.debug("Duplicate file found: %s (id=%d, status=%s)",
file_path_str, existing["id"], existing["status"])
return result
# Probe video
metadata = self.prober.probe(file_path_str)
result.update(metadata.to_dict())
# Validate codec
if metadata.is_valid:
is_supported, reason = self.codec_validator.validate(metadata.codec)
if is_supported:
result["status"] = "PENDING"
result["action"] = "new"
self._total_files_new += 1
else:
result["status"] = "UNSCANNABLE"
result["error_message"] = f"Unsupported codec: {metadata.codec} ({reason})"
result["action"] = "unscannable"
self._total_files_unscannable += 1
logger.warning("Unsupported codec for %s: %s", file_path_str, metadata.codec)
else:
result["status"] = "UNSCANNABLE"
result["action"] = "unscannable"
self._total_files_unscannable += 1
except Exception as e:
result["status"] = "ERROR"
result["error_message"] = str(e)
logger.error("Error processing %s: %s", file_path_str, e, exc_info=True)
return result
def _compute_hash(self, file_path: str) -> str:
"""Compute SHA-256 hash of the first 1MB of a file."""
chunk_size = 1024 * 1024 # 1MB
sha256 = hashlib.sha256()
try:
with open(file_path, "rb") as f:
chunk = f.read(chunk_size)
if chunk:
sha256.update(chunk)
return sha256.hexdigest()
except (OSError, IOError) as e:
logger.error("Cannot hash file %s: %s", file_path, e)
return hashlib.sha256(file_path.encode()).hexdigest() # fallback
def _update_db(self, results: List[dict], files_removed: List[str]) -> dict:
"""Update database with scan results."""
stats = {"discovered": len(results), "new": 0, "modified": 0, "unscannable": 0}
# Batch insert new files
new_files = [r for r in results if r.get("action") == "new"]
if new_files:
self._batch_insert_new_files(new_files)
stats["new"] = len(new_files)
# Update unscannable files
unscannable = [r for r in results if r.get("action") == "unscannable" and r.get("video_id")]
if unscannable:
self._batch_update_status(unscannable, "UNSCANNABLE")
stats["unscannable"] = len(unscannable)
# Mark removed files
if files_removed:
self._mark_files_removed(files_removed)
stats["removed"] = len(files_removed)
return stats
def _batch_insert_new_files(self, files: List[dict]):
"""Batch insert new files into DB."""
now = datetime.now(timezone.utc).replace(tzinfo=None)
for file_info in files:
try:
self.db.execute(
"""INSERT INTO videos
(file_path, file_hash, resolution_w, resolution_h, codec,
duration, status, last_scan_time, created_at, updated_at)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)""",
(
file_info["file_path"],
file_info["file_hash"],
file_info.get("resolution_w"),
file_info.get("resolution_h"),
file_info.get("codec"),
file_info.get("duration"),
"PENDING",
now,
now,
now,
),
transaction=True,
)
except Exception as e:
logger.error("Failed to insert %s: %s", file_info["file_path"], e)
def _batch_update_status(self, files: List[dict], status: str):
"""Batch update file status."""
now = datetime.now(timezone.utc).replace(tzinfo=None)
for file_info in files:
video_id = file_info.get("video_id")
if video_id:
try:
self.db.execute(
"""UPDATE videos SET status = %s, error_message = %s,
last_scan_time = %s, updated_at = %s
WHERE id = %s""",
(status, file_info.get("error_message"), now, now, video_id),
transaction=True,
)
except Exception as e:
logger.error("Failed to update %s: %s", file_info["file_path"], e)
def _mark_files_removed(self, file_paths: List[str]):
"""Mark files as removed in DB."""
now = datetime.now(timezone.utc).replace(tzinfo=None)
for fp in file_paths:
try:
self.db.execute(
"UPDATE videos SET status = 'REMOVED', last_scan_time = %s, updated_at = %s WHERE file_path = %s",
(now, now, fp),
transaction=True,
)
except Exception as e:
logger.error("Failed to mark %s as removed: %s", fp, e)
def _record_scan_history(self, stats: dict, duration: float):
"""Record scan history in DB."""
now = datetime.now(timezone.utc).replace(tzinfo=None)
self.db.execute(
"""INSERT INTO scan_history
(scan_start, scan_end, files_discovered, files_new, files_modified,
files_removed, files_unscannable, duration_seconds, status)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)""",
(
now,
now,
stats.get("discovered", 0),
stats.get("new", 0),
stats.get("modified", 0),
stats.get("removed", 0),
stats.get("unscannable", 0),
duration,
"COMPLETED",
),
transaction=True,
)
def get_stats(self) -> dict:
"""Get scanner statistics."""
return {
"scan_count": self._scan_count,
"total_files_discovered": self._total_files_discovered,
"total_files_new": self._total_files_new,
"total_files_modified": self._total_files_modified,
"total_files_unscannable": self._total_files_unscannable,
"instance_id": self._instance_id,
"scan_interval_seconds": self.scan_interval,
"lock_lease_seconds": self._lock_lease_seconds,
"scan_in_progress": self._scan_lock.locked(),
"input_path": str(self.input_path),
"is_running": self._running,
}