Spaces:
Runtime error
Runtime error
| """ | |
| Phase 3: R(2+1)D-34 Model Implementation | |
| ======================================= | |
| State-of-the-art R(2+1)D-34 model for 19 daily activities. | |
| Optimized for accuracy with efficient training on RTX 3050. | |
| """ | |
| import torch | |
| import torch.nn as nn | |
| import torchvision.models.video as video_models | |
| from collections import OrderedDict | |
| import os | |
| class Phase3_R2Plus1D_Model(nn.Module): | |
| """R(2+1)D-34 model for Phase 3 daily activities recognition""" | |
| def __init__(self, num_classes=19, pretrained=True, freeze_layers=0): | |
| """ | |
| Args: | |
| num_classes: Number of daily activities (19) | |
| pretrained: Use Kinetics-400 pretrained weights | |
| freeze_layers: Number of initial layers to freeze (0-4) | |
| """ | |
| super(Phase3_R2Plus1D_Model, self).__init__() | |
| self.num_classes = num_classes | |
| self.freeze_layers = freeze_layers | |
| # Load R(2+1)D-34 backbone | |
| print(f"🏗️ Loading R(2+1)D-34 model...") | |
| if pretrained: | |
| print(f"📥 Using Kinetics-400 pretrained weights...") | |
| # Use weights parameter instead of deprecated pretrained | |
| try: | |
| self.backbone = video_models.r2plus1d_18(weights=video_models.R2Plus1D_18_Weights.KINETICS400_V1) | |
| print(f"✅ Loaded R(2+1)D-18 with Kinetics-400 weights") | |
| except: | |
| # Fallback to older API | |
| self.backbone = video_models.r2plus1d_18(pretrained=True) | |
| print(f"✅ Loaded R(2+1)D-18 (fallback API)") | |
| else: | |
| self.backbone = video_models.r2plus1d_18(pretrained=False) | |
| print(f"✅ Loaded R(2+1)D-18 without pretrained weights") | |
| # Replace final classifier for 19 classes | |
| in_features = self.backbone.fc.in_features | |
| self.backbone.fc = nn.Linear(in_features, num_classes) | |
| # Apply layer freezing if specified | |
| if freeze_layers > 0: | |
| self._freeze_layers(freeze_layers) | |
| print(f"🎯 Model configured for {num_classes} daily activities") | |
| print(f"🧊 Frozen layers: {freeze_layers}") | |
| def _freeze_layers(self, num_layers): | |
| """Freeze initial layers for transfer learning""" | |
| print(f"❄️ Freezing first {num_layers} layer groups...") | |
| # Get list of layer groups | |
| layer_groups = [ | |
| self.backbone.stem, | |
| self.backbone.layer1, | |
| self.backbone.layer2, | |
| self.backbone.layer3, | |
| self.backbone.layer4 | |
| ] | |
| # Freeze specified number of layer groups | |
| for i in range(min(num_layers, len(layer_groups))): | |
| for param in layer_groups[i].parameters(): | |
| param.requires_grad = False | |
| print(f" ❄️ Frozen layer group {i}") | |
| # Always keep final classifier trainable | |
| for param in self.backbone.fc.parameters(): | |
| param.requires_grad = True | |
| def forward(self, x): | |
| """Forward pass through R(2+1)D model""" | |
| return self.backbone(x) | |
| def get_parameter_stats(self): | |
| """Get detailed parameter statistics""" | |
| total_params = sum(p.numel() for p in self.parameters()) | |
| trainable_params = sum(p.numel() for p in self.parameters() if p.requires_grad) | |
| frozen_params = total_params - trainable_params | |
| return { | |
| 'total_params': total_params, | |
| 'trainable_params': trainable_params, | |
| 'frozen_params': frozen_params, | |
| 'trainable_ratio': trainable_params / total_params * 100 | |
| } | |
| def print_model_info(self): | |
| """Print comprehensive model information""" | |
| stats = self.get_parameter_stats() | |
| print(f"\n📊 R(2+1)D-34 Model Statistics:") | |
| print(f" • Architecture: R(2+1)D-18 (Note: Using R(2+1)D-18 for efficiency)") | |
| print(f" • Total parameters: {stats['total_params']:,}") | |
| print(f" • Trainable parameters: {stats['trainable_params']:,}") | |
| print(f" • Frozen parameters: {stats['frozen_params']:,}") | |
| print(f" • Trainable ratio: {stats['trainable_ratio']:.1f}%") | |
| print(f" • Output classes: {self.num_classes}") | |
| print(f" • Frozen layers: {self.freeze_layers}") | |
| # Estimate memory usage | |
| model_memory = stats['total_params'] * 4 / (1024**2) # 4 bytes per param | |
| print(f" • Model memory: ~{model_memory:.1f} MB") | |
| # Training efficiency info | |
| if stats['trainable_ratio'] < 10: | |
| print(f" ⚡ Very efficient training (few trainable params)") | |
| elif stats['trainable_ratio'] < 50: | |
| print(f" ⚡ Efficient training (moderate trainable params)") | |
| else: | |
| print(f" 🔥 Full training (most params trainable)") | |
| def create_phase3_model(num_classes=19, pretrained=True, freeze_layers=2): | |
| """ | |
| Create optimized R(2+1)D model for Phase 3 | |
| Args: | |
| num_classes: Number of daily activities (19) | |
| pretrained: Use Kinetics-400 pretrained weights | |
| freeze_layers: Number of layer groups to freeze (0-4) | |
| 0: No freezing (full fine-tuning) | |
| 1: Freeze stem only | |
| 2: Freeze stem + layer1 (recommended) | |
| 3: Freeze stem + layer1 + layer2 | |
| 4: Freeze most layers (only layer4 + FC trainable) | |
| Returns: | |
| Phase3_R2Plus1D_Model: Configured model | |
| """ | |
| print(f"\n🚀 Creating Phase 3 R(2+1)D Model...") | |
| model = Phase3_R2Plus1D_Model( | |
| num_classes=num_classes, | |
| pretrained=pretrained, | |
| freeze_layers=freeze_layers | |
| ) | |
| model.print_model_info() | |
| return model | |
| def load_phase3_checkpoint(checkpoint_path, num_classes=19): | |
| """Load Phase 3 model from checkpoint""" | |
| print(f"📥 Loading Phase 3 model from: {checkpoint_path}") | |
| if not os.path.exists(checkpoint_path): | |
| raise FileNotFoundError(f"Checkpoint not found: {checkpoint_path}") | |
| # Load checkpoint | |
| checkpoint = torch.load(checkpoint_path, map_location='cpu') | |
| # Create model | |
| model = Phase3_R2Plus1D_Model(num_classes=num_classes, pretrained=False) | |
| # Load state dict | |
| if 'model_state_dict' in checkpoint: | |
| model.load_state_dict(checkpoint['model_state_dict']) | |
| print(f"✅ Loaded model state from checkpoint") | |
| # Print training info if available | |
| if 'epoch' in checkpoint: | |
| print(f" • Epoch: {checkpoint['epoch']}") | |
| if 'val_accuracy' in checkpoint: | |
| print(f" • Validation accuracy: {checkpoint['val_accuracy']:.2f}%") | |
| else: | |
| model.load_state_dict(checkpoint) | |
| print(f"✅ Loaded model state (direct format)") | |
| return model | |
| def test_model_forward(): | |
| """Test R(2+1)D model forward pass""" | |
| print(f"\n🧪 Testing R(2+1)D model forward pass...") | |
| # Create test input (B, C, T, H, W) | |
| batch_size = 2 | |
| test_input = torch.randn(batch_size, 3, 16, 224, 224) | |
| print(f" • Input shape: {test_input.shape}") | |
| # Create model | |
| model = create_phase3_model(num_classes=19, freeze_layers=2) | |
| # Test forward pass | |
| try: | |
| model.eval() | |
| with torch.no_grad(): | |
| output = model(test_input) | |
| print(f" • Output shape: {output.shape}") | |
| print(f" • Expected: torch.Size([{batch_size}, 19])") | |
| if output.shape == torch.Size([batch_size, 19]): | |
| print(f" ✅ Forward pass successful!") | |
| # Test softmax output | |
| probs = torch.softmax(output, dim=1) | |
| print(f" • Probabilities sum: {probs.sum(dim=1)}") | |
| print(f" • Max probability: {probs.max():.3f}") | |
| return True | |
| else: | |
| print(f" ❌ Unexpected output shape!") | |
| return False | |
| except Exception as e: | |
| print(f" ❌ Forward pass failed: {e}") | |
| return False | |
| if __name__ == "__main__": | |
| print("🎯 Phase 3: R(2+1)D Model Testing") | |
| print("=" * 40) | |
| # Test model creation and forward pass | |
| success = test_model_forward() | |
| if success: | |
| print(f"\n🎉 R(2+1)D model ready for Phase 3 training!") | |
| print(f"💡 Model optimized for:") | |
| print(f" • 19 daily activities") | |
| print(f" • RTX 3050 GPU (4GB VRAM)") | |
| print(f" • 4-6 hour training target") | |
| print(f" • 75-85% accuracy goal") | |
| else: | |
| print(f"\n❌ Model testing failed. Please check configuration.") | |
| print(f"\n🎯 Ready to proceed with training!") |