Video Analysis and Processing

1import cv2
2import numpy as np
3from typing import Generator, Tuple, List
4from collections import deque
5
6class VideoProcessor:
7    """
8    Fundamental video processing utilities.
9    """
10
11    def __init__(self, video_path: str = None, camera_id: int = 0):
12        """
13        Initialize video processor.
14
15        Args:
16            video_path: Path to video file (None for camera)
17            camera_id: Camera device ID if video_path is None
18        """
19        if video_path:
20            self.cap = cv2.VideoCapture(video_path)
21        else:
22            self.cap = cv2.VideoCapture(camera_id)
23
24        self.fps = self.cap.get(cv2.CAP_PROP_FPS)
25        self.width = int(self.cap.get(cv2.CAP_PROP_FRAME_WIDTH))
26        self.height = int(self.cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
27        self.frame_count = int(self.cap.get(cv2.CAP_PROP_FRAME_COUNT))
28
29    def read_frames(self, skip: int = 0) -> Generator[Tuple[int, np.ndarray], None, None]:
30        """
31        Generator to read frames from video.
32
33        Args:
34            skip: Number of frames to skip between yields
35
36        Yields:
37            (frame_number, frame) tuples
38        """
39        frame_num = 0
40        while True:
41            ret, frame = self.cap.read()
42            if not ret:
43                break
44
45            if frame_num % (skip + 1) == 0:
46                yield frame_num, frame
47
48            frame_num += 1
49
50    def get_frame(self, frame_number: int) -> np.ndarray:
51        """Get specific frame by number."""
52        self.cap.set(cv2.CAP_PROP_POS_FRAMES, frame_number)
53        ret, frame = self.cap.read()
54        return frame if ret else None
55
56    def release(self):
57        """Release video capture."""
58        self.cap.release()
59
60
61class TemporalBuffer:
62    """
63    Buffer for temporal analysis across frames.
64    Useful for motion analysis, temporal smoothing, etc.
65    """
66
67    def __init__(self, buffer_size: int = 30):
68        self.buffer_size = buffer_size
69        self.frames = deque(maxlen=buffer_size)
70        self.timestamps = deque(maxlen=buffer_size)
71
72    def add(self, frame: np.ndarray, timestamp: float = None):
73        """Add frame to buffer."""
74        self.frames.append(frame)
75        self.timestamps.append(timestamp or len(self.frames))
76
77    def get_temporal_difference(self, offset: int = 1) -> np.ndarray:
78        """
79        Compute frame difference for motion detection.
80
81        Args:
82            offset: Number of frames back to compare
83
84        Returns:
85            Absolute difference between current and past frame
86        """
87        if len(self.frames) <= offset:
88            return None
89
90        current = self.frames[-1]
91        previous = self.frames[-(offset + 1)]
92
93        # Convert to grayscale if color
94        if len(current.shape) == 3:
95            current = cv2.cvtColor(current, cv2.COLOR_BGR2GRAY)
96            previous = cv2.cvtColor(previous, cv2.COLOR_BGR2GRAY)
97
98        return cv2.absdiff(current, previous)
99
100    def get_running_average(self) -> np.ndarray:
101        """Compute running average of recent frames."""
102        if not self.frames:
103            return None
104
105        frames_array = np.array(list(self.frames), dtype=np.float32)
106        return np.mean(frames_array, axis=0).astype(np.uint8)
107
108
109class BackgroundSubtractor:
110    """
111    Background subtraction for moving object detection.
112    """
113
114    def __init__(self, method: str = 'MOG2'):
115        if method == 'MOG2':
116            self.bg_subtractor = cv2.createBackgroundSubtractorMOG2(
117                history=500,
118                varThreshold=16,
119                detectShadows=True
120            )
121        elif method == 'KNN':
122            self.bg_subtractor = cv2.createBackgroundSubtractorKNN(
123                history=500,
124                dist2Threshold=400.0,
125                detectShadows=True
126            )
127        else:
128            raise ValueError(f"Unknown method: {method}")
129
130    def apply(self, frame: np.ndarray,
131             learning_rate: float = -1) -> np.ndarray:
132        """
133        Apply background subtraction.
134
135        Args:
136            frame: Input frame
137            learning_rate: Learning rate (-1 for auto)
138
139        Returns:
140            Foreground mask
141        """
142        fg_mask = self.bg_subtractor.apply(frame, learningRate=learning_rate)
143
144        # Clean up mask with morphological operations
145        kernel = cv2.getStructuringElement(cv2.MORPH_ELLIPSE, (5, 5))
146        fg_mask = cv2.morphologyEx(fg_mask, cv2.MORPH_OPEN, kernel)
147        fg_mask = cv2.morphologyEx(fg_mask, cv2.MORPH_CLOSE, kernel)
148
149        return fg_mask
150
151    def get_foreground_contours(self, fg_mask: np.ndarray,
152                               min_area: int = 500) -> List[np.ndarray]:
153        """Extract contours from foreground mask."""
154        contours, _ = cv2.findContours(
155            fg_mask, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE
156        )
157
158        # Filter by area
159        return [c for c in contours if cv2.contourArea(c) > min_area]
160
161
162# Example usage
163print("Video Processing Utilities loaded")
164print("Classes: VideoProcessor, TemporalBuffer, BackgroundSubtractor")

1import cv2
2import numpy as np
3from typing import Tuple, Optional
4
5class OpticalFlowAnalyzer:
6    """
7    Optical flow computation and visualization.
8    """
9
10    def __init__(self, method: str = 'farneback'):
11        self.method = method
12        self.prev_gray = None
13
14        # Lucas-Kanade parameters
15        self.lk_params = dict(
16            winSize=(15, 15),
17            maxLevel=2,
18            criteria=(cv2.TERM_CRITERIA_EPS | cv2.TERM_CRITERIA_COUNT, 10, 0.03)
19        )
20
21        # Farneback parameters
22        self.farneback_params = dict(
23            pyr_scale=0.5,
24            levels=3,
25            winsize=15,
26            iterations=3,
27            poly_n=5,
28            poly_sigma=1.2,
29            flags=0
30        )
31
32    def compute_dense_flow(self, frame: np.ndarray) -> Optional[np.ndarray]:
33        """
34        Compute dense optical flow using Farneback method.
35
36        Returns:
37            Flow field of shape (H, W, 2) with (u, v) velocities
38        """
39        gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
40
41        if self.prev_gray is None:
42            self.prev_gray = gray
43            return None
44
45        flow = cv2.calcOpticalFlowFarneback(
46            self.prev_gray, gray, None, **self.farneback_params
47        )
48
49        self.prev_gray = gray
50        return flow
51
52    def compute_sparse_flow(self, frame: np.ndarray,
53                           points: np.ndarray) -> Tuple[np.ndarray, np.ndarray]:
54        """
55        Compute sparse optical flow using Lucas-Kanade.
56
57        Args:
58            frame: Current frame
59            points: Points to track (N, 1, 2)
60
61        Returns:
62            new_points: Updated point positions
63            status: Track status for each point
64        """
65        gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
66
67        if self.prev_gray is None:
68            self.prev_gray = gray
69            return points, np.ones(len(points), dtype=np.uint8)
70
71        new_points, status, _ = cv2.calcOpticalFlowPyrLK(
72            self.prev_gray, gray, points, None, **self.lk_params
73        )
74
75        self.prev_gray = gray
76        return new_points, status.flatten()
77
78    def visualize_flow(self, flow: np.ndarray,
79                      scale: float = 1.0) -> np.ndarray:
80        """
81        Visualize dense optical flow as color image.
82
83        Uses HSV encoding:
84        - Hue: Direction of motion
85        - Saturation: Always 255
86        - Value: Magnitude of motion
87        """
88        h, w = flow.shape[:2]
89
90        # Compute magnitude and angle
91        magnitude, angle = cv2.cartToPolar(flow[..., 0], flow[..., 1])
92
93        # Create HSV image
94        hsv = np.zeros((h, w, 3), dtype=np.uint8)
95        hsv[..., 0] = angle * 180 / np.pi / 2  # Hue
96        hsv[..., 1] = 255  # Saturation
97        hsv[..., 2] = cv2.normalize(magnitude * scale, None, 0, 255,
98                                    cv2.NORM_MINMAX)  # Value
99
100        return cv2.cvtColor(hsv, cv2.COLOR_HSV2BGR)
101
102    def draw_flow_arrows(self, frame: np.ndarray, flow: np.ndarray,
103                        step: int = 16, scale: float = 1.0) -> np.ndarray:
104        """Draw flow vectors as arrows on frame."""
105        h, w = flow.shape[:2]
106        output = frame.copy()
107
108        # Create grid of points
109        y, x = np.mgrid[step//2:h:step, step//2:w:step].reshape(2, -1).astype(int)
110
111        # Get flow at grid points
112        fx, fy = flow[y, x].T
113
114        # Draw arrows
115        lines = np.vstack([x, y, x + fx * scale, y + fy * scale]).T.reshape(-1, 2, 2)
116        lines = np.int32(lines + 0.5)
117
118        for (x1, y1), (x2, y2) in lines:
119            cv2.arrowedLine(output, (x1, y1), (x2, y2), (0, 255, 0), 1,
120                           tipLength=0.3)
121
122        return output
123
124
125class MotionAnalysis:
126    """
127    Higher-level motion analysis from optical flow.
128    """
129
130    @staticmethod
131    def detect_motion_regions(flow: np.ndarray,
132                             threshold: float = 2.0) -> np.ndarray:
133        """
134        Detect regions with significant motion.
135
136        Returns:
137            Binary mask of motion regions
138        """
139        magnitude = np.sqrt(flow[..., 0]**2 + flow[..., 1]**2)
140        motion_mask = (magnitude > threshold).astype(np.uint8) * 255
141
142        # Clean up
143        kernel = cv2.getStructuringElement(cv2.MORPH_ELLIPSE, (5, 5))
144        motion_mask = cv2.morphologyEx(motion_mask, cv2.MORPH_OPEN, kernel)
145        motion_mask = cv2.morphologyEx(motion_mask, cv2.MORPH_CLOSE, kernel)
146
147        return motion_mask
148
149    @staticmethod
150    def estimate_camera_motion(flow: np.ndarray) -> Tuple[float, float]:
151        """
152        Estimate global camera motion from flow field.
153
154        Returns:
155            (dx, dy) estimated camera translation
156        """
157        # Use median to be robust to moving objects
158        dx = np.median(flow[..., 0])
159        dy = np.median(flow[..., 1])
160        return dx, dy
161
162    @staticmethod
163    def segment_motion(flow: np.ndarray, n_clusters: int = 3) -> np.ndarray:
164        """
165        Segment scene by motion patterns using K-means.
166        """
167        h, w = flow.shape[:2]
168
169        # Prepare features: flow + position
170        y, x = np.mgrid[0:h, 0:w].astype(np.float32)
171        features = np.dstack([
172            flow[..., 0],
173            flow[..., 1],
174            x / w * 2,  # Normalized position
175            y / h * 2
176        ]).reshape(-1, 4)
177
178        # K-means clustering
179        criteria = (cv2.TERM_CRITERIA_EPS + cv2.TERM_CRITERIA_MAX_ITER, 10, 1.0)
180        _, labels, _ = cv2.kmeans(
181            features, n_clusters, None, criteria, 10, cv2.KMEANS_RANDOM_CENTERS
182        )
183
184        return labels.reshape(h, w)
185
186
187# Example
188flow_analyzer = OpticalFlowAnalyzer(method='farneback')
189print("Optical Flow Analyzer initialized")

1import numpy as np
2from typing import List, Tuple, Dict
3from scipy.optimize import linear_sum_assignment
4from filterpy.kalman import KalmanFilter
5
6class KalmanTracker:
7    """
8    Kalman filter based tracker for a single object.
9    State: [x, y, s, r, vx, vy, vs]
10    - (x, y): center position
11    - s: area
12    - r: aspect ratio
13    - (vx, vy, vs): velocities
14    """
15
16    count = 0
17
18    def __init__(self, bbox: np.ndarray):
19        """
20        Initialize tracker with bounding box [x1, y1, x2, y2].
21        """
22        self.kf = KalmanFilter(dim_x=7, dim_z=4)
23
24        # State transition matrix
25        self.kf.F = np.array([
26            [1, 0, 0, 0, 1, 0, 0],
27            [0, 1, 0, 0, 0, 1, 0],
28            [0, 0, 1, 0, 0, 0, 1],
29            [0, 0, 0, 1, 0, 0, 0],
30            [0, 0, 0, 0, 1, 0, 0],
31            [0, 0, 0, 0, 0, 1, 0],
32            [0, 0, 0, 0, 0, 0, 1]
33        ])
34
35        # Measurement matrix
36        self.kf.H = np.array([
37            [1, 0, 0, 0, 0, 0, 0],
38            [0, 1, 0, 0, 0, 0, 0],
39            [0, 0, 1, 0, 0, 0, 0],
40            [0, 0, 0, 1, 0, 0, 0]
41        ])
42
43        # Measurement noise
44        self.kf.R[2:, 2:] *= 10.
45        # Covariance matrix
46        self.kf.P[4:, 4:] *= 1000.
47        self.kf.P *= 10.
48        # Process noise
49        self.kf.Q[-1, -1] *= 0.01
50        self.kf.Q[4:, 4:] *= 0.01
51
52        # Initialize state with bbox
53        self.kf.x[:4] = self._bbox_to_z(bbox)
54
55        self.time_since_update = 0
56        self.id = KalmanTracker.count
57        KalmanTracker.count += 1
58        self.history = []
59        self.hits = 0
60        self.hit_streak = 0
61        self.age = 0
62
63    def _bbox_to_z(self, bbox: np.ndarray) -> np.ndarray:
64        """Convert [x1,y1,x2,y2] to [cx, cy, s, r]."""
65        w = bbox[2] - bbox[0]
66        h = bbox[3] - bbox[1]
67        cx = bbox[0] + w / 2
68        cy = bbox[1] + h / 2
69        s = w * h
70        r = w / (h + 1e-6)
71        return np.array([[cx], [cy], [s], [r]])
72
73    def _z_to_bbox(self, z: np.ndarray) -> np.ndarray:
74        """Convert [cx, cy, s, r] to [x1, y1, x2, y2]."""
75        w = np.sqrt(z[2] * z[3])
76        h = z[2] / (w + 1e-6)
77        return np.array([
78            z[0] - w/2,
79            z[1] - h/2,
80            z[0] + w/2,
81            z[1] + h/2
82        ]).flatten()
83
84    def update(self, bbox: np.ndarray):
85        """Update state with observed bbox."""
86        self.time_since_update = 0
87        self.history = []
88        self.hits += 1
89        self.hit_streak += 1
90        self.kf.update(self._bbox_to_z(bbox))
91
92    def predict(self) -> np.ndarray:
93        """Predict next state and return bbox."""
94        if self.kf.x[6] + self.kf.x[2] <= 0:
95            self.kf.x[6] *= 0.0
96
97        self.kf.predict()
98        self.age += 1
99
100        if self.time_since_update > 0:
101            self.hit_streak = 0
102
103        self.time_since_update += 1
104        self.history.append(self._z_to_bbox(self.kf.x[:4]))
105        return self.history[-1]
106
107    def get_state(self) -> np.ndarray:
108        """Get current state as bbox."""
109        return self._z_to_bbox(self.kf.x[:4])
110
111
112class SORT:
113    """
114    Simple Online and Realtime Tracking (SORT).
115
116    Assigns detections to tracks using IoU matching
117    and Kalman filtering for state estimation.
118    """
119
120    def __init__(self, max_age: int = 3, min_hits: int = 3,
121                 iou_threshold: float = 0.3):
122        self.max_age = max_age
123        self.min_hits = min_hits
124        self.iou_threshold = iou_threshold
125        self.trackers: List[KalmanTracker] = []
126        self.frame_count = 0
127
128    @staticmethod
129    def iou_batch(bb_test: np.ndarray, bb_gt: np.ndarray) -> np.ndarray:
130        """
131        Compute IoU between all pairs of boxes.
132
133        Args:
134            bb_test: (N, 4) array of test boxes
135            bb_gt: (M, 4) array of GT boxes
136
137        Returns:
138            (N, M) IoU matrix
139        """
140        bb_gt = np.expand_dims(bb_gt, 0)
141        bb_test = np.expand_dims(bb_test, 1)
142
143        xx1 = np.maximum(bb_test[..., 0], bb_gt[..., 0])
144        yy1 = np.maximum(bb_test[..., 1], bb_gt[..., 1])
145        xx2 = np.minimum(bb_test[..., 2], bb_gt[..., 2])
146        yy2 = np.minimum(bb_test[..., 3], bb_gt[..., 3])
147
148        w = np.maximum(0., xx2 - xx1)
149        h = np.maximum(0., yy2 - yy1)
150        intersection = w * h
151
152        area_test = (bb_test[..., 2] - bb_test[..., 0]) *                    (bb_test[..., 3] - bb_test[..., 1])
153        area_gt = (bb_gt[..., 2] - bb_gt[..., 0]) *                  (bb_gt[..., 3] - bb_gt[..., 1])
154
155        union = area_test + area_gt - intersection
156
157        return intersection / (union + 1e-6)
158
159    def associate_detections(self, detections: np.ndarray,
160                            trackers: np.ndarray
161                            ) -> Tuple[np.ndarray, np.ndarray, np.ndarray]:
162        """
163        Assign detections to tracked objects using Hungarian algorithm.
164        """
165        if len(trackers) == 0:
166            return np.empty((0, 2), dtype=int),                    np.arange(len(detections)),                    np.empty((0,), dtype=int)
167
168        iou_matrix = self.iou_batch(detections, trackers)
169
170        if min(iou_matrix.shape) > 0:
171            # Use Hungarian algorithm for optimal assignment
172            row_indices, col_indices = linear_sum_assignment(-iou_matrix)
173            matched_indices = np.array(list(zip(row_indices, col_indices)))
174        else:
175            matched_indices = np.empty(shape=(0, 2))
176
177        unmatched_detections = []
178        for d in range(len(detections)):
179            if d not in matched_indices[:, 0]:
180                unmatched_detections.append(d)
181
182        unmatched_trackers = []
183        for t in range(len(trackers)):
184            if t not in matched_indices[:, 1]:
185                unmatched_trackers.append(t)
186
187        # Filter out low IoU matches
188        matches = []
189        for m in matched_indices:
190            if iou_matrix[m[0], m[1]] < self.iou_threshold:
191                unmatched_detections.append(m[0])
192                unmatched_trackers.append(m[1])
193            else:
194                matches.append(m.reshape(1, 2))
195
196        matches = np.concatenate(matches, axis=0) if matches else np.empty((0, 2), dtype=int)
197
198        return matches, np.array(unmatched_detections), np.array(unmatched_trackers)
199
200    def update(self, detections: np.ndarray) -> np.ndarray:
201        """
202        Update tracker with new detections.
203
204        Args:
205            detections: (N, 5) array [x1, y1, x2, y2, score]
206
207        Returns:
208            (M, 5) array of active tracks [x1, y1, x2, y2, track_id]
209        """
210        self.frame_count += 1
211
212        # Get predicted positions from existing trackers
213        predicted_positions = np.zeros((len(self.trackers), 4))
214        to_delete = []
215
216        for t, tracker in enumerate(self.trackers):
217            pos = tracker.predict()
218            predicted_positions[t] = pos
219            if np.any(np.isnan(pos)):
220                to_delete.append(t)
221
222        # Remove invalid trackers
223        for t in reversed(to_delete):
224            self.trackers.pop(t)
225        predicted_positions = np.delete(predicted_positions, to_delete, axis=0)
226
227        # Associate detections with trackers
228        matched, unmatched_dets, unmatched_trks = self.associate_detections(
229            detections[:, :4], predicted_positions
230        )
231
232        # Update matched trackers
233        for m in matched:
234            self.trackers[m[1]].update(detections[m[0], :4])
235
236        # Create new trackers for unmatched detections
237        for i in unmatched_dets:
238            self.trackers.append(KalmanTracker(detections[i, :4]))
239
240        # Return active tracks
241        results = []
242        for tracker in self.trackers:
243            if tracker.time_since_update <= self.max_age and                (tracker.hit_streak >= self.min_hits or self.frame_count <= self.min_hits):
244                state = tracker.get_state()
245                results.append(np.concatenate([state, [tracker.id]]))
246
247        # Remove dead trackers
248        self.trackers = [t for t in self.trackers if t.time_since_update <= self.max_age]
249
250        return np.array(results) if results else np.empty((0, 5))
251
252
253# Example
254tracker = SORT(max_age=3, min_hits=3, iou_threshold=0.3)
255print("SORT Tracker initialized")
256print("Call tracker.update(detections) with detections [x1,y1,x2,y2,score]")

1import torch
2import torch.nn as nn
3import numpy as np
4from typing import List, Tuple
5
6class TemporalConvolution(nn.Module):
7    """
8    Temporal convolutional block for video understanding.
9    """
10
11    def __init__(self, in_channels: int, out_channels: int,
12                 kernel_size: int = 3):
13        super().__init__()
14
15        self.conv = nn.Conv3d(
16            in_channels, out_channels,
17            kernel_size=(kernel_size, 1, 1),
18            padding=(kernel_size // 2, 0, 0)
19        )
20        self.bn = nn.BatchNorm3d(out_channels)
21        self.relu = nn.ReLU(inplace=True)
22
23    def forward(self, x: torch.Tensor) -> torch.Tensor:
24        return self.relu(self.bn(self.conv(x)))
25
26
27class C3DBlock(nn.Module):
28    """
29    C3D-style 3D convolutional block.
30    """
31
32    def __init__(self, in_channels: int, out_channels: int):
33        super().__init__()
34
35        self.conv = nn.Conv3d(
36            in_channels, out_channels,
37            kernel_size=3, padding=1
38        )
39        self.bn = nn.BatchNorm3d(out_channels)
40        self.relu = nn.ReLU(inplace=True)
41        self.pool = nn.MaxPool3d(kernel_size=2, stride=2)
42
43    def forward(self, x: torch.Tensor) -> torch.Tensor:
44        x = self.relu(self.bn(self.conv(x)))
45        return self.pool(x)
46
47
48class SimpleActionRecognizer(nn.Module):
49    """
50    Simple 3D CNN for action recognition.
51
52    Input: Video clip of shape (B, C, T, H, W)
53    Output: Action class logits (B, num_classes)
54    """
55
56    def __init__(self, num_classes: int = 10, clip_length: int = 16):
57        super().__init__()
58
59        self.features = nn.Sequential(
60            C3DBlock(3, 64),      # (T/2, H/2, W/2)
61            C3DBlock(64, 128),    # (T/4, H/4, W/4)
62            C3DBlock(128, 256),   # (T/8, H/8, W/8)
63            C3DBlock(256, 512),   # (T/16, H/16, W/16)
64        )
65
66        self.classifier = nn.Sequential(
67            nn.AdaptiveAvgPool3d((1, 1, 1)),
68            nn.Flatten(),
69            nn.Linear(512, 256),
70            nn.ReLU(),
71            nn.Dropout(0.5),
72            nn.Linear(256, num_classes)
73        )
74
75    def forward(self, x: torch.Tensor) -> torch.Tensor:
76        features = self.features(x)
77        return self.classifier(features)
78
79
80class TwoStreamNetwork(nn.Module):
81    """
82    Two-Stream architecture for action recognition.
83
84    Combines:
85    - Spatial stream: Appearance from RGB frames
86    - Temporal stream: Motion from optical flow
87    """
88
89    def __init__(self, num_classes: int = 10):
90        super().__init__()
91
92        # Spatial stream (RGB)
93        self.spatial_stream = nn.Sequential(
94            C3DBlock(3, 64),
95            C3DBlock(64, 128),
96            C3DBlock(128, 256),
97            nn.AdaptiveAvgPool3d((1, 1, 1)),
98            nn.Flatten()
99        )
100
101        # Temporal stream (optical flow)
102        self.temporal_stream = nn.Sequential(
103            C3DBlock(2, 64),  # 2 channels for flow (u, v)
104            C3DBlock(64, 128),
105            C3DBlock(128, 256),
106            nn.AdaptiveAvgPool3d((1, 1, 1)),
107            nn.Flatten()
108        )
109
110        # Fusion
111        self.classifier = nn.Sequential(
112            nn.Linear(256 + 256, 512),
113            nn.ReLU(),
114            nn.Dropout(0.5),
115            nn.Linear(512, num_classes)
116        )
117
118    def forward(self, rgb: torch.Tensor,
119                flow: torch.Tensor) -> torch.Tensor:
120        spatial_features = self.spatial_stream(rgb)
121        temporal_features = self.temporal_stream(flow)
122
123        combined = torch.cat([spatial_features, temporal_features], dim=1)
124        return self.classifier(combined)
125
126
127class SlowFastNetwork(nn.Module):
128    """
129    Simplified SlowFast architecture.
130
131    - Slow pathway: Low frame rate, learns spatial semantics
132    - Fast pathway: High frame rate, captures fine motion
133    """
134
135    def __init__(self, num_classes: int = 10, alpha: int = 8):
136        """
137        Args:
138            alpha: Speed ratio between Fast and Slow (e.g., 8x faster)
139        """
140        super().__init__()
141        self.alpha = alpha
142
143        # Slow pathway
144        self.slow_conv1 = C3DBlock(3, 64)
145        self.slow_conv2 = C3DBlock(64 + 64 // 8, 128)  # +lateral connection
146        self.slow_conv3 = C3DBlock(128, 256)
147
148        # Fast pathway (narrower channels)
149        self.fast_conv1 = C3DBlock(3, 8)
150        self.fast_conv2 = C3DBlock(8, 16)
151        self.fast_conv3 = C3DBlock(16, 32)
152
153        # Lateral connections (fast -> slow)
154        self.lateral1 = nn.Conv3d(8, 64 // 8, kernel_size=1)
155
156        # Classifier
157        self.classifier = nn.Sequential(
158            nn.AdaptiveAvgPool3d((1, 1, 1)),
159            nn.Flatten(),
160            nn.Linear(256 + 32, num_classes)
161        )
162
163    def forward(self, slow_input: torch.Tensor,
164                fast_input: torch.Tensor) -> torch.Tensor:
165        """
166        Args:
167            slow_input: (B, C, T, H, W) slow pathway input
168            fast_input: (B, C, T*alpha, H, W) fast pathway input
169        """
170        # Fast pathway
171        fast_x = self.fast_conv1(fast_input)
172        fast_x = self.fast_conv2(fast_x)
173        fast_x = self.fast_conv3(fast_x)
174
175        # Slow pathway with lateral connections
176        slow_x = self.slow_conv1(slow_input)
177
178        # Lateral connection from fast to slow
179        lateral = self.lateral1(self.fast_conv1(fast_input))
180        # Temporally downsample fast features
181        lateral = lateral[:, :, ::self.alpha, :, :]
182        slow_x = torch.cat([slow_x, lateral], dim=1)
183
184        slow_x = self.slow_conv2(slow_x)
185        slow_x = self.slow_conv3(slow_x)
186
187        # Combine pathways
188        slow_pool = nn.AdaptiveAvgPool3d((1, 1, 1))(slow_x)
189        fast_pool = nn.AdaptiveAvgPool3d((1, 1, 1))(fast_x)
190
191        combined = torch.cat([slow_pool, fast_pool], dim=1)
192        combined = combined.flatten(1)
193
194        return self.classifier(combined)
195
196
197# Example
198action_model = SimpleActionRecognizer(num_classes=10, clip_length=16)
199print(f"Action Recognizer parameters: {sum(p.numel() for p in action_model.parameters()):,}")
200
201# Test input
202clip = torch.randn(2, 3, 16, 112, 112)  # B, C, T, H, W
203output = action_model(clip)
204print(f"Input shape: {clip.shape}")
205print(f"Output shape: {output.shape}")