image_captioning.py

import torch
from transformers import AutoModel, AutoTokenizer
from modelscope.hub.snapshot_download import snapshot_download
from PIL import Image
from decord import VideoReader, cpu
import os
import gc
import cv2
import tempfile
import shutil
import subprocess
from yolo_detection import is_image, is_video

# Constants for video processing
MAX_NUM_FRAMES = 32

# Check if CUDA is available
DEVICE = "cuda" if torch.cuda.is_available() else "cpu"
global TOTAL_CHUNKS 
TOTAL_CHUNKS  = 1

# Initialize GPU if available
if DEVICE == "cuda":
    def debug():
        torch.randn(10).cuda()
    debug()

# Model configuration
MODEL_NAME = 'iic/mPLUG-Owl3-7B-240728'
MODEL_CACHE_DIR = "/data/models"
os.makedirs(MODEL_CACHE_DIR, exist_ok=True)

# Download and cache the model
try:
    model_path = snapshot_download(MODEL_NAME, cache_dir=MODEL_CACHE_DIR)
except Exception as e:
    print(f"Error downloading model: {str(e)}")
    model_path = os.path.join(MODEL_CACHE_DIR, MODEL_NAME)


# Model configuration and existing functions remain unchanged...
def load_model_and_tokenizer():
    """Load a fresh instance of the model and tokenizer"""
    try:
        # Clear GPU memory if using CUDA
        if DEVICE == "cuda":
            torch.cuda.empty_cache()
            gc.collect()
        
        model = AutoModel.from_pretrained(
            model_path,
            attn_implementation='sdpa',
            trust_remote_code=True,
            torch_dtype=torch.half,
            device_map='auto'
        )

        tokenizer = AutoTokenizer.from_pretrained(
            model_path,
            trust_remote_code=True
        )
        model.eval()
        processor = model.init_processor(tokenizer)
        return model, tokenizer, processor
    except Exception as e:
        print(f"Error loading model: {str(e)}")
        raise

def process_image(image_path, model, tokenizer, processor, prompt):
    """Process single image with mPLUG model"""
    try:
        image = Image.open(image_path)
        messages = [{
            "role": "user",
            "content": prompt,
            "images": [image]
        }]

        model_messages = []
        images = []

        for msg in messages:
            content_str = msg["content"]
            if "images" in msg and msg["images"]:
                content_str += "<|image|>"
                images.extend(msg["images"])
            model_messages.append({
                "role": msg["role"],
                "content": content_str
            })

        model_messages.append({
            "role": "assistant",
            "content": ""
        })

        inputs = processor(
            model_messages,
            images=images,
            videos=None
        )
        inputs.to('cuda')
        inputs.update({
            'tokenizer': tokenizer,
            'max_new_tokens': 100,
            'decode_text': True,
        })

        response = model.generate(**inputs)
        return response[0]
    except Exception as e:
        print(f"Error processing image: {str(e)}")
        return "Error processing image"

def process_video_chunk(video_frames, model, tokenizer, processor, prompt):
    """Process a chunk of video frames with mPLUG model"""
    messages = [
        {
            "role": "user",
            "content": prompt,
            "video_frames": video_frames
        }
    ]

    model_messages = []
    videos = []

    for msg in messages:
        content_str = msg["content"]
        if "video_frames" in msg and msg["video_frames"]:
            content_str += "<|video|>"
            videos.append(msg["video_frames"])
        model_messages.append({
            "role": msg["role"],
            "content": content_str
        })

    model_messages.append({
        "role": "assistant",
        "content": ""
    })

    inputs = processor(
        model_messages,
        images=None,
        videos=videos if videos else None
    )
    inputs.to('cuda')
    inputs.update({
        'tokenizer': tokenizer,
        'max_new_tokens': 100,
        'decode_text': True,
    })

    response = model.generate(**inputs)
    return response[0]

