shivanis14/SeniorSafetyMonitoringSystem

1import gradio as gr
2import io
3import numpy as np
4import torch
5#from decord import cpu, VideoReader, bridge
6from transformers import AutoModelForCausalLM, AutoTokenizer, BitsAndBytesConfig
7from orb_motion_detection import detect_fast_motion
8import time, os
9
10def process_video(video, start_time, end_time, quant=8):
11    start = time.time()
12
13    output_dir = "motion_detection_results"
14    os.system(f"rm -rf {output_dir}")
15    os.system(f"mkdir {output_dir}")
16
17    DEVICE = 'cuda' if torch.cuda.is_available() else 'cpu'
18    TORCH_TYPE = torch.bfloat16 if torch.cuda.is_available() and torch.cuda.get_device_capability()[0] >= 8 else torch.float16
19
20    MODEL_PATH = "THUDM/cogvlm2-video-llama3-base"
21
22    if 'int4' in MODEL_PATH:
23        quant = 4
24
25    strategy = 'base' if 'cogvlm2-video-llama3-base' in MODEL_PATH else 'chat'
26    print(f"Using {strategy} model")
27
28    timestamps, fast_frames = detect_fast_motion(video.name, output_dir, end_time, start_time, motion_threshold=1.5)
29
30    history = []
31    if len(fast_frames) > 0:
32        video_data = np.array(fast_frames[0:min(48, len(fast_frames))])  # Shape: (num_frames, height, width, channels)
33        video_data = np.transpose(video_data, (3, 0, 1, 2))  # RGB channels first
34        video_tensor = torch.tensor(video_data)  # Convert to tensor
35
36        tokenizer = AutoTokenizer.from_pretrained(MODEL_PATH, trust_remote_code=True)
37
38        if quant == 4:
39            model = AutoModelForCausalLM.from_pretrained(
40                MODEL_PATH,
41                torch_dtype=TORCH_TYPE,
42                trust_remote_code=True,
43                quantization_config=BitsAndBytesConfig(
44                    load_in_4bit=True,
45                    bnb_4bit_compute_dtype=TORCH_TYPE,
46                ),
47                low_cpu_mem_usage=True
48            ).eval()
49        elif quant == 8:
50            model = AutoModelForCausalLM.from_pretrained(
51                MODEL_PATH,
52                torch_dtype=TORCH_TYPE,
53                trust_remote_code=True,
54                quantization_config=BitsAndBytesConfig(
55                    load_in_8bit=True,
56                    bnb_4bit_compute_dtype=TORCH_TYPE,
57                ),
58                low_cpu_mem_usage=True
59            ).eval()
60        else:
61            model = AutoModelForCausalLM.from_pretrained(
62                MODEL_PATH,
63                torch_dtype=TORCH_TYPE,
64                trust_remote_code=True
65            ).eval().to(DEVICE)
66
67        query = "Describe the actions in the video frames focusing on physical abuse, violence, or someone falling down."
68        print(f"Query: {query}")
69
70        inputs = model.build_conversation_input_ids(
71            tokenizer=tokenizer,
72            query=query,
73            images=[video_tensor],
74            history=history,
75            template_version=strategy
76        )
77
78        inputs = {
79            'input_ids': inputs['input_ids'].unsqueeze(0).to(DEVICE),
80            'token_type_ids': inputs['token_type_ids'].unsqueeze(0).to(DEVICE),
81            'attention_mask': inputs['attention_mask'].unsqueeze(0).to(DEVICE),
82            'images': [[inputs['images'][0].to('cuda').to(TORCH_TYPE)]],
83        }
84
85        gen_kwargs = {
86            "max_new_tokens": 2048,
87            "pad_token_id": 128002,
88            "top_k": 1,
89            "do_sample": True,
90            "top_p": 0.1,
91            "temperature": 0.1,
92        }
93
94        with torch.no_grad():
95            outputs = model.generate(**inputs, **gen_kwargs)
96            outputs = outputs[:, inputs['input_ids'].shape[1]:]
97            response = tokenizer.decode(outputs[0], skip_special_tokens=True)
98            print("\nCogVLM2-Video:", response)
99        history.append((query, response))
100
101        result = f"Response: {response}"
102    else:
103        result = "No aggressive behaviour found. Nobody falling down."
104
105    end = time.time()
106    execution_time = f"Execution time for {video.name}: {end - start} seconds. Duration of the video was {end_time - start_time} seconds."
107
108    return result
109
110
111# Create Gradio Interface
112def gradio_interface():
113    video_input = gr.File(label="Upload video file (.mp4)", type="filepath")
114    start_time = gr.Number(value=0.0, label="Start time (seconds)")
115    end_time = gr.Number(value=15.0, label="End time (seconds)")
116
117    interface = gr.Interface(
118        fn=process_video,
119        inputs=[video_input, start_time, end_time],
120        outputs="text",
121        title="Senior Safety Monitoring System",
122        description="Upload a video and specify the time range for analysis. The model will detect fast motion and describe actions such as physical abuse or someone falling down."
123    )
124
125    interface.launch(share=True)
126
127
128if __name__ == "__main__":
129    gradio_interface()