import cv2 import numpy as np from ultralytics import YOLO from datetime import datetime import json from pathlib import Path from database import Database from energy_analyzer import EnergyAnalyzer from blockchain import BlockchainManager from config import config import os import uuid import copy class NumpyEncoder(json.JSONEncoder): """ Custom encoder for numpy data types """ def default(self, obj): if isinstance(obj, (np.int_, np.intc, np.intp, np.int8, np.int16, np.int32, np.int64, np.uint8, np.uint16, np.uint32, np.uint64)): return int(obj) elif isinstance(obj, (np.float_, np.float16, np.float32, np.float64)): return float(obj) elif isinstance(obj, (np.ndarray,)): return obj.tolist() elif isinstance(obj, (np.bool_)): return bool(obj) return json.JSONEncoder.default(self, obj) class CVProcessor: # Class-level cache for face cascades (shared across instances) _face_cascade = None _face_cascade_profile = None _qr_detector = None @classmethod def _get_face_cascade(cls): """Lazy load and cache face cascade classifier""" if cls._face_cascade is None: cls._face_cascade = cv2.CascadeClassifier(cv2.data.haarcascades + 'haarcascade_frontalface_default.xml') return cls._face_cascade @classmethod def _get_face_cascade_profile(cls): """Lazy load and cache profile face cascade classifier""" if cls._face_cascade_profile is None: cls._face_cascade_profile = cv2.CascadeClassifier(cv2.data.haarcascades + 'haarcascade_profileface.xml') return cls._face_cascade_profile @classmethod def _get_qr_detector(cls): """Lazy load and cache QR code detector""" if cls._qr_detector is None: cls._qr_detector = cv2.QRCodeDetector() return cls._qr_detector def __init__(self, use_database=True, room_id="CS_LAB_101", verify_location=True, optimization_mode=None, db_instance=None): # Lazy load YOLO model (deferred until first use) self.model = None # Force high-fidelity 'precision' mode in Production if not specified if optimization_mode is None: optimization_mode = 'precision' if config.is_production() else 'balanced' # Consistent path resolution relative to project root self.base_dir = config.BASE_DIR # Determine paths (Docker/Cloud vs Local) current_dir = Path(__file__).parent # Accuracy Strategy: Prefer 'Small' model over 'Nano' if available for higher accuracy # Check local/flat directory (Docker) first, then structured path if (current_dir / 'yolov8s.pt').exists(): model_s = current_dir / 'yolov8s.pt' else: model_s = self.base_dir / 'backend' / 'yolov8s.pt' if (current_dir / 'yolov8n.pt').exists(): model_n = current_dir / 'yolov8n.pt' else: model_n = self.base_dir / 'backend' / 'yolov8n.pt' if model_s.exists(): self._model_path = str(model_s) else: self._model_path = str(model_n) if config.is_production(): print("ℹ Tip: For absolute accuracy in Production, consider uploading 'yolov8s.pt' to the backend folder.") self.room_id = room_id self.department = room_id.split('_')[0] if '_' in room_id else 'GENERAL' self.verify_location = verify_location self.location_verified = False self.location_confidence = 0.5 # Optimization mode self.optimization_mode = optimization_mode self.set_thresholds_by_mode(optimization_mode) # Cache internal detectors self.qr_detector = self._get_qr_detector() self.face_cascade = self._get_face_cascade() self.face_cascade_profile = self._get_face_cascade_profile() # Tracking setup self.known_faces = {} self.person_counter = 0 self.person_logs = {} self.current_frame_number = 0 self.person_temporal_buffer = {} self.min_detections_for_verification = 5 self.last_seen_face_url = None # Buffer for negligence attribution self.kalman_filters = {} self.bg_subtractor = cv2.createBackgroundSubtractorMOG2(detectShadows=True) self.occupancy_buffer = [] self.occupancy_buffer_size = 5 # Path configuration # For serverless storage fallback is_vercel = os.environ.get('VERCEL') == '1' if is_vercel: self.face_db_path = Path('/tmp') / 'outputs' / 'face_database' else: self.face_db_path = self.base_dir / 'outputs' / 'face_database' self.face_db_path.mkdir(parents=True, exist_ok=True) is_vercel = os.environ.get('VERCEL') == '1' if is_vercel: self.faces_folder = Path('/tmp') / 'uploads' / 'faces' else: self.faces_folder = self.base_dir / 'uploads' / 'faces' self.faces_folder.mkdir(parents=True, exist_ok=True) # Database & Analytics self.use_database = use_database self.db = db_instance if db_instance else (Database() if use_database else None) if self.use_database and self.db: self._load_known_faces() self.energy_analyzer = EnergyAnalyzer(self.room_id, optimization_mode=optimization_mode) self.previous_devices_state = [] self.previous_occupancy = False self.previous_lights_on = False self.blockchain = BlockchainManager() def _load_known_faces(self): """Sync identities from DB""" try: persons = self.db.get_all_persons() for person in persons: person_id = person.person_id self.known_faces[person_id] = { 'histograms': [], 'last_bbox': None, 'frame_last_seen': 0, 'detection_count': person.total_detections, 'confidence_history': [0.8], 'wallet_address': person.wallet_address } if person_id.startswith('person_'): try: idx = int(person_id.split('_')[1]) self.person_counter = max(self.person_counter, idx + 1) except: pass except Exception as e: print(f"⚠ Sync warning: {e}") def _ensure_model_loaded(self): """Actual YOLOv8 Loading - Auto-downloads if missing""" if self.model is None: # If path doesn't exist, use name string to trigger auto-download load_path = self._model_path if os.path.exists(self._model_path) else "yolov8n.pt" print(f"🧠 Loading YOLOv8 Neural Engine ({load_path})...") self.model = YOLO(load_path) def set_thresholds_by_mode(self, mode): if mode == 'precision': self.yolo_conf_threshold = 0.35 self.person_match_threshold = 0.60 self.action_confidence_min = 0.80 elif mode == 'recall': self.yolo_conf_threshold = 0.15 self.person_match_threshold = 0.40 self.action_confidence_min = 0.60 else: self.yolo_conf_threshold = 0.28 self.person_match_threshold = 0.50 self.action_confidence_min = 0.70 def process_frame(self, frame): self._ensure_model_loaded() if self.verify_location and self.current_frame_number % 150 == 0: self.verify_room_location(frame) results = self.model(frame, verbose=False) return results[0] def _calculate_iou(self, box1, box2): x1_min, y1_min, x1_max, y1_max = box1 x2_min, y2_min, x2_max, y2_max = box2 xi1, yi1, xi2, yi2 = max(x1_min, x2_min), max(y1_min, y2_min), min(x1_max, x2_max), min(y1_max, y2_max) if xi2 < xi1 or yi2 < yi1: return 0.0 inter = (xi2 - xi1) * (yi2 - yi1) union = (x1_max - x1_min) * (y1_max - y1_min) + (x2_max - x2_min) * (y2_max - y2_min) - inter return inter / union if union > 0 else 0.0 def detect_occupancy(self, results): boxes = [] for box in results.boxes: if int(box.cls[0]) == 0: x1, y1, x2, y2 = box.xyxy[0].cpu().numpy() boxes.append([int(x1), int(y1), int(x2), int(y2)]) return len(boxes) > 0, len(boxes), boxes def detect_devices(self, results, frame=None): devices = [] # Expanded vocabulary for campus device detection device_classes = { 62: 'tv', 63: 'laptop', 64: 'mouse', 66: 'keyboard', 67: 'cell phone', 65: 'remote' } for box in results.boxes: cls_id = int(box.cls[0]) conf = float(box.conf[0]) if conf < self.yolo_conf_threshold: continue if cls_id in device_classes: x1, y1, x2, y2 = box.xyxy[0].cpu().numpy() dev_type = device_classes[cls_id] dev_id = f"{dev_type}_{int(x1)//20}_{int(y1)//20}" dev_info = {'type': dev_type, 'confidence': conf, 'bbox': [int(x1), int(y1), int(x2), int(y2)], 'device_id': dev_id} if frame is not None: state = self.energy_analyzer.detect_device_state(frame, dev_info['bbox'], dev_type, dev_id) dev_info.update(state) devices.append(dev_info) return devices def generate_event(self, occupancy, person_count, devices, person_boxes=None, video_file=None, frame_number=None, frame=None, duration_minutes=5.0): devices_on = [d for d in devices if d.get('state') == 'ON'] devices_off = [d for d in devices if d.get('state') == 'OFF'] lights_on = self.energy_analyzer.detect_lights_state(frame).get('lights_on', False) if frame is not None else False # New multi-action detection actions = self.energy_analyzer.detect_sustainable_action( devices, self.previous_devices_state, occupancy, self.previous_occupancy, person_boxes=person_boxes ) savings = self.energy_analyzer.calculate_energy_savings(devices_on, devices_off, duration_minutes=duration_minutes) # If no specific actions detected, we return an empty list (Optimization: Skip auditing neutral parts) if not actions: return [] # Create separate events for each action found events = [] for action in actions: evt = { "timestamp": datetime.now().isoformat(), "room_id": self.room_id, "overall_confidence": action.get('confidence', 0.9), "occupancy": bool(occupancy), "person_count": person_count, "devices_detected": devices, "devices_on": devices_on, "devices_off": devices_off, "lights_on": lights_on, "action_detected": action.get('name'), "action_type": action.get('action_type'), "energy_saved_estimate": savings.get('energy_saved_kwh', 0) / len(actions), # Split savings "blockchain_credits": action.get('credits', 0), "status": "verified" if action.get('credits', 0) > 0 else "pending", "actor_index": action.get('actor_index', -1), "device_id": action.get('device_id'), "video_file": video_file } # Neural Face Extraction: Capture headshot idx = action.get('actor_index', -1) # If specifically attributed to an actor, use their box. # If attributed to station operator but people are present, use the most prominent person. target_idx = idx if (idx != -1) else (0 if (person_boxes and len(person_boxes) > 0) else -1) if target_idx != -1 and person_boxes and target_idx < len(person_boxes) and frame is not None: try: face_filename = f"face_actor_{target_idx}_{uuid.uuid4().hex[:6]}.jpg" face_path = self.faces_folder / face_filename if self.extract_actor_face(frame, person_boxes[target_idx], str(face_path)): evt['actor_face_url'] = f"faces/{face_filename}" self.last_seen_face_url = evt['actor_face_url'] # Cache for negligence fallback # print(f"DEBUG: Neural Face Extracted: {evt['actor_face_url']}") except Exception as e: print(f"⚠️ Face extraction error: {e}") elif idx == -1 and self.last_seen_face_url: # Attribution fallback for Room Exit Negligence (use last person who was in the room) evt['actor_face_url'] = self.last_seen_face_url events.append(evt) return events def extract_actor_face(self, frame, person_box, output_path): """Extract a high-fidelity facial crop from a detected person box using multi-stage CV""" try: x1, y1, x2, y2 = person_box # Ensure coordinates are within frame boundaries h, w = frame.shape[:2] x1, y1 = max(0, x1), max(0, y1) x2, y2 = min(w, x2), min(h, y2) # Focus on the head region (top 35% of the person box) head_h = int((y2 - y1) * 0.35) head_crop = frame[y1:min(h, y1 + head_h), x1:x2] if head_crop.size == 0: # print(f"DEBUG: Head crop is empty for box {person_box}") return False # Pre-processing for better detection in varying light gray = cv2.cvtColor(head_crop, cv2.COLOR_BGR2GRAY) gray = cv2.equalizeHist(gray) # Normalize contrast face_cascade = self._get_face_cascade() profile_cascade = self._get_face_cascade_profile() # Attempt 1: Frontal Face faces = face_cascade.detectMultiScale(gray, 1.1, 4) if not face_cascade.empty() else [] # Attempt 2: Profile Face (if frontal fails) if len(faces) == 0 and not profile_cascade.empty(): faces = profile_cascade.detectMultiScale(gray, 1.1, 4) if len(faces) > 0: # Use detected face region fx, fy, fw, fh = faces[0] # Add 25% padding for better UI aesthetics pad_w = int(fw * 0.25) pad_h = int(fh * 0.25) crop = head_crop[max(0, fy-pad_h):min(head_crop.shape[0], fy+fh+pad_h), max(0, fx-pad_w):min(head_crop.shape[1], fx+fw+pad_w)] else: # Fallback: Use the centered top-half of the head region as the face thumbprint # This ensures we always have a recognizable "who" even if they are facing away cw, ch = head_crop.shape[1], head_crop.shape[0] crop_w = int(cw * 0.8) crop_h = int(ch * 0.8) start_x = (cw - crop_w) // 2 start_y = (ch - crop_h) // 2 crop = head_crop[start_y:start_y+crop_h, start_x:start_x+crop_w] if crop.size > 0: # Normalize to 256x256 for consistent high-fidelity UI rendering final = cv2.resize(crop, (256, 256), interpolation=cv2.INTER_CUBIC) cv2.imwrite(output_path, final) return True except Exception as e: print(f"⚠️ Neural face extraction failed: {e}") return False def process_video(self, video_path, output_json_path=None, confidence_threshold=0.5, skip_frames=None, progress_callback=None): """ Process a video file and detect events Args: video_path: Path to video file output_json_path: Path to save JSON results (optional) confidence_threshold: Minimum confidence threshold skip_frames: Number of frames to skip (1 = process all, None = use default/optimized) progress_callback: Optional function called with (current_frame, total_frames, percentage) Returns: Dictionary with processing results """ self._ensure_model_loaded() cap = cv2.VideoCapture(str(video_path)) if not cap.isOpened(): raise ValueError(f"Could not open video file: {video_path}") total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT)) if total_frames <= 0: # Fallback for streams or malformed files total_frames = 1000 fps = cap.get(cv2.CAP_PROP_FPS) # EXPERT CV LOGIC: Human actions (flipping switches, entering rooms) # typically occur over 0.3s to 1.5s. # Sampling at ~2Hz to ~5Hz is the "Goldilocks" zone for temporal fidelity. # Determine skip interval: if skip_frames is None: # Optimize based on FPS to maintain a consistent temporal resolution actual_fps = fps if (fps and fps > 0) else 30.0 if self.optimization_mode == 'precision': # ~5 samples per second (0.2s resolution) - absolute precision skip_interval = max(1, int(actual_fps / 5)) elif self.optimization_mode == 'recall': # ~1 sample per second (1.0s resolution) - efficient detection skip_interval = max(1, int(actual_fps / 1)) else: # ~2 samples per second (0.5s resolution) - BALANCED EXPERT CHOICE # This is the industry standard for activity monitoring. skip_interval = max(1, int(actual_fps / 2)) else: skip_interval = max(1, int(skip_frames)) events = [] raw_sig_events = [] frame_number = 0 print(f"🎥 Processing video: {Path(video_path).name} ({total_frames} frames @ {fps}fps, interval: {skip_interval})") while cap.isOpened(): ret, frame = cap.read() if not ret: break frame_number += 1 self.current_frame_number = frame_number # Use dynamic interval skipping if frame_number > 1 and frame_number % skip_interval != 0: continue # Process frame results = self.process_frame(frame) # Detect occupancy and devices occupancy, person_count, person_boxes = self.detect_occupancy(results) devices = self.detect_devices(results, frame) # Calculate duration for this interval # If fps is valid, duration = interval/fps seconds. duration_sec = skip_interval / fps if fps and fps > 0 else 1.0 duration_min = duration_sec / 60.0 # 3. PERPETUAL FACIAL CACHING: Update 'last seen face' whenever students are in the frame. if person_boxes and frame is not None: try: # Refresh cache if missing or periodically to capture movement if self.last_seen_face_url is None or frame_number % (skip_interval * 10) == 0: face_filename = f"face_cache_{uuid.uuid4().hex[:6]}.jpg" face_path = self.faces_folder / face_filename if self.extract_actor_face(frame, person_boxes[0], str(face_path)): self.last_seen_face_url = f"faces/{face_filename}" # print(f"DEBUG: Cached face updated at frame {frame_number}") except: pass # Generate event(s) - now returns a list frame_events = self.generate_event( occupancy=occupancy, person_count=person_count, devices=devices, person_boxes=person_boxes, video_file=str(Path(video_path).name), frame_number=frame_number, frame=frame, duration_minutes=duration_min ) # FILTRATION LOGIC: Collect events with neural significance (credits != 0) # We don't extract yet; we collect for span grouping significant_events = [e for e in frame_events if e.get('blockchain_credits', 0) != 0] for sig_event in significant_events: raw_sig_events.append({ 'frame': frame_number, 'data': sig_event }) # Update state self.previous_devices_state = devices self.previous_occupancy = occupancy # Report progress if frame_number % (skip_interval * 10) == 0 or frame_number == total_frames: progress_pct = int(min(1, frame_number/total_frames) * 100) if total_frames > 0 else 0 print(f" Processed {frame_number}/{total_frames} frames ({progress_pct}%)") if progress_callback: progress_callback(frame_number, total_frames, progress_pct) cap.release() # POST-PROCESSING: Group continuous actions into logical 'Impact Spans' final_audited_events = [] if raw_sig_events: # Sort by frame raw_sig_events.sort(key=lambda x: x['frame']) spans = [] current_span = None # Grouping Logic: Any significant actions within 5 seconds of each other # This creates a "Scene" that might contain multiple people/actions for item in raw_sig_events: f, data = item['frame'], item['data'] if current_span and (f - current_span['end_frame']) <= (actual_fps * 5): current_span['end_frame'] = f # Track individual contributions in this span actor_key = str(data.get('actor_index', -1)) if actor_key not in current_span['contributors']: current_span['contributors'][actor_key] = { 'actor_index': data.get('actor_index', -1), 'actions': [], 'total_credits': 0.0, 'energy_saved': 0.0 } contrib = current_span['contributors'][actor_key] contrib['actions'].append(data['action_detected']) contrib['total_credits'] += data.get('blockchain_credits', 0.0) contrib['energy_saved'] += data.get('energy_saved_estimate', 0.0) if not contrib.get('face_url') and data.get('actor_face_url'): contrib['face_url'] = data.get('actor_face_url') current_span['total_credits'] += data.get('blockchain_credits', 0.0) else: if current_span: spans.append(current_span) actor_key = str(data.get('actor_index', -1)) current_span = { 'start_frame': f, 'end_frame': f, 'total_credits': data.get('blockchain_credits', 0.0), 'contributors': { actor_key: { 'actor_index': data.get('actor_index', -1), 'actions': [data['action_detected']], 'total_credits': data.get('blockchain_credits', 0.0), 'energy_saved': data.get('energy_saved_estimate', 0.0), 'face_url': data.get('actor_face_url') } }, 'base_data': copy.deepcopy(data) } if current_span: spans.append(current_span) # EXTRACTION & ATTRIBUTION: Extract evidence and finalize multi-user reports upload_dir = Path(video_path).parent for span in spans: # 2.5 sec padding for better human context start_f = max(0, span['start_frame'] - int(actual_fps * 2.5)) end_f = min(total_frames, span['end_frame'] + int(actual_fps * 2.5)) clip_filename = f"audit_scene_{uuid.uuid4().hex[:6]}.mp4" clip_path = upload_dir / clip_filename if self.extract_clip(str(video_path), str(clip_path), start_f, end_f): evt = span['base_data'] evt['video_file'] = clip_filename evt['blockchain_credits'] = round(span['total_credits'], 2) evt['frame_start'] = start_f evt['frame_end'] = end_f # Add detailed impact analytics for UI evt['impact_analytics'] = [ { 'actor_label': f"Student Node #{c['actor_index'] + 1}" if c['actor_index'] != -1 else "Station Operator", 'actor_face_url': c.get('face_url'), 'impact_actions': list(set(c['actions'])), 'credits': round(c['total_credits'], 2), 'energy_saved': round(c['energy_saved'], 4) } for c in span['contributors'].values() ] # Update summary label if multi-user if len(evt['impact_analytics']) > 1: evt['action_detected'] = "Multi-User Sustainability Event" final_audited_events.append(evt) # Update results with spans events = final_audited_events # Compile results results = { "video_file": None, # Source purged for optimization "audit_type": "Action Spans (Scliced Evidence)", "total_frames": total_frames, "frames_processed": frame_number, "fps": fps, "total_events": len(events), "events": events, "summary": { "occupancy_detected": sum(1 for e in events if e['occupancy']), "total_devices": sum(len(e['devices_detected']) for e in events), "energy_saved_kwh": sum(e['energy_saved_estimate'] for e in events), "credits_earned": sum(e['blockchain_credits'] for e in events) } } # Save to JSON if path provided if output_json_path: with open(output_json_path, 'w') as f: json.dump(results, f, cls=NumpyEncoder, indent=2) # Re-upload/Verify video files are correctly named as .mp4 for the frontend # The extract_clip might have changed extensions if it used AVI fallback previously. # But the metadata expects .mp4. We ensure consistency here. print(f"✅ Results saved to: {output_json_path}") print(f"✅ Video processing complete: {len(events)} significant events audited") # SPACE OPTIMIZATION: Purge the original source video after slicing evidence try: if os.path.exists(video_path): os.remove(video_path) print(f"🗑️ Cleaned up source: {Path(video_path).name}") except Exception as e: print(f"⚠️ Cleanup failed: {e}") return results def extract_clip(self, src_path, dst_path, start_frame, end_frame): """Extract a segment of video for high-fidelity evidence storage using ffmpeg. Guarantees H.264/MP4 compatibility for browsers using libx264 and yuv420p. """ try: # 1. Get FPS to calculate timestamps cap = cv2.VideoCapture(src_path) fps = cap.get(cv2.CAP_PROP_FPS) total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT)) cap.release() if fps <= 0: fps = 30.0 start_time = max(0, start_frame / fps) # Add a small buffer (0.5s) to duration to ensure the action is fully visible duration = max(0.5, (end_frame - start_frame) / fps + 0.5) # 2. Use ffmpeg directly for superior encoding compatibility import subprocess # Command optimized for: Small size, Web Streaming, Browser Compatibility cmd = [ 'ffmpeg', '-y', '-ss', str(start_time), # Seek before -i for speed '-t', str(duration), '-i', src_path, '-c:v', 'libx264', # H.264 Software Encoding '-preset', 'ultrafast', # Max speed for Real-time feels '-crf', '30', # Good compression '-pix_fmt', 'yuv420p', # ESSENTIAL: Most browsers only play yuv420p '-an', # Strip audio to save space '-movflags', '+faststart', # Allow video to start playing before fully downloaded dst_path ] print(f"🎬 Slicing Evidence: {Path(dst_path).name} ({start_time:.1f}s -> {start_time+duration:.1f}s)") try: # Run ffmpeg (suppress output unless error) subprocess.run(cmd, check=True, capture_output=True) if os.path.exists(dst_path) and os.path.getsize(dst_path) > 1000: return True except (subprocess.CalledProcessError, FileNotFoundError) as e: print(f"⚠️ ffmpeg extraction effort failed: {e}") except Exception as e: print(f"⚠️ Pre-extraction prep failed: {e}") # VERY LAST RESORT: OpenCV Fallback (Limited browser compatibility) print("🔄 Falling back to OpenCV software extraction...") cap = cv2.VideoCapture(src_path) if not cap.isOpened(): return False width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)) height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)) fps = cap.get(cv2.CAP_PROP_FPS) or 30 # mp4v is the most likely to work in a generic .mp4 container fourcc = cv2.VideoWriter_fourcc(*'mp4v') writer = cv2.VideoWriter(dst_path, fourcc, fps, (width, height)) if not writer or not writer.isOpened(): cap.release() return False cap.set(cv2.CAP_PROP_POS_FRAMES, start_frame) written = 0 limit = int(end_frame - start_frame) + 30 # +1s buffer while written < limit: ret, frame = cap.read() if not ret: break writer.write(frame) written += 1 cap.release() writer.release() return written > 0 def verify_room_location(self, frame): data, _, _ = self.qr_detector.detectAndDecode(frame) if data and data.startswith('ROOM:'): if data.split(':', 1)[1] == self.room_id: self.location_verified = True self.location_confidence = 1.0 return self.location_verified