Advanced Computer Vision

1# R-CNN Pipeline (conceptual)
2def rcnn_detect(image):
3    # 1. Generate region proposals (Selective Search)
4    regions = selective_search(image)  # ~2000 regions
5
6    # 2. Extract features for each region
7    features = []
8    for region in regions:
9        # Warp region to fixed size
10        warped = warp_region(region, (227, 227))
11        # Extract CNN features
12        feature = alexnet.forward(warped)
13        features.append(feature)
14
15    # 3. Classify each region with SVM
16    predictions = svm.predict(features)
17
18    # 4. Refine bounding boxes with regressor
19    refined_boxes = bbox_regressor.refine(regions, predictions)
20
21    return refined_boxes, predictions

1import torch
2import torch.nn as nn
3import torchvision
4
5class FastRCNN(nn.Module):
6    def __init__(self, num_classes):
7        super().__init__()
8
9        # Backbone CNN (entire image)
10        self.backbone = torchvision.models.resnet50(pretrained=True)
11        self.backbone = nn.Sequential(*list(self.backbone.children())[:-2])
12
13        # RoI pooling
14        self.roi_pool = torchvision.ops.RoIPool(
15            output_size=(7, 7),
16            spatial_scale=1.0/16  # depends on backbone stride
17        )
18
19        # Classification and bbox regression heads
20        self.fc = nn.Sequential(
21            nn.Linear(2048 * 7 * 7, 4096),
22            nn.ReLU(),
23            nn.Dropout(0.5),
24            nn.Linear(4096, 4096),
25            nn.ReLU(),
26            nn.Dropout(0.5)
27        )
28
29        self.cls_score = nn.Linear(4096, num_classes)
30        self.bbox_pred = nn.Linear(4096, num_classes * 4)
31
32    def forward(self, image, rois):
33        # Extract features from entire image
34        features = self.backbone(image)
35
36        # Pool features for each RoI
37        pooled_features = self.roi_pool(features, rois)
38        pooled_features = pooled_features.view(pooled_features.size(0), -1)
39
40        # Fully connected layers
41        fc_features = self.fc(pooled_features)
42
43        # Classification and bbox regression
44        class_scores = self.cls_score(fc_features)
45        bbox_deltas = self.bbox_pred(fc_features)
46
47        return class_scores, bbox_deltas

1class RegionProposalNetwork(nn.Module):
2    """
3    RPN generates region proposals from feature maps.
4    Uses anchor boxes of multiple scales and aspect ratios.
5    """
6    def __init__(self, in_channels, num_anchors=9):
7        super().__init__()
8
9        # Shared 3x3 conv
10        self.conv = nn.Conv2d(in_channels, 512, kernel_size=3, padding=1)
11
12        # Classification: object vs background
13        self.cls_logits = nn.Conv2d(512, num_anchors * 2, kernel_size=1)
14
15        # Bbox regression: refine anchor boxes
16        self.bbox_pred = nn.Conv2d(512, num_anchors * 4, kernel_size=1)
17
18    def forward(self, features):
19        # features: (B, C, H, W)
20        x = torch.relu(self.conv(features))
21
22        # For each anchor at each spatial location
23        cls_logits = self.cls_logits(x)  # (B, num_anchors*2, H, W)
24        bbox_pred = self.bbox_pred(x)    # (B, num_anchors*4, H, W)
25
26        return cls_logits, bbox_pred

1import torch
2import torch.nn as nn
3
4class YOLOv3(nn.Module):
5    """
6    YOLOv3 architecture with Darknet-53 backbone.
7    """
8    def __init__(self, num_classes=80):
9        super().__init__()
10        self.num_classes = num_classes
11
12        # Darknet-53 backbone
13        self.backbone = Darknet53()
14
15        # Detection heads at 3 scales
16        self.detection_scale1 = self._make_detection_layer(1024, num_classes)
17        self.detection_scale2 = self._make_detection_layer(512, num_classes)
18        self.detection_scale3 = self._make_detection_layer(256, num_classes)
19
20    def _make_detection_layer(self, in_channels, num_classes):
21        """
22        Each detection layer predicts:
23        - 3 anchor boxes per cell
24        - Each box: (x, y, w, h, confidence, class_probs)
25        """
26        num_anchors = 3
27        out_channels = num_anchors * (5 + num_classes)
28
29        return nn.Sequential(
30            ConvBlock(in_channels, in_channels * 2, 3),
31            nn.Conv2d(in_channels * 2, out_channels, 1)
32        )
33
34    def forward(self, x):
35        # Backbone forward pass (feature pyramid)
36        feat_small, feat_medium, feat_large = self.backbone(x)
37
38        # Detections at 3 scales
39        det_small = self.detection_scale1(feat_small)    # 13x13
40        det_medium = self.detection_scale2(feat_medium)  # 26x26
41        det_large = self.detection_scale3(feat_large)    # 52x52
42
43        return det_small, det_medium, det_large
44
45
46class Darknet53(nn.Module):
47    """Darknet-53 backbone for feature extraction."""
48    def __init__(self):
49        super().__init__()
50
51        # Initial conv
52        self.conv1 = ConvBlock(3, 32, 3)
53
54        # Residual blocks
55        self.res1 = self._make_layer(32, 64, 1)
56        self.res2 = self._make_layer(64, 128, 2)
57        self.res3 = self._make_layer(128, 256, 8)
58        self.res4 = self._make_layer(256, 512, 8)
59        self.res5 = self._make_layer(512, 1024, 4)
60
61    def _make_layer(self, in_channels, out_channels, num_blocks):
62        layers = [ConvBlock(in_channels, out_channels, 3, stride=2)]
63        for _ in range(num_blocks):
64            layers.append(ResidualBlock(out_channels))
65        return nn.Sequential(*layers)
66
67    def forward(self, x):
68        x = self.conv1(x)
69        x = self.res1(x)
70        x = self.res2(x)
71        feat_large = self.res3(x)    # 52x52, for small objects
72        feat_medium = self.res4(feat_large)  # 26x26, for medium objects
73        feat_small = self.res5(feat_medium)  # 13x13, for large objects
74
75        return feat_small, feat_medium, feat_large
76
77
78class ConvBlock(nn.Module):
79    """Conv + BatchNorm + LeakyReLU"""
80    def __init__(self, in_channels, out_channels, kernel_size, stride=1):
81        super().__init__()
82        padding = (kernel_size - 1) // 2
83        self.conv = nn.Conv2d(in_channels, out_channels, kernel_size,
84                              stride=stride, padding=padding, bias=False)
85        self.bn = nn.BatchNorm2d(out_channels)
86        self.activation = nn.LeakyReLU(0.1)
87
88    def forward(self, x):
89        return self.activation(self.bn(self.conv(x)))
90
91
92class ResidualBlock(nn.Module):
93    """Residual block: 1x1 conv -> 3x3 conv with skip connection"""
94    def __init__(self, channels):
95        super().__init__()
96        self.conv1 = ConvBlock(channels, channels // 2, 1)
97        self.conv2 = ConvBlock(channels // 2, channels, 3)
98
99    def forward(self, x):
100        residual = x
101        out = self.conv1(x)
102        out = self.conv2(out)
103        return out + residual

1def decode_yolo_predictions(predictions, anchors, num_classes, img_size):
2    """
3    Decode YOLO network output to bounding boxes.
4
5    predictions: (batch, num_anchors * (5 + num_classes), grid_h, grid_w)
6    anchors: predefined anchor box sizes
7    """
8    batch_size = predictions.size(0)
9    grid_size = predictions.size(2)
10    num_anchors = len(anchors)
11
12    # Reshape predictions
13    predictions = predictions.view(
14        batch_size, num_anchors, 5 + num_classes, grid_size, grid_size
15    ).permute(0, 1, 3, 4, 2).contiguous()
16
17    # Extract components
18    x = torch.sigmoid(predictions[..., 0])  # Center x
19    y = torch.sigmoid(predictions[..., 1])  # Center y
20    w = predictions[..., 2]  # Width
21    h = predictions[..., 3]  # Height
22    conf = torch.sigmoid(predictions[..., 4])  # Objectness confidence
23    pred_cls = torch.sigmoid(predictions[..., 5:])  # Class probabilities
24
25    # Generate grid offsets
26    grid_x = torch.arange(grid_size).repeat(grid_size, 1).float()
27    grid_y = torch.arange(grid_size).repeat(grid_size, 1).t().float()
28
29    # Calculate bbox coordinates
30    stride = img_size / grid_size
31    pred_boxes = torch.zeros_like(predictions[..., :4])
32    pred_boxes[..., 0] = (x + grid_x) * stride  # Center x in pixels
33    pred_boxes[..., 1] = (y + grid_y) * stride  # Center y in pixels
34    pred_boxes[..., 2] = torch.exp(w) * anchors[:, 0] * stride  # Width
35    pred_boxes[..., 3] = torch.exp(h) * anchors[:, 1] * stride  # Height
36
37    return pred_boxes, conf, pred_cls
38
39
40class YOLOLoss(nn.Module):
41    """
42    YOLO loss combines:
43    1. Localization loss (bbox coordinates)
44    2. Confidence loss (objectness)
45    3. Classification loss (class probabilities)
46    """
47    def __init__(self, num_classes):
48        super().__init__()
49        self.num_classes = num_classes
50        self.mse_loss = nn.MSELoss()
51        self.bce_loss = nn.BCELoss()
52
53    def forward(self, predictions, targets):
54        # Extract predictions
55        pred_boxes, pred_conf, pred_cls = predictions
56
57        # Build target tensors
58        obj_mask = targets['obj_mask']  # Which cells have objects
59        noobj_mask = targets['noobj_mask']  # Which cells don't
60        tx, ty, tw, th = targets['boxes']  # Target boxes
61        tcls = targets['classes']  # Target classes
62
63        # Localization loss (only for cells with objects)
64        loss_x = self.mse_loss(pred_boxes[..., 0][obj_mask], tx[obj_mask])
65        loss_y = self.mse_loss(pred_boxes[..., 1][obj_mask], ty[obj_mask])
66        loss_w = self.mse_loss(pred_boxes[..., 2][obj_mask], tw[obj_mask])
67        loss_h = self.mse_loss(pred_boxes[..., 3][obj_mask], th[obj_mask])
68        loss_loc = loss_x + loss_y + loss_w + loss_h
69
70        # Confidence loss
71        loss_conf_obj = self.bce_loss(pred_conf[obj_mask], torch.ones_like(pred_conf[obj_mask]))
72        loss_conf_noobj = self.bce_loss(pred_conf[noobj_mask], torch.zeros_like(pred_conf[noobj_mask]))
73        loss_conf = loss_conf_obj + 0.5 * loss_conf_noobj  # Weight down no-object loss
74
75        # Classification loss
76        loss_cls = self.bce_loss(pred_cls[obj_mask], tcls[obj_mask])
77
78        # Total loss
79        total_loss = loss_loc + loss_conf + loss_cls
80
81        return total_loss, {
82            'localization': loss_loc.item(),
83            'confidence': loss_conf.item(),
84            'classification': loss_cls.item()
85        }

1import torch
2from PIL import Image
3import torchvision.transforms as transforms
4
5# Load pretrained YOLOv5 (using Ultralytics implementation)
6model = torch.hub.load('ultralytics/yolov5', 'yolov5s', pretrained=True)
7model.eval()
8
9def detect_objects(image_path, conf_threshold=0.5):
10    """Detect objects using YOLOv5"""
11
12    # Load image
13    img = Image.open(image_path)
14
15    # Inference
16    results = model(img)
17
18    # Parse results
19    detections = results.pandas().xyxy[0]  # Bounding boxes as pandas df
20
21    # Filter by confidence
22    detections = detections[detections['confidence'] > conf_threshold]
23
24    print(f"Detected {len(detections)} objects:")
25    for _, det in detections.iterrows():
26        print(f"  {det['name']}: {det['confidence']:.2f} at "
27              f"[{det['xmin']:.0f}, {det['ymin']:.0f}, {det['xmax']:.0f}, {det['ymax']:.0f}]")
28
29    # Visualize
30    results.show()
31
32    return detections
33
34# Usage
35detections = detect_objects('street_scene.jpg', conf_threshold=0.6)

1class UNet(nn.Module):
2    """
3    U-Net architecture for semantic segmentation.
4    Encoder-decoder structure with skip connections.
5    """
6    def __init__(self, in_channels=3, num_classes=21):
7        super().__init__()
8
9        # Encoder (downsampling path)
10        self.enc1 = self._double_conv(in_channels, 64)
11        self.enc2 = self._double_conv(64, 128)
12        self.enc3 = self._double_conv(128, 256)
13        self.enc4 = self._double_conv(256, 512)
14
15        # Bottleneck
16        self.bottleneck = self._double_conv(512, 1024)
17
18        # Decoder (upsampling path)
19        self.upconv4 = nn.ConvTranspose2d(1024, 512, 2, stride=2)
20        self.dec4 = self._double_conv(1024, 512)  # 1024 from skip connection
21
22        self.upconv3 = nn.ConvTranspose2d(512, 256, 2, stride=2)
23        self.dec3 = self._double_conv(512, 256)
24
25        self.upconv2 = nn.ConvTranspose2d(256, 128, 2, stride=2)
26        self.dec2 = self._double_conv(256, 128)
27
28        self.upconv1 = nn.ConvTranspose2d(128, 64, 2, stride=2)
29        self.dec1 = self._double_conv(128, 64)
30
31        # Output
32        self.out = nn.Conv2d(64, num_classes, 1)
33
34        # Pooling
35        self.pool = nn.MaxPool2d(2, 2)
36
37    def _double_conv(self, in_channels, out_channels):
38        """Two conv layers with BatchNorm and ReLU"""
39        return nn.Sequential(
40            nn.Conv2d(in_channels, out_channels, 3, padding=1),
41            nn.BatchNorm2d(out_channels),
42            nn.ReLU(inplace=True),
43            nn.Conv2d(out_channels, out_channels, 3, padding=1),
44            nn.BatchNorm2d(out_channels),
45            nn.ReLU(inplace=True)
46        )
47
48    def forward(self, x):
49        # Encoder
50        enc1 = self.enc1(x)        # 64 channels
51        enc2 = self.enc2(self.pool(enc1))  # 128 channels
52        enc3 = self.enc3(self.pool(enc2))  # 256 channels
53        enc4 = self.enc4(self.pool(enc3))  # 512 channels
54
55        # Bottleneck
56        bottleneck = self.bottleneck(self.pool(enc4))  # 1024 channels
57
58        # Decoder with skip connections
59        dec4 = self.upconv4(bottleneck)
60        dec4 = torch.cat([dec4, enc4], dim=1)  # Skip connection
61        dec4 = self.dec4(dec4)
62
63        dec3 = self.upconv3(dec4)
64        dec3 = torch.cat([dec3, enc3], dim=1)
65        dec3 = self.dec3(dec3)
66
67        dec2 = self.upconv2(dec3)
68        dec2 = torch.cat([dec2, enc2], dim=1)
69        dec2 = self.dec2(dec2)
70
71        dec1 = self.upconv1(dec2)
72        dec1 = torch.cat([dec1, enc1], dim=1)
73        dec1 = self.dec1(dec1)
74
75        # Output
76        return self.out(dec1)
77
78
79# Training U-Net
80def train_unet(model, train_loader, val_loader, num_epochs=50):
81    device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
82    model = model.to(device)
83
84    criterion = nn.CrossEntropyLoss()
85    optimizer = torch.optim.Adam(model.parameters(), lr=0.001)
86
87    for epoch in range(num_epochs):
88        model.train()
89        train_loss = 0.0
90
91        for images, masks in train_loader:
92            images, masks = images.to(device), masks.to(device)
93
94            # Forward pass
95            outputs = model(images)
96            loss = criterion(outputs, masks)
97
98            # Backward pass
99            optimizer.zero_grad()
100            loss.backward()
101            optimizer.step()
102
103            train_loss += loss.item()
104
105        # Validation
106        model.eval()
107        val_loss = 0.0
108        with torch.no_grad():
109            for images, masks in val_loader:
110                images, masks = images.to(device), masks.to(device)
111                outputs = model(images)
112                loss = criterion(outputs, masks)
113                val_loss += loss.item()
114
115        print(f"Epoch {epoch+1}/{num_epochs}")
116        print(f"  Train Loss: {train_loss/len(train_loader):.4f}")
117        print(f"  Val Loss: {val_loss/len(val_loader):.4f}")

1class AtrousConv(nn.Module):
2    """Atrous (dilated) convolution for larger receptive field"""
3    def __init__(self, in_channels, out_channels, dilation_rate):
4        super().__init__()
5        self.conv = nn.Conv2d(
6            in_channels, out_channels, 3,
7            padding=dilation_rate,
8            dilation=dilation_rate,
9            bias=False
10        )
11        self.bn = nn.BatchNorm2d(out_channels)
12        self.relu = nn.ReLU(inplace=True)
13
14    def forward(self, x):
15        return self.relu(self.bn(self.conv(x)))
16
17
18class ASPP(nn.Module):
19    """
20    Atrous Spatial Pyramid Pooling (ASPP)
21    Multiple parallel atrous convolutions with different rates
22    """
23    def __init__(self, in_channels, out_channels):
24        super().__init__()
25
26        self.conv1 = nn.Conv2d(in_channels, out_channels, 1, bias=False)
27        self.bn1 = nn.BatchNorm2d(out_channels)
28
29        self.atrous1 = AtrousConv(in_channels, out_channels, dilation_rate=6)
30        self.atrous2 = AtrousConv(in_channels, out_channels, dilation_rate=12)
31        self.atrous3 = AtrousConv(in_channels, out_channels, dilation_rate=18)
32
33        self.global_pool = nn.AdaptiveAvgPool2d(1)
34        self.conv_pool = nn.Conv2d(in_channels, out_channels, 1, bias=False)
35        self.bn_pool = nn.BatchNorm2d(out_channels)
36
37        self.conv_out = nn.Conv2d(out_channels * 5, out_channels, 1, bias=False)
38        self.bn_out = nn.BatchNorm2d(out_channels)
39
40    def forward(self, x):
41        size = x.shape[2:]
42
43        # Different atrous rates
44        feat1 = self.bn1(self.conv1(x))
45        feat2 = self.atrous1(x)
46        feat3 = self.atrous2(x)
47        feat4 = self.atrous3(x)
48
49        # Global pooling branch
50        feat5 = self.global_pool(x)
51        feat5 = self.bn_pool(self.conv_pool(feat5))
52        feat5 = torch.nn.functional.interpolate(feat5, size=size, mode='bilinear')
53
54        # Concatenate all features
55        out = torch.cat([feat1, feat2, feat3, feat4, feat5], dim=1)
56        out = self.bn_out(self.conv_out(out))
57
58        return out

1# Using torchvision's Mask R-CNN
2import torchvision
3from torchvision.models.detection import maskrcnn_resnet50_fpn
4from torchvision.models.detection.faster_rcnn import FastRCNNPredictor
5from torchvision.models.detection.mask_rcnn import MaskRCNNPredictor
6
7def get_instance_segmentation_model(num_classes):
8    """Load pretrained Mask R-CNN and modify for custom classes"""
9
10    # Load pretrained model
11    model = maskrcnn_resnet50_fpn(pretrained=True)
12
13    # Replace box predictor
14    in_features = model.roi_heads.box_predictor.cls_score.in_features
15    model.roi_heads.box_predictor = FastRCNNPredictor(in_features, num_classes)
16
17    # Replace mask predictor
18    in_features_mask = model.roi_heads.mask_predictor.conv5_mask.in_channels
19    hidden_layer = 256
20    model.roi_heads.mask_predictor = MaskRCNNPredictor(
21        in_features_mask, hidden_layer, num_classes
22    )
23
24    return model
25
26
27def predict_instance_segmentation(model, image, threshold=0.5):
28    """Predict instance segmentation masks"""
29    model.eval()
30    device = next(model.parameters()).device
31
32    # Preprocess
33    transform = transforms.Compose([transforms.ToTensor()])
34    image_tensor = transform(image).unsqueeze(0).to(device)
35
36    # Predict
37    with torch.no_grad():
38        predictions = model(image_tensor)[0]
39
40    # Filter by score
41    keep = predictions['scores'] > threshold
42
43    boxes = predictions['boxes'][keep].cpu().numpy()
44    labels = predictions['labels'][keep].cpu().numpy()
45    scores = predictions['scores'][keep].cpu().numpy()
46    masks = predictions['masks'][keep].cpu().numpy()
47
48    return boxes, labels, scores, masks
49
50
51# Visualize instance segmentation
52def visualize_instance_segmentation(image, boxes, labels, scores, masks, class_names):
53    import matplotlib.pyplot as plt
54    import matplotlib.patches as patches
55
56    fig, ax = plt.subplots(1, figsize=(12, 8))
57    ax.imshow(image)
58
59    colors = plt.cm.get_cmap('tab10')
60
61    for i in range(len(boxes)):
62        # Draw bounding box
63        x1, y1, x2, y2 = boxes[i]
64        rect = patches.Rectangle(
65            (x1, y1), x2-x1, y2-y1,
66            linewidth=2, edgecolor=colors(i % 10), facecolor='none'
67        )
68        ax.add_patch(rect)
69
70        # Draw label
71        label_text = f"{class_names[labels[i]]}: {scores[i]:.2f}"
72        ax.text(x1, y1-5, label_text, color='white', fontsize=10,
73                bbox=dict(facecolor=colors(i % 10), alpha=0.8))
74
75        # Draw mask
76        mask = masks[i, 0] > 0.5
77        colored_mask = np.zeros((*mask.shape, 4))
78        colored_mask[mask] = (*colors(i % 10)[:3], 0.5)
79        ax.imshow(colored_mask)
80
81    plt.axis('off')
82    plt.tight_layout()
83    plt.show()

1# Using OpenPose-style architecture
2class PoseNet(nn.Module):
3    """
4    Multi-person pose estimation network.
5    Predicts heatmaps for keypoints and Part Affinity Fields (PAFs) for limb connections.
6    """
7    def __init__(self, num_keypoints=18):
8        super().__init__()
9        self.num_keypoints = num_keypoints
10
11        # Backbone
12        self.backbone = torchvision.models.vgg19(pretrained=True).features
13
14        # Stage 1: Initial prediction
15        self.stage1_heatmap = self._make_stage(512, num_keypoints)
16        self.stage1_paf = self._make_stage(512, num_keypoints * 2)
17
18        # Refinement stages (typically 6 stages total)
19        self.refinement_stages = nn.ModuleList([
20            self._make_refinement_stage(num_keypoints)
21            for _ in range(5)
22        ])
23
24    def _make_stage(self, in_channels, out_channels):
25        return nn.Sequential(
26            nn.Conv2d(in_channels, 512, 3, padding=1),
27            nn.ReLU(inplace=True),
28            nn.Conv2d(512, out_channels, 1)
29        )
30
31    def _make_refinement_stage(self, num_keypoints):
32        # Refines both heatmaps and PAFs
33        return nn.ModuleDict({
34            'heatmap': self._make_stage(512 + num_keypoints + num_keypoints * 2, num_keypoints),
35            'paf': self._make_stage(512 + num_keypoints + num_keypoints * 2, num_keypoints * 2)
36        })
37
38    def forward(self, x):
39        # Extract features
40        features = self.backbone(x)
41
42        # Stage 1
43        heatmap = self.stage1_heatmap(features)
44        paf = self.stage1_paf(features)
45
46        # Refinement stages
47        for stage in self.refinement_stages:
48            # Concatenate features, previous heatmap, and PAF
49            combined = torch.cat([features, heatmap, paf], dim=1)
50            heatmap = stage['heatmap'](combined)
51            paf = stage['paf'](combined)
52
53        return heatmap, paf

1# 1. Model quantization for faster inference
2import torch.quantization
3
4def quantize_model(model):
5    """Apply dynamic quantization"""
6    quantized_model = torch.quantization.quantize_dynamic(
7        model, {nn.Linear, nn.Conv2d}, dtype=torch.qint8
8    )
9    return quantized_model
10
11# 2. TorchScript for deployment
12def export_torchscript(model, example_input):
13    """Convert to TorchScript for C++ deployment"""
14    model.eval()
15    traced_model = torch.jit.trace(model, example_input)
16    traced_model.save('model_traced.pt')
17
18# 3. ONNX export for cross-platform deployment
19def export_onnx(model, example_input, output_path):
20    """Export to ONNX format"""
21    torch.onnx.export(
22        model, example_input, output_path,
23        export_params=True,
24        opset_version=11,
25        input_names=['input'],
26        output_names=['output'],
27        dynamic_axes={'input': {0: 'batch_size'}, 'output': {0: 'batch_size'}}
28    )