def split_original_video(video_path, chunk_info):
    """Split original video into chunks using precise timestamps"""
    original_chunks = []
    tmp_dir = os.path.join('.', 'tmp')
    
    for chunk in chunk_info:
        output_path = os.path.join(tmp_dir, f"original_chunk_{chunk['chunk_id']}.mp4")
        # Use ffmpeg for precise splitting without re-encoding
        cmd = [
            'ffmpeg',
            '-ss', str(chunk['start_time']),
            '-to', str(chunk['end_time']),
            '-i', video_path,
            '-c', 'copy',
            output_path
        ]
        subprocess.run(cmd, check=True)
        original_chunks.append(output_path)
    
    return original_chunks

def encode_video_in_chunks(video_path):
    """Extract frames from a video in chunks and save chunks to disk"""
    global TOTAL_CHUNKS
    vr = VideoReader(video_path, ctx=cpu(0))
    original_fps = vr.get_avg_fps()
    sample_fps = round(original_fps / 1)  # 1 FPS
    frame_idx = [i for i in range(0, len(vr), sample_fps)]
    fps = vr.get_avg_fps()

    # Create tmp directory if it doesn't exist
    tmp_dir = os.path.join('.', 'tmp')
    os.makedirs(tmp_dir, exist_ok=True)

    # Split frame indices into chunks
    chunks = [
        frame_idx[i:i + MAX_NUM_FRAMES]
        for i in range(0, len(frame_idx), MAX_NUM_FRAMES)
    ]
    
    # Set global TOTAL_CHUNKS before processing
    TOTAL_CHUNKS = len(chunks)
    print(f"Total chunks: {TOTAL_CHUNKS}")

    # Information about saved chunks
    chunk_info = []

    for chunk_idx, chunk in enumerate(chunks):
        # Get frames for this chunk
        frames = vr.get_batch(chunk).asnumpy()
        frames_pil = [Image.fromarray(v.astype('uint8')) for v in frames]
        
        # Save chunk as a video file
        chunk_path = os.path.join(tmp_dir, f"chunk_{chunk_idx}.mp4")
        
        # Calculate start and end times for this chunk
        if chunk:
            start_frame = chunk[0]
            end_frame = chunk[-1]
            start_time = start_frame / fps
            end_time = end_frame / fps
            
            # Save chunk info for later use
            chunk_info.append({
                'chunk_id': chunk_idx,
                'path': chunk_path,
                'start_time': start_time,
                'end_time': end_time,
                'start_frame': start_frame,
                'end_frame': end_frame,
                'original_fps': fps  # Use actual fps from video
            })
            
            # Use OpenCV to create video from frames
            height, width, _ = frames[0].shape
            fourcc = cv2.VideoWriter_fourcc(*'mp4v')
            out = cv2.VideoWriter(chunk_path, fourcc, fps, (width, height))
            
            for frame in frames:
                # Convert RGB to BGR (OpenCV format)
                frame_bgr = cv2.cvtColor(frame, cv2.COLOR_RGB2BGR)
                out.write(frame_bgr)
            
            out.release()
            print(f"Saved chunk {chunk_idx} to {chunk_path}")

        yield chunk_idx, frames_pil, chunk_info[-1] if chunk_info else None

    # Split original video after processing all chunks
    original_chunks = split_original_video(video_path, chunk_info)

def analyze_image_activities(image_path):
    """Analyze construction site image and generate activity description"""
    from datetime import datetime, timedelta
    try:
        # Sample structured response - Replace with actual model processing
        return [
            {
                'time': datetime.now().strftime("%I:%M %p"),
                'summary': 'Excavation work in progress',
                'objects': ['excavator', 'worker', 'dump-truck']
            },
            {
                'time': (datetime.now() - timedelta(minutes=30)).strftime("%I:%M %p"),
                'summary': 'Material loading operation',
                'objects': ['loader', 'worker', 'gravel']
            }
        ]
    except Exception as e:
        print(f"Error analyzing image: {str(e)}")
        return []  # Return empty list on error
    

def generate_thumbnails(video_path, num_chunks):
    """Extract thumbnails for each chunk
    Args:
        video_path: Path to video file
        num_chunks: Number of chunks to generate thumbnails for
    """
    vr = VideoReader(video_path, ctx=cpu(0))
    thumbnails = []
    total_frames = len(vr)
    
    # Create/clear tmp directory in current working directory
    tmp_dir = os.path.join('.', 'tmp')
    # Remove existing directory if it exists
    if os.path.exists(tmp_dir):
        shutil.rmtree(tmp_dir)
    os.makedirs(tmp_dir, exist_ok=True)
    
    # Calculate frame step size based on number of chunks
    frame_step = total_frames // num_chunks
    
    for chunk_idx in range(num_chunks):
        # Take frame at start of each chunk
        frame_idx = chunk_idx * frame_step
        if frame_idx < total_frames:
            frame = vr[frame_idx].asnumpy()
            img = Image.fromarray(frame)
            temp_path = os.path.join(tmp_dir, f"thumbnail_{chunk_idx}.jpg")
            img.save(temp_path)
            thumbnails.append({
                "path": temp_path,
                "time": frame_idx/vr.get_avg_fps()
            })
    
    return thumbnails

def analyze_video_activities(video_path):
    """Analyze video using mPLUG model with chunking"""
    global TOTAL_CHUNKS
    try:
        # Existing chunk processing
        all_activities = []
        # Calculate total chunks first
        vr = VideoReader(video_path, ctx=cpu(0))
        sample_fps = round(vr.get_avg_fps() / 1)
        frame_idx = [i for i in range(0, len(vr), sample_fps)]
        TOTAL_CHUNKS = len([frame_idx[i:i + MAX_NUM_FRAMES] 
                          for i in range(0, len(frame_idx), MAX_NUM_FRAMES)])
        
        # Generate thumbnails with known chunk count
        thumbnails = generate_thumbnails(video_path, num_chunks=TOTAL_CHUNKS)
        
        # Now process chunks
        chunk_generator = encode_video_in_chunks(video_path)
        model, tokenizer, processor = load_model_and_tokenizer()

        for chunk_idx, video_frames, chunk_info in chunk_generator:
            prompt = "Analyze this construction site video chunk and describe the activities happening. Focus on construction activities, machinery usage, and worker actions. Include any construction equipment or machinery you can identify."
            response = process_video_chunk(video_frames, model, tokenizer, processor, prompt)
            print(f"Chunk {chunk_idx}: {response}")
            
            # Map responses to thumbnails
            time_start = chunk_idx * MAX_NUM_FRAMES
            chunk_thumbnails = [t for t in thumbnails 
                              if time_start <= t['time'] < time_start + MAX_NUM_FRAMES]
            
            # Extract time from frame position
            for thumbnail in chunk_thumbnails:
                # Calculate timestamp in minutes:seconds format
                seconds = int(thumbnail['time'])
                minutes = seconds // 60
                seconds = seconds % 60
                timestamp = f"{minutes:02d}:{seconds:02d}"
                
                # Extract objects using basic text parsing from the response
                # In a production system, you might want to use more sophisticated NLP
                objects = []
                lower_response = response.lower()
                possible_objects = ["excavator", "bulldozer", "crane", "truck", "loader", 
                                   "worker", "concrete", "scaffold", "beam", "pipe", 
                                   "rebar", "formwork", "drill", "grader", "roller"]
                
                for obj in possible_objects:
                    if obj in lower_response:
                        objects.append(obj)
                
                activity = {
                    'time': timestamp,
                    'timestamp_seconds': thumbnail['time'],  # Store raw seconds for sorting
                    'summary': response,
                    'objects': objects,
                    'thumbnail': thumbnail["path"],
                    'chunk_id': chunk_idx,
                    'chunk_path': chunk_info['path'] if chunk_info else None
                }
                
                all_activities.append(activity)
        
        # Sort activities by timestamp
        all_activities.sort(key=lambda x: x['timestamp_seconds'])
        return all_activities
    except Exception as e:
        print(f"Error analyzing video: {str(e)}")
        return []  # Maintain consistent return type