`ultralytics 8.2.47` YOLOv8 zero-shot action recognition example (#13935)

Co-authored-by: UltralyticsAssistant <web@ultralytics.com>
Co-authored-by: Glenn Jocher <glenn.jocher@ultralytics.com>
pull/14107/head v8.2.47
fatih c. akyon 5 months ago committed by GitHub
parent 645c83671f
commit 7c1999929a
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
  1. 457
      examples/YOLOv8-Action-Recognition/action_recognition.py
  2. 116
      examples/YOLOv8-Action-Recognition/readme.md
  3. 4
      examples/YOLOv8-Action-Recognition/requirements.txt
  4. 2
      ultralytics/__init__.py

@ -0,0 +1,457 @@
# Ultralytics YOLO 🚀, AGPL-3.0 license
import argparse
import time
from collections import defaultdict
from typing import List, Optional, Tuple
from urllib.parse import urlparse
import cv2
import numpy as np
import torch
from transformers import AutoModel, AutoProcessor
from ultralytics import YOLO
from ultralytics.data.loaders import get_best_youtube_url
from ultralytics.utils.plotting import Annotator
from ultralytics.utils.torch_utils import select_device
class TorchVisionVideoClassifier:
from torchvision.models.video import (
MViT_V1_B_Weights,
MViT_V2_S_Weights,
R3D_18_Weights,
S3D_Weights,
Swin3D_B_Weights,
Swin3D_T_Weights,
mvit_v1_b,
mvit_v2_s,
r3d_18,
s3d,
swin3d_b,
swin3d_t,
)
model_name_to_model_and_weights = {
"s3d": (s3d, S3D_Weights.DEFAULT),
"r3d_18": (r3d_18, R3D_18_Weights.DEFAULT),
"swin3d_t": (swin3d_t, Swin3D_T_Weights.DEFAULT),
"swin3d_b": (swin3d_b, Swin3D_B_Weights.DEFAULT),
"mvit_v1_b": (mvit_v1_b, MViT_V1_B_Weights.DEFAULT),
"mvit_v2_s": (mvit_v2_s, MViT_V2_S_Weights.DEFAULT),
}
def __init__(self, model_name: str, device: str or torch.device = ""):
"""
Initialize the VideoClassifier with the specified model name and device.
Args:
model_name (str): The name of the model to use.
device (str or torch.device, optional): The device to run the model on. Defaults to "".
Raises:
ValueError: If an invalid model name is provided.
"""
if model_name not in self.model_name_to_model_and_weights:
raise ValueError(f"Invalid model name '{model_name}'. Available models: {self.available_model_names()}")
model, self.weights = self.model_name_to_model_and_weights[model_name]
self.device = select_device(device)
self.model = model(weights=self.weights).to(self.device).eval()
@staticmethod
def available_model_names() -> List[str]:
"""
Get the list of available model names.
Returns:
list: List of available model names.
"""
return list(TorchVisionVideoClassifier.model_name_to_model_and_weights.keys())
def preprocess_crops_for_video_cls(self, crops: List[np.ndarray], input_size: list = [224, 224]) -> torch.Tensor:
"""
Preprocess a list of crops for video classification.
Args:
crops (List[np.ndarray]): List of crops to preprocess. Each crop should have dimensions (H, W, C)
input_size (tuple, optional): The target input size for the model. Defaults to (224, 224).
Returns:
torch.Tensor: Preprocessed crops as a tensor with dimensions (1, T, C, H, W).
"""
from torchvision.transforms import v2
transform = v2.Compose(
[
v2.ToDtype(torch.float32, scale=True),
v2.Resize(input_size, antialias=True),
v2.Normalize(mean=self.weights.transforms().mean, std=self.weights.transforms().std),
]
)
processed_crops = [transform(torch.from_numpy(crop).permute(2, 0, 1)) for crop in crops]
return torch.stack(processed_crops).unsqueeze(0).permute(0, 2, 1, 3, 4).to(self.device)
def __call__(self, sequences: torch.Tensor):
"""
Perform inference on the given sequences.
Args:
sequences (torch.Tensor): The input sequences for the model. The expected input dimensions are
(B, T, C, H, W) for batched video frames or (T, C, H, W) for single video frames.
Returns:
torch.Tensor: The model's output.
"""
with torch.inference_mode():
return self.model(sequences)
def postprocess(self, outputs: torch.Tensor) -> Tuple[List[str], List[float]]:
"""
Postprocess the model's batch output.
Args:
outputs (torch.Tensor): The model's output.
Returns:
List[str]: The predicted labels.
List[float]: The predicted confidences.
"""
pred_labels = []
pred_confs = []
for output in outputs:
pred_class = output.argmax(0).item()
pred_label = self.weights.meta["categories"][pred_class]
pred_labels.append(pred_label)
pred_conf = output.softmax(0)[pred_class].item()
pred_confs.append(pred_conf)
return pred_labels, pred_confs
class HuggingFaceVideoClassifier:
def __init__(
self,
labels: List[str],
model_name: str = "microsoft/xclip-base-patch16-zero-shot",
device: str or torch.device = "",
fp16: bool = False,
):
"""
Initialize the HuggingFaceVideoClassifier with the specified model name.
Args:
labels (List[str]): List of labels for zero-shot classification.
model_name (str): The name of the model to use. Defaults to "microsoft/xclip-base-patch16-zero-shot".
device (str or torch.device, optional): The device to run the model on. Defaults to "".
fp16 (bool, optional): Whether to use FP16 for inference. Defaults to False.
"""
self.fp16 = fp16
self.labels = labels
self.device = select_device(device)
self.processor = AutoProcessor.from_pretrained(model_name)
model = AutoModel.from_pretrained(model_name).to(self.device)
if fp16:
model = model.half()
self.model = model.eval()
def preprocess_crops_for_video_cls(self, crops: List[np.ndarray], input_size: list = [224, 224]) -> torch.Tensor:
"""
Preprocess a list of crops for video classification.
Args:
crops (List[np.ndarray]): List of crops to preprocess. Each crop should have dimensions (H, W, C)
input_size (tuple, optional): The target input size for the model. Defaults to (224, 224).
Returns:
torch.Tensor: Preprocessed crops as a tensor (1, T, C, H, W).
"""
from torchvision.transforms import v2
transform = v2.Compose(
[
v2.ToDtype(torch.float32, scale=True),
v2.Resize(input_size, antialias=True),
v2.Normalize(
mean=self.processor.image_processor.image_mean, std=self.processor.image_processor.image_std
),
]
)
processed_crops = [transform(torch.from_numpy(crop).permute(2, 0, 1)) for crop in crops] # (T, C, H, W)
output = torch.stack(processed_crops).unsqueeze(0).to(self.device) # (1, T, C, H, W)
if self.fp16:
output = output.half()
return output
def __call__(self, sequences: torch.Tensor) -> torch.Tensor:
"""
Perform inference on the given sequences.
Args:
sequences (torch.Tensor): The input sequences for the model. Batched video frames with shape (B, T, H, W, C).
Returns:
torch.Tensor: The model's output.
"""
input_ids = self.processor(text=self.labels, return_tensors="pt", padding=True)["input_ids"].to(self.device)
inputs = {"pixel_values": sequences, "input_ids": input_ids}
with torch.inference_mode():
outputs = self.model(**inputs)
return outputs.logits_per_video
def postprocess(self, outputs: torch.Tensor) -> Tuple[List[List[str]], List[List[float]]]:
"""
Postprocess the model's batch output.
Args:
outputs (torch.Tensor): The model's output.
Returns:
List[List[str]]: The predicted top3 labels.
List[List[float]]: The predicted top3 confidences.
"""
pred_labels = []
pred_confs = []
with torch.no_grad():
logits_per_video = outputs # Assuming outputs is already the logits tensor
probs = logits_per_video.softmax(dim=-1) # Use softmax to convert logits to probabilities
for prob in probs:
top2_indices = prob.topk(2).indices.tolist()
top2_labels = [self.labels[idx] for idx in top2_indices]
top2_confs = prob[top2_indices].tolist()
pred_labels.append(top2_labels)
pred_confs.append(top2_confs)
return pred_labels, pred_confs
def crop_and_pad(frame, box, margin_percent):
"""Crop box with margin and take square crop from frame."""
x1, y1, x2, y2 = map(int, box)
w, h = x2 - x1, y2 - y1
# Add margin
margin_x, margin_y = int(w * margin_percent / 100), int(h * margin_percent / 100)
x1, y1 = max(0, x1 - margin_x), max(0, y1 - margin_y)
x2, y2 = min(frame.shape[1], x2 + margin_x), min(frame.shape[0], y2 + margin_y)
# Take square crop from frame
size = max(y2 - y1, x2 - x1)
center_y, center_x = (y1 + y2) // 2, (x1 + x2) // 2
half_size = size // 2
square_crop = frame[
max(0, center_y - half_size) : min(frame.shape[0], center_y + half_size),
max(0, center_x - half_size) : min(frame.shape[1], center_x + half_size),
]
return cv2.resize(square_crop, (224, 224), interpolation=cv2.INTER_LINEAR)
def run(
weights: str = "yolov8n.pt",
device: str = "",
source: str = "https://www.youtube.com/watch?v=dQw4w9WgXcQ",
output_path: Optional[str] = None,
crop_margin_percentage: int = 10,
num_video_sequence_samples: int = 8,
skip_frame: int = 2,
video_cls_overlap_ratio: float = 0.25,
fp16: bool = False,
video_classifier_model: str = "microsoft/xclip-base-patch32",
labels: List[str] = [
"walking",
"running",
"brushing teeth",
"looking into phone",
"weight lifting",
"cooking",
"sitting",
],
) -> None:
"""
Run action recognition on a video source using YOLO for object detection and a video classifier.
Args:
weights (str): Path to the YOLO model weights. Defaults to "yolov8n.pt".
device (str): Device to run the model on. Use 'cuda' for NVIDIA GPU, 'mps' for Apple Silicon, or 'cpu'. Defaults to auto-detection.
source (str): Path to mp4 video file or YouTube URL. Defaults to a sample YouTube video.
output_path (Optional[str], optional): Path to save the output video. Defaults to None.
crop_margin_percentage (int, optional): Percentage of margin to add around detected objects. Defaults to 10.
num_video_sequence_samples (int, optional): Number of video frames to use for classification. Defaults to 8.
skip_frame (int, optional): Number of frames to skip between detections. Defaults to 4.
video_cls_overlap_ratio (float, optional): Overlap ratio between video sequences. Defaults to 0.25.
fp16 (bool, optional): Whether to use half-precision floating point. Defaults to False.
video_classifier_model (str, optional): Name or path of the video classifier model. Defaults to "microsoft/xclip-base-patch32".
labels (List[str], optional): List of labels for zero-shot classification. Defaults to predefined list.
Returns:
None</edit>
"""
# Initialize models and device
device = select_device(device)
yolo_model = YOLO(weights).to(device)
if video_classifier_model in TorchVisionVideoClassifier.available_model_names():
print("'fp16' is not supported for TorchVisionVideoClassifier. Setting fp16 to False.")
print(
"'labels' is not used for TorchVisionVideoClassifier. Ignoring the provided labels and using Kinetics-400 labels."
)
video_classifier = TorchVisionVideoClassifier(video_classifier_model, device=device)
else:
video_classifier = HuggingFaceVideoClassifier(
labels, model_name=video_classifier_model, device=device, fp16=fp16
)
# Initialize video capture
if source.startswith("http") and urlparse(source).hostname in {"www.youtube.com", "youtube.com", "youtu.be"}:
source = get_best_youtube_url(source)
elif source.endswith(".mp4"):
pass
else:
raise ValueError("Invalid source. Supported sources are YouTube URLs and MP4 files.")
cap = cv2.VideoCapture(source)
# Get video properties
frame_width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
frame_height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
fps = cap.get(cv2.CAP_PROP_FPS)
# Initialize VideoWriter
if output_path is not None:
fourcc = cv2.VideoWriter_fourcc(*"mp4v")
out = cv2.VideoWriter(output_path, fourcc, fps, (frame_width, frame_height))
# Initialize track history
track_history = defaultdict(list)
frame_counter = 0
track_ids_to_infer = []
crops_to_infer = []
pred_labels = []
pred_confs = []
while cap.isOpened():
success, frame = cap.read()
if not success:
break
frame_counter += 1
# Run YOLO tracking
results = yolo_model.track(frame, persist=True, classes=[0]) # Track only person class
if results[0].boxes.id is not None:
boxes = results[0].boxes.xyxy.cpu().numpy()
track_ids = results[0].boxes.id.cpu().numpy()
# Visualize prediction
annotator = Annotator(frame, line_width=3, font_size=10, pil=False)
if frame_counter % skip_frame == 0:
crops_to_infer = []
track_ids_to_infer = []
for box, track_id in zip(boxes, track_ids):
if frame_counter % skip_frame == 0:
crop = crop_and_pad(frame, box, crop_margin_percentage)
track_history[track_id].append(crop)
if len(track_history[track_id]) > num_video_sequence_samples:
track_history[track_id].pop(0)
if len(track_history[track_id]) == num_video_sequence_samples and frame_counter % skip_frame == 0:
start_time = time.time()
crops = video_classifier.preprocess_crops_for_video_cls(track_history[track_id])
end_time = time.time()
preprocess_time = end_time - start_time
print(f"video cls preprocess time: {preprocess_time:.4f} seconds")
crops_to_infer.append(crops)
track_ids_to_infer.append(track_id)
if crops_to_infer and (
not pred_labels
or frame_counter % int(num_video_sequence_samples * skip_frame * (1 - video_cls_overlap_ratio)) == 0
):
crops_batch = torch.cat(crops_to_infer, dim=0)
start_inference_time = time.time()
output_batch = video_classifier(crops_batch)
end_inference_time = time.time()
inference_time = end_inference_time - start_inference_time
print(f"video cls inference time: {inference_time:.4f} seconds")
pred_labels, pred_confs = video_classifier.postprocess(output_batch)
if track_ids_to_infer and crops_to_infer:
for box, track_id, pred_label, pred_conf in zip(boxes, track_ids_to_infer, pred_labels, pred_confs):
top2_preds = sorted(zip(pred_label, pred_conf), key=lambda x: x[1], reverse=True)
label_text = " | ".join([f"{label} ({conf:.2f})" for label, conf in top2_preds])
annotator.box_label(box, label_text, color=(0, 0, 255))
# Write the annotated frame to the output video
if output_path is not None:
out.write(frame)
# Display the annotated frame
cv2.imshow("YOLOv8 Tracking with S3D Classification", frame)
if cv2.waitKey(1) & 0xFF == ord("q"):
break
cap.release()
if output_path is not None:
out.release()
cv2.destroyAllWindows()
def parse_opt():
"""Parse command line arguments."""
parser = argparse.ArgumentParser()
parser.add_argument("--weights", type=str, default="yolov8n.pt", help="ultralytics detector model path")
parser.add_argument("--device", default="", help='cuda device, i.e. 0 or 0,1,2,3 or cpu/mps, "" for auto-detection')
parser.add_argument(
"--source",
type=str,
default="https://www.youtube.com/watch?v=dQw4w9WgXcQ",
help="video file path or youtube URL",
)
parser.add_argument("--output-path", type=str, default="output_video.mp4", help="output video file path")
parser.add_argument(
"--crop-margin-percentage", type=int, default=10, help="percentage of margin to add around detected objects"
)
parser.add_argument(
"--num-video-sequence-samples", type=int, default=8, help="number of video frames to use for classification"
)
parser.add_argument("--skip-frame", type=int, default=2, help="number of frames to skip between detections")
parser.add_argument(
"--video-cls-overlap-ratio", type=float, default=0.25, help="overlap ratio between video sequences"
)
parser.add_argument("--fp16", action="store_true", help="use FP16 for inference")
parser.add_argument(
"--video-classifier-model", type=str, default="microsoft/xclip-base-patch32", help="video classifier model name"
)
parser.add_argument(
"--labels",
nargs="+",
type=str,
default=["dancing", "singing a song"],
help="labels for zero-shot video classification",
)
return parser.parse_args()
def main(opt):
"""Main function."""
run(**vars(opt))
if __name__ == "__main__":
opt = parse_opt()
main(opt)

@ -0,0 +1,116 @@
# Zero-shot Action Recognition with YOLOv8 (Inference on Video)
- Action recognition is a technique used to identify and classify actions performed by individuals in a video. This process enables more advanced analyses when multiple actions are considered. The actions can be detected and classified in real time.
- The system can be customized to recognize specific actions based on the user's preferences and requirements.
## Table of Contents
- [Step 1: Install the Required Libraries](#step-1-install-the-required-libraries)
- [Step 2: Run the Action Recognition Using Ultralytics YOLOv8](#step-2-run-the-action-recognition-using-ultralytics-yolov8)
- [Usage Options](#usage-options)
- [FAQ](#faq)
## Step 1: Install the Required Libraries
Clone the repository, install dependencies and `cd` to this local directory for commands in Step 2.
```bash
# Clone ultralytics repo
git clone https://github.com/ultralytics/ultralytics
# cd to local directory
cd examples/YOLOv8-Action-Recognition
# Install dependencies
pip install -U -r requirements.txt
```
## Step 2: Run the Action Recognition Using Ultralytics YOLOv8
Here are the basic commands for running the inference:
### Note
The action recognition model will automatically detect and track people in the video, and classify their actions based on the specified labels. The results will be displayed in real-time on the video output. You can customize the action labels by modifying the `--labels` argument when running the script.
```bash
# Quick start
python action_recognition.py
# Basic usage
python action_recognition.py --source "https://www.youtube.com/watch?v=dQw4w9WgXcQ" --labels "dancing" "singing a song"
# Use local video file
python action_recognition.py --source path/to/video.mp4
# Better detector performance
python action_recognition.py --weights yolov8m.pt
# Run on CPU
python action_recognition.py --device cpu
# Use a different video classifier model
python action_recognition.py --video-classifier-model "s3d"
# Use FP16 for inference (only for HuggingFace models)
python action_recognition.py --fp16
# Export output as mp4
python action_recognition.py --output-path output.mp4
# Combine multiple options
python action_recognition.py --source "https://www.youtube.com/watch?v=dQw4w9WgXcQ" --device 0 --video-classifier-model "microsoft/xclip-base-patch32" --labels "dancing" "singing a song" --fp16
```
## Usage Options
- `--weights`: Path to the YOLO model weights (default: "yolov8n.pt")
- `--device`: Cuda device, i.e. 0 or 0,1,2,3 or cpu (default: auto-detect)
- `--source`: Video file path or YouTube URL (default: "[rickroll](https://www.youtube.com/watch?v=dQw4w9WgXcQ)")
- `--output-path`: Output video file path
- `--crop-margin-percentage`: Percentage of margin to add around detected objects (default: 10)
- `--num-video-sequence-samples`: Number of video frames to use for classification (default: 8)
- `--skip-frame`: Number of frames to skip between detections (default: 1)
- `--video-cls-overlap-ratio`: Overlap ratio between video sequences (default: 0.25)
- `--fp16`: Use FP16 for inference (only for HuggingFace models)
- `--video-classifier-model`: Video classifier model name or path (default: "microsoft/xclip-base-patch32")
- `--labels`: Labels for zero-shot video classification (default: \["dancing" "singing a song"\])
## FAQ
**1. What Does Action Recognition Involve?**
Action recognition is a computational method used to identify and classify actions or activities performed by individuals in recorded video or real-time streams. This technique is widely used in video analysis, surveillance, and human-computer interaction, enabling the detection and understanding of human behaviors based on their motion patterns and context.
**2. Is Custom Action Labels Supported by the Action Recognition?**
Yes, custom action labels are supported by the action recognition system. The `action_recognition.py` script allows users to specify their own custom labels for zero-shot video classification. This can be done using the `--labels` argument when running the script. For example:
```bash
python action_recognition.py --source https://www.youtube.com/watch?v=dQw4w9WgXcQ --labels "dancing" "singing" "jumping"
```
You can adjust these labels to match the specific actions you want to recognize in your video. The system will then attempt to classify the detected actions based on these custom labels.
Additionally, you can choose between different video classification models:
1. For Hugging Face models, you can use any compatible video classification model. The default is set to:
- "microsoft/xclip-base-patch32"
2. For TorchVision models (no support for zero-shot labels), you can select from the following options:
- "s3d"
- "r3d_18"
- "swin3d_t"
- "swin3d_b"
- "mvit_v1_b"
- "mvit_v2_s"
**3. Why Combine Action Recognition with YOLOv8?**
YOLOv8 specializes in the detection and tracking of objects in video streams. Action recognition complements this by enabling the identification and classification of actions performed by individuals, making it a valuable application of YOLOv8.
**4. Can I Employ Other YOLO Versions?**
Certainly, you have the flexibility to specify different YOLO model weights using the `--weights` option.

@ -0,0 +1,4 @@
# Ultralytics YOLO 🚀, AGPL-3.0 license
ultralytics
transformers

@ -1,6 +1,6 @@
# Ultralytics YOLO 🚀, AGPL-3.0 license
__version__ = "8.2.46"
__version__ = "8.2.47"
import os

Loading…
Cancel
Save