💡 Pro Tip: Try accessing this demo from your phone's browser. You can use your phone's camera to capture and analyze egocentric videos, making the experience more interactive and personal.
EgoGPT-7B is built upon LLaVA-OV and has been finetuned on the EgoIT dataset and a partially de-identified EgoLife dataset. Its primary goal is to serve as an egocentric captioner, supporting EgoRAG for EgoLifeQA tasks. Please note that due to inherent biases in the EgoLife dataset, the model may occasionally hallucinate details about people in custom videos based on patterns from the training data (for example, describing someone as "wearing a blue t-shirt" or "with pink hair"). We are actively working on improving the model to make it more universally applicable and will continue to release updates regularly. If you're interested in contributing to the development of future iterations of EgoGPT or the EgoLife project, we welcome you to reach out and contact us. (Contact us at jingkang001@e.ntu.edu.sg)
"""
bibtext = """
### Citation
```
@inproceedings{yang2025egolife,
title={EgoLife: Towards Egocentric Life Assistant},
author={Yang, Jingkang and Liu, Shuai and Guo, Hongming and Dong, Yuhao and Zhang, Xiamengwei and Zhang, Sicheng and Wang, Pengyun and Zhou, Zitang and Xie, Binzhu and Wang, Ziyue and Ouyang, Bei and Lin, Zhengyu and Cominelli, Marco and Cai, Zhongang and Zhang, Yuanhan and Zhang, Peiyuan and Hong, Fangzhou and Widmer, Joerg and Gringoli, Francesco and Yang, Lei and Li, Bo and Liu, Ziwei},
booktitle={The IEEE/CVF Conference on Computer Vision and Pattern Recognition},
year={2025},
}
```
"""
# cur_dir = os.path.dirname(os.path.abspath(__file__))
cur_dir = '.'
# Add this after cur_dir definition
UPLOADS_DIR = os.path.join(cur_dir, "user_uploads")
os.makedirs(UPLOADS_DIR, exist_ok=True)
def time_to_frame_idx(time_int: int, fps: int) -> int:
"""
Convert time in HHMMSSFF format (integer or string) to frame index.
:param time_int: Time in HHMMSSFF format, e.g., 10483000 (10:48:30.00) or "10483000".
:param fps: Frames per second of the video.
:return: Frame index corresponding to the given time.
"""
# Ensure time_int is a string for slicing
time_str = str(time_int).zfill(
8) # Pad with zeros if necessary to ensure it's 8 digits
hours = int(time_str[:2])
minutes = int(time_str[2:4])
seconds = int(time_str[4:6])
frames = int(time_str[6:8])
total_seconds = hours * 3600 + minutes * 60 + seconds
total_frames = total_seconds * fps + frames # Convert to total frames
return total_frames
def split_text(text, keywords):
# 创建一个正则表达式模式，将所有关键词用 | 连接，并使用捕获组
pattern = '(' + '|'.join(map(re.escape, keywords)) + ')'
# 使用 re.split 保留分隔符
parts = re.split(pattern, text)
# 去除空字符串
parts = [part for part in parts if part]
return parts
warnings.filterwarnings("ignore")
# Create FastAPI instance
app = FastAPI()
def load_video(
video_path: Optional[str] = None,
max_frames_num: int = 16,
fps: int = 1,
video_start_time: Optional[float] = None,
start_time: Optional[float] = None,
end_time: Optional[float] = None,
time_based_processing: bool = False
) -> tuple:
vr = VideoReader(video_path, ctx=cpu(0), num_threads=1)
target_sr = 16000
# Process video frames first
if time_based_processing:
# Initialize video reader
vr = decord.VideoReader(video_path, ctx=decord.cpu(0), num_threads=1)
total_frame_num = len(vr)
video_fps = vr.get_avg_fps()
# Convert time to frame index based on the actual video FPS
video_start_frame = int(time_to_frame_idx(video_start_time, video_fps))
start_frame = int(time_to_frame_idx(start_time, video_fps))
end_frame = int(time_to_frame_idx(end_time, video_fps))
print("start frame", start_frame)
print("end frame", end_frame)
# Ensure the end time does not exceed the total frame number
if end_frame - start_frame > total_frame_num:
end_frame = total_frame_num + start_frame
# Adjust start_frame and end_frame based on video start time
start_frame -= video_start_frame
end_frame -= video_start_frame
start_frame = max(0, int(round(start_frame))) # 确保不会小于0
end_frame = min(total_frame_num, int(round(end_frame))) # 确保不会超过总帧数
start_frame = int(round(start_frame))
end_frame = int(round(end_frame))
# Sample frames based on the provided fps (e.g., 1 frame per second)
frame_idx = [i for i in range(start_frame, end_frame) if (i - start_frame) % int(video_fps / fps) == 0]
# Get the video frames for the sampled indices
video = vr.get_batch(frame_idx).asnumpy()
else:
# Original video processing logic
total_frame_num = len(vr)
avg_fps = round(vr.get_avg_fps() / fps)
frame_idx = [i for i in range(0, total_frame_num, avg_fps)]
if max_frames_num > 0:
if len(frame_idx) > max_frames_num:
uniform_sampled_frames = np.linspace(0, total_frame_num - 1, max_frames_num, dtype=int)
frame_idx = uniform_sampled_frames.tolist()
video = vr.get_batch(frame_idx).asnumpy()
# Try to load audio, return None for speech if failed
try:
if time_based_processing:
y, _ = librosa.load(video_path, sr=target_sr)
start_sample = int(start_time * target_sr)
end_sample = int(end_time * target_sr)
speech = y[start_sample:end_sample]
else:
speech, _ = librosa.load(video_path, sr=target_sr)
# Process audio if it exists
speech = whisper.pad_or_trim(speech.astype(np.float32))
speech = whisper.log_mel_spectrogram(speech, n_mels=128).permute(1, 0)
speech_lengths = torch.LongTensor([speech.shape[0]])
return video, speech, speech_lengths, True # True indicates real audio
except Exception as e:
print(f"Warning: Could not load audio from video: {e}")
# Create dummy silent audio
duration = 10 # 10 seconds
speech = np.zeros(duration * target_sr, dtype=np.float32)
speech = whisper.pad_or_trim(speech)
speech = whisper.log_mel_spectrogram(speech, n_mels=128).permute(1, 0)
speech_lengths = torch.LongTensor([speech.shape[0]])
return video, speech, speech_lengths, False # False indicates no real audio
class PromptRequest(BaseModel):
prompt: str
video_path: str = None
max_frames_num: int = 16
fps: int = 1
video_start_time: float = None
start_time: float = None
end_time: float = None
time_based_processing: bool = False
# @spaces.GPU(duration=120)
def save_interaction(video_path, prompt, output, audio_path=None):
"""Save user interaction data and files"""
if not video_path:
return
# Create timestamped directory for this interaction
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
interaction_dir = os.path.join(UPLOADS_DIR, timestamp)
os.makedirs(interaction_dir, exist_ok=True)
# Copy video file
video_ext = os.path.splitext(video_path)[1]
new_video_path = os.path.join(interaction_dir, f"video{video_ext}")
shutil.copy2(video_path, new_video_path)
# Save metadata
metadata = {
"timestamp": timestamp,
"prompt": prompt,
"output": output,
"video_path": new_video_path,
}
# Only try to save audio if it's a file path (str), not audio data (tuple)
if audio_path and isinstance(audio_path, (str, bytes, os.PathLike)):
audio_ext = os.path.splitext(audio_path)[1]
new_audio_path = os.path.join(interaction_dir, f"audio{audio_ext}")
shutil.copy2(audio_path, new_audio_path)
metadata["audio_path"] = new_audio_path
with open(os.path.join(interaction_dir, "metadata.json"), "w") as f:
json.dump(metadata, f, indent=4)
def extract_audio_from_video(video_path, audio_path=None):
print('Processing audio from video...', video_path, audio_path)
if video_path is None:
return None
if isinstance(video_path, dict) and 'name' in video_path:
video_path = video_path['name']
try:
y, sr = librosa.load(video_path, sr=8000, mono=True, res_type='kaiser_fast')
# Check if the audio is silent
if np.abs(y).mean() < 0.001:
print("Video appears to be silent")
return None
return (sr, y)
except Exception as e:
print(f"Warning: Could not extract audio from video: {e}")
return None
import time
@spaces.GPU
def generate_text(video_path, audio_track, prompt):
streamer = TextIteratorStreamer(tokenizer, skip_special_tokens=True)
max_frames_num = 30
fps = 1
conv_template = "qwen_1_5"
if video_path is None and audio_track is None:
question = prompt
speech = None
speech_lengths = None
has_real_audio = False
image = None
image_sizes= None
modalities = ["image"]
image_tensor=None
# Load video and potentially audio
else:
video, speech, speech_lengths, has_real_audio = load_video(
video_path=video_path,
max_frames_num=max_frames_num,
fps=fps,
)
# Prepare the prompt based on whether we have real audio
if not has_real_audio:
question = f"