视频理解：动作识别与时序建模

import torch
import torch.nn as nn
from typing import Tuple

class VideoInputSpec:
    """视频输入规格定义"""
    def __init__(self, num_frames: int = 32, height: int = 224, width: int = 224):
        self.num_frames = num_frames
        self.height = height
        self.width = width

    def tensor_shape(self, batch_size: int = 8) -> Tuple[int, ...]:
        # (B, C, T, H, W) - 标准的 5D 视频张量
        return (batch_size, 3, self.num_frames, self.height, self.width)

    def memory_mb(self, batch_size: int = 8) -> float:
        shape = self.tensor_shape(batch_size)
        total_elements = 1
        for dim in shape:
            total_elements *= dim
        return total_elements * 4 / (1024 * 1024)  # float32

spec = VideoInputSpec(num_frames=64, height=224, width=224)
print(f"Shape: {spec.tensor_shape()}")       # (8, 3, 64, 224, 224)
print(f"Memory: {spec.memory_mb():.1f} MB")  # ~144.0 MB

import torch
from torch.utils.data import Dataset, DataLoader
from typing import List, Tuple
import os

class VideoActionDataset(Dataset):
    """视频动作识别数据集基类"""
    def __init__(self, video_dir: str, label_file: str, num_frames: int = 32):
        self.video_dir = video_dir
        self.num_frames = num_frames
        # 加载 label 映射
        self.samples: List[Tuple[str, int]] = []
        with open(label_file, 'r') as f:
            for line in f:
                video_name, label = line.strip().split()
                self.samples.append((video_name, int(label)))
        self.classes = sorted(set(l for _, l in self.samples))

    def __len__(self) -> int:
        return len(self.samples)

    def __getitem__(self, idx: int):
        video_name, label = self.samples[idx]
        # 实际项目中这里使用 decord 或 PyAV 读取视频帧
        frames = self._load_frames(video_name)
        tensor = torch.stack(frames)  # (T, C, H, W)
        tensor = tensor.permute(1, 0, 2, 3)  # (C, T, H, W)
        return tensor, label

    def _load_frames(self, video_name: str) -> List[torch.Tensor]:
        raise NotImplementedError

import torch
import torch.nn as nn
import torch.nn.functional as F

class C3D(nn.Module):
    """C3D 网络简化实现"""
    def __init__(self, num_classes: int = 400):
        super().__init__()
        self.features = nn.Sequential(
            nn.Conv3d(3, 64, kernel_size=3, padding=1),
            nn.ReLU(), nn.MaxPool3d((1, 2, 2)),
            nn.Conv3d(64, 128, kernel_size=3, padding=1),
            nn.ReLU(), nn.MaxPool3d((2, 2, 2)),
            nn.Conv3d(128, 256, kernel_size=3, padding=1),
            nn.ReLU(), nn.Conv3d(256, 256, kernel_size=3, padding=1),
            nn.ReLU(), nn.MaxPool3d((2, 2, 2)),
            nn.Conv3d(256, 512, kernel_size=3, padding=1),
            nn.ReLU(), nn.Conv3d(512, 512, kernel_size=3, padding=1),
            nn.ReLU(), nn.MaxPool3d((2, 2, 2)),
        )
        self.classifier = nn.Sequential(
            nn.Linear(512 * 4 * 7 * 7, 4096),
            nn.ReLU(), nn.Dropout(0.5),
            nn.Linear(4096, 4096),
            nn.ReLU(), nn.Dropout(0.5),
            nn.Linear(4096, num_classes),
        )

    def forward(self, x):
        # x: (B, C, T, H, W)
        x = self.features(x)
        x = x.view(x.size(0), -1)
        return self.classifier(x)

import torch
import torch.nn as nn

def inflate_conv2d_to_3d(conv2d: nn.Conv2d, time_dim: int = 3) -> nn.Conv3d:
    """将 2D 卷积核膨胀为 3D 卷积核（I3D 核心操作）"""
    # 2D weight shape: (out_ch, in_ch, kH, kW)
    # 3D weight shape: (out_ch, in_ch, kT, kH, kW)
    weight_2d = conv2d.weight.data
    weight_3d = weight_2d.unsqueeze(2).repeat(1, 1, time_dim, 1, 1)
    # 时间维度归一化，保持输出尺度一致
    weight_3d = weight_3d / time_dim

    conv3d = nn.Conv3d(
        in_channels=conv2d.in_channels,
        out_channels=conv2d.out_channels,
        kernel_size=(time_dim, *conv2d.kernel_size),
        stride=(1, *conv2d.stride),
        padding=(time_dim // 2, *conv2d.padding),
        bias=conv2d.bias is not None,
    )
    conv3d.weight.data = weight_3d
    if conv2d.bias is not None:
        conv3d.bias.data = conv2d.bias.data
    return conv3d

# 验证膨胀操作
conv2d = nn.Conv2d(64, 128, kernel_size=3, padding=1)
conv3d = inflate_conv2d_to_3d(conv2d, time_dim=3)
print(f"3D kernel shape: {conv3d.weight.shape}")  # (128, 64, 3, 3, 3)

import torch
import torch.nn as nn
import torchvision.models as models
from typing import Tuple

class TwoStreamNetwork(nn.Module):
    """双分支 Two-Stream 动作识别网络"""
    def __init__(self, num_classes: int = 400, spatial_weight: float = 1.0,
                 temporal_weight: float = 2.0):
        super().__init__()
        self.spatial_weight = spatial_weight
        self.temporal_weight = temporal_weight

        # 空间分支：RGB 单帧输入
        spatial = models.resnet50(weights=models.ResNet50_Weights.DEFAULT)
        spatial.conv1 = nn.Conv2d(3, 64, 7, 2, 3, bias=False)
        spatial.fc = nn.Linear(spatial.fc.in_features, num_classes)
        self.spatial_stream = spatial

        # 时间分支：光流堆叠输入 (10 帧光流 = 20 通道)
        temporal = models.resnet50(weights=None)
        temporal.conv1 = nn.Conv2d(20, 64, 7, 2, 3, bias=False)
        temporal.fc = nn.Linear(temporal.fc.in_features, num_classes)
        self.temporal_stream = temporal

    def forward(self, rgb: torch.Tensor, flow: torch.Tensor) -> torch.Tensor:
        spatial_logits = self.spatial_stream(rgb)
        temporal_logits = self.temporal_stream(flow)
        # 加权融合
        fused = (self.spatial_weight * spatial_logits +
                 self.temporal_weight * temporal_logits)
        return fused

import cv2
import numpy as np
from typing import List

def compute_dense_optical_flow(frames: List[np.ndarray]) -> np.ndarray:
    """使用 Farneback 方法计算密集光流"""
    if len(frames) < 2:
        raise ValueError("至少需要 2 帧")

    flows = []
    for i in range(len(frames) - 1):
        prev = cv2.cvtColor(frames[i], cv2.COLOR_RGB2GRAY)
        next_ = cv2.cvtColor(frames[i + 1], cv2.COLOR_RGB2GRAY)

        flow = cv2.calcOpticalFlowFarneback(
            prev, next_, None,
            pyr_scale=0.5, levels=3, winsize=15,
            iterations=3, poly_n=5, poly_sigma=1.2,
            flags=0,
        )
        flows.append(flow)  # shape: (H, W, 2)

    # 堆叠为 (T-1, H, W, 2) 并转置为 (2*(T-1), H, W)
    flows = np.stack(flows, axis=0)
    stacked = np.concatenate([flows[:, :, :, 0], flows[:, :, :, 1]], axis=0)
    return stacked.astype(np.float32)

# 使用示例
frames = [np.random.randint(0, 255, (224, 224, 3), dtype=np.uint8) for _ in range(11)]
flow_stacked = compute_dense_optical_flow(frames)
print(f"Flow shape: {flow_stacked.shape}")  # (20, 224, 224)

import torch
import torch.nn as nn
from typing import Optional

class TemporalConvBlock(nn.Module):
    """时间卷积模块（TCN 核心组件）"""
    def __init__(self, in_channels: int, out_channels: int,
                 kernel_size: int = 3, dilation: int = 1):
        super().__init__()
        padding = (kernel_size - 1) * dilation // 2
        self.conv = nn.Conv1d(in_channels, out_channels, kernel_size,
                              padding=padding, dilation=dilation)
        self.bn = nn.BatchNorm1d(out_channels)
        self.relu = nn.ReLU()
        self.dropout = nn.Dropout(0.2)

    def forward(self, x: torch.Tensor) -> torch.Tensor:
        # x: (B, C, T) - 每帧特征作为序列
        x = self.conv(x)
        x = self.bn(x)
        x = self.relu(x)
        return self.dropout(x)

class SimpleTCN(nn.Module):
    """简化版 TCN 用于视频时序建模"""
    def __init__(self, feature_dim: int = 2048, num_classes: int = 400):
        super().__init__()
        self.blocks = nn.Sequential(
            TemporalConvBlock(feature_dim, 512, dilation=1),
            TemporalConvBlock(512, 256, dilation=2),
            TemporalConvBlock(256, 128, dilation=4),
        )
        self.classifier = nn.Linear(128, num_classes)

    def forward(self, frame_features: torch.Tensor) -> torch.Tensor:
        # frame_features: (B, T, C) -> 转置为 (B, C, T)
        x = frame_features.permute(0, 2, 1)
        x = self.blocks(x)
        # 全局时间平均池化
        x = x.mean(dim=-1)
        return self.classifier(x)

import torch
import torch.nn as nn
from einops import rearrange

class TimeSformerAttention(nn.Module):
    """TimeSformer 的空间-时间分离注意力"""
    def __init__(self, dim: int, num_heads: int = 8, num_frames: int = 8):
        super().__init__()
        self.num_frames = num_frames
        self.num_heads = num_heads
        self.scale = (dim // num_heads) ** -0.5

        self.qkv = nn.Linear(dim, dim * 3)
        self.proj = nn.Linear(dim, dim)

        # 空间注意力和时间注意力的投影
        self.spatial_attn = nn.MultiheadAttention(dim, num_heads, batch_first=True)
        self.temporal_attn = nn.MultiheadAttention(dim, num_heads, batch_first=True)
        self.norm1 = nn.LayerNorm(dim)
        self.norm2 = nn.LayerNorm(dim)

    def forward(self, x: torch.Tensor) -> torch.Tensor:
        # x: (B, T*P, D) - T帧, 每帧P个patch
        B, TP, D = x.shape
        T = self.num_frames
        P = TP // T

        # 第一步：空间自注意力（对每帧独立做）
        x_spatial = rearrange(x, 'b (t p) d -> (b t) p d', t=T, p=P)
        x_spatial, _ = self.spatial_attn(x_spatial, x_spatial, x_spatial)
        x_spatial = rearrange(x_spatial, '(b t) p d -> b (t p) d', t=T)
        x = x + x_spatial
        x = self.norm1(x)

        # 第二步：时间自注意力（对每个 patch 跨帧做）
        x_temporal = rearrange(x, 'b (t p) d -> (b p) t d', t=T, p=P)
        x_temporal, _ = self.temporal_attn(x_temporal, x_temporal, x_temporal)
        x_temporal = rearrange(x_temporal, '(b p) t d -> b (t p) d', t=T, p=P)
        x = x + x_temporal
        x = self.norm2(x)
        return x

import os
import json
from typing import Dict, List

class Kinetics400Builder:
    """Kinetics-400 数据集构建工具"""
    def __init__(self, root_dir: str, split: str = "train"):
        self.root_dir = root_dir
        self.split = split
        self.classes = self._load_classes()

    def _load_classes(self) -> List[str]:
        classes_file = os.path.join(self.root_dir, "kinetics_400_classes.txt")
        with open(classes_file, 'r') as f:
            return [line.strip() for line in f if line.strip()]

    def get_class_distribution(self, annotations_file: str) -> Dict[str, int]:
        """统计每个类别的样本数量"""
        distribution = {}
        with open(annotations_file, 'r') as f:
            for line in f:
                parts = line.strip().split()
                if len(parts) >= 2:
                    label = int(parts[-1])
                    cls_name = self.classes[label] if label < len(self.classes) else f"unknown_{label}"
                    distribution[cls_name] = distribution.get(cls_name, 0) + 1
        return distribution

    def get_statistics(self) -> Dict:
        return {
            "num_classes": len(self.classes),
            "split": self.split,
            "top_classes": ["playing guitar", "riding bike", "eating"],
            "avg_duration_sec": 10.0,
            "resolution_range": "varies (YouTube source)",
        }

builder = Kinetics400Builder("/data/kinetics400")
print(json.dumps(builder.get_statistics(), indent=2))

import torch
from torch.utils.data import Dataset, DataLoader
from typing import Tuple
import random

class UCF101Dataset(Dataset):
    """UCF101 数据集"""
    def __init__(self, root: str, split_file: str, num_frames: int = 32,
                 transform=None):
        self.root = root
        self.num_frames = num_frames
        self.transform = transform
        self.samples = self._load_split(split_file)
        self.classes = sorted(set(s[1] for s in self.samples))
        self.class_to_idx = {c: i for i, c in enumerate(self.classes)}

    def _load_split(self, split_file: str):
        samples = []
        with open(split_file, 'r') as f:
            for line in f:
                parts = line.strip().split()
                if len(parts) == 2:
                    video_path, label = parts
                    samples.append((video_path, int(label) - 1))
        return samples

    def __len__(self) -> int:
        return len(self.samples)

    def __getitem__(self, idx: int) -> Tuple[torch.Tensor, int]:
        video_path, label = self.samples[idx]
        frames = self._load_video_frames(video_path)
        tensor = torch.stack(frames)  # (T, C, H, W)
        if self.transform:
            tensor = self.transform(tensor)
        return tensor.permute(1, 0, 2, 3), label

    def _load_video_frames(self, path: str):
        raise NotImplementedError  # 使用 decord 或 PyAV 实现

import torch
import torch.nn as nn
from typing import List

class SlowFastBlock(nn.Module):
    """SlowFast 单阶段模块"""
    def __init__(self, slow_in_channels: int, fast_in_channels: int,
                 slow_out_channels: int, fast_out_channels: int,
                 alpha: int = 8, beta: int = 8):
        """
        alpha: 时间采样率比 (fast_frames / slow_frames)
        beta:  通道数比 (slow_channels / fast_channels)
        """
        super().__init__()
        self.alpha = alpha
        self.beta = beta

        # Slow 分支
        self.slow_path = nn.Sequential(
            nn.Conv3d(slow_in_channels, slow_out_channels,
                      kernel_size=(3, 3, 3), padding=(1, 1, 1)),
            nn.BatchNorm3d(slow_out_channels),
            nn.ReLU(),
        )

        # Fast 分支
        self.fast_path = nn.Sequential(
            nn.Conv3d(fast_in_channels, fast_out_channels,
                      kernel_size=(3, 3, 3), padding=(1, 1, 1)),
            nn.BatchNorm3d(fast_out_channels),
            nn.ReLU(),
        )

        # 横向连接：fast -> slow
        self.lateral = nn.Sequential(
            nn.Conv3d(fast_out_channels, slow_out_channels // beta,
                      kernel_size=(5, 1, 1), stride=(alpha, 1, 1),
                      padding=(2, 0, 0)),
            nn.BatchNorm3d(slow_out_channels // beta),
            nn.ReLU(),
        )

    def forward(self, slow_input: torch.Tensor,
                fast_input: torch.Tensor) -> List[torch.Tensor]:
        slow_out = self.slow_path(slow_input)
        fast_out = self.fast_path(fast_input)

        # 横向连接融合
        lateral_feat = self.lateral(fast_out)
        slow_out = torch.cat([slow_out, lateral_feat], dim=1)

        return [slow_out, fast_out]

import torch
import torch.nn as nn

class SlowFastNetwork(nn.Module):
    """完整的 SlowFast 网络"""
    def __init__(self, num_classes: int = 400, alpha: int = 8, beta: int = 8):
        super().__init__()
        self.alpha = alpha  # 时间采样率比

        # 输入通道: slow=3(RGB), fast=3(RGB)
        # slow 每 8 帧取 1 帧，fast 每帧都取
        self.slow_conv1 = nn.Conv3d(3, 64, (1, 7, 7), (1, 2, 2), (0, 3, 3))
        self.fast_conv1 = nn.Conv3d(3, 8, (5, 7, 7), (8, 2, 2), (2, 3, 3))
        self.slow_bn1 = nn.BatchNorm3d(64)
        self.fast_bn1 = nn.BatchNorm3d(8)

        # 多个 SlowFast 阶段
        self.stage1 = SlowFastBlock(64, 8, 256, 32, alpha, beta)
        self.stage2 = SlowFastBlock(256 + 32, 32, 512, 64, alpha, beta)
        self.stage3 = SlowFastBlock(512 + 64, 64, 1024, 128, alpha, beta)
        self.stage4 = SlowFastBlock(1024 + 128, 128, 2048, 256, alpha, beta)

        # 全局池化 + 分类
        self.pool = nn.AdaptiveAvgPool3d(1)
        self.classifier = nn.Linear(2048 + 256, num_classes)

    def forward(self, slow_input: torch.Tensor,
                fast_input: torch.Tensor) -> torch.Tensor:
        # 初始卷积
        slow = torch.relu(self.slow_bn1(self.slow_conv1(slow_input)))
        fast = torch.relu(self.fast_bn1(self.fast_conv1(fast_input)))

        # 逐级通过 SlowFast 模块
        slow, fast = self.stage1(slow, fast)
        slow, fast = self.stage2(slow, fast)
        slow, fast = self.stage3(slow, fast)
        slow, fast = self.stage4(slow, fast)

        # 全局池化 + 拼接
        slow = self.pool(slow).flatten(1)
        fast = self.pool(fast).flatten(1)
        fused = torch.cat([slow, fast], dim=1)
        return self.classifier(fused)

import torch
import torch.nn.functional as F
import numpy as np
from typing import List, Tuple, Dict

class I3DPredictor:
    """I3D 视频动作识别推理管线"""
    def __init__(self, model_path: str, classes_file: str,
                 num_frames: int = 64, device: str = "cuda"):
        self.device = torch.device(device)
        self.num_frames = num_frames

        # 加载模型
        self.model = self._load_model(model_path)
        self.model.to(self.device)
        self.model.eval()

        # 加载类别名称
        with open(classes_file, 'r') as f:
            self.classes = [line.strip() for line in f if line.strip()]

        # ImageNet 归一化参数
        self.mean = torch.tensor([0.485, 0.456, 0.406]).view(1, 3, 1, 1, 1)
        self.std = torch.tensor([0.229, 0.224, 0.225]).view(1, 3, 1, 1, 1)

    def _load_model(self, path: str) -> torch.nn.Module:
        from i3d import InceptionI3d  # 假设已定义
        model = InceptionI3d(num_classes=400)
        model.load_state_dict(torch.load(path, map_location='cpu'))
        return model

    def preprocess(self, frames: np.ndarray) -> torch.Tensor:
        """预处理：resize → normalize → to tensor"""
        # frames: (T, H, W, C), uint8 [0, 255]
        tensor = torch.from_numpy(frames).float() / 255.0
        tensor = tensor.permute(3, 0, 1, 2).unsqueeze(0)  # (1, C, T, H, W)
        tensor = (tensor - self.mean) / self.std
        return tensor.to(self.device)

    def predict(self, frames: np.ndarray, top_k: int = 5) -> List[Dict]:
        """执行推理并返回 Top-K 预测"""
        tensor = self.preprocess(frames)
        with torch.no_grad():
            logits = self.model(tensor)
            probs = F.softmax(logits, dim=1)

        top_probs, top_indices = probs.topk(top_k, dim=1)
        results = []
        for prob, idx in zip(top_probs[0], top_indices[0]):
            results.append({
                "class": self.classes[idx.item()],
                "probability": prob.item(),
            })
        return results

import decord
from decord import VideoReader
import numpy as np
from typing import List

class VideoSampler:
    """视频帧采样策略"""
    def __init__(self, num_frames: int = 64):
        self.num_frames = num_frames

    def uniform_sample(self, video_path: str) -> np.ndarray:
        """均匀采样：从视频中均匀抽取 N 帧"""
        vr = VideoReader(video_path, ctx=decord.cpu(0))
        total_frames = len(vr)

        if total_frames <= self.num_frames:
            # 视频太短，重复最后一帧
            indices = list(range(total_frames))
            indices += [total_frames - 1] * (self.num_frames - total_frames)
        else:
            # 均匀间隔采样
            indices = np.linspace(0, total_frames - 1, self.num_frames)
            indices = np.round(indices).astype(int)

        frames = vr.get_batch(indices).asnumpy()  # (T, H, W, C)
        return frames

    def multi_crop_sample(self, video_path: str,
                          num_crops: int = 3) -> List[np.ndarray]:
        """多段采样：用于提升推理精度"""
        vr = VideoReader(video_path, ctx=decord.cpu(0))
        total_frames = len(vr)
        segment_length = total_frames // num_crops

        crops = []
        for i in range(num_crops):
            start = i * segment_length
            end = start + segment_length
            indices = np.linspace(start, end - 1, self.num_frames)
            indices = np.round(indices).astype(int)
            frames = vr.get_batch(indices).asnumpy()
            crops.append(frames)
        return crops

# 使用示例
sampler = VideoSampler(num_frames=64)
frames = sampler.uniform_sample("video.mp4")
print(f"Sampled shape: {frames.shape}")  # (64, H, W, 3)