{"metadata":{"kernelspec":{"display_name":"Python 3","language":"python","name":"python3"},"language_info":{"codemirror_mode":{"name":"ipython","version":3},"file_extension":".py","mimetype":"text/x-python","name":"python","nbconvert_exporter":"python","pygments_lexer":"ipython3","version":"3.11.13"},"kaggle":{"accelerator":"nvidiaTeslaT4","dataSources":[{"sourceId":30892,"databundleVersionId":2715462,"isSourceIdPinned":false,"sourceType":"competition"},{"sourceId":12531607,"sourceType":"datasetVersion","datasetId":7910828}],"dockerImageVersionId":31090,"isInternetEnabled":true,"language":"python","sourceType":"notebook","isGpuEnabled":true}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"code","source":"\"\"\"\nFile cấu hình chứa tất cả các tham số quan trọng cho project\nMục đích: Tập trung quản lý các hyperparameter, đường dẫn và cài đặt\n\"\"\"\n\nimport os\nfrom dataclasses import dataclass, field\nfrom typing import List, Tuple, Optional, Dict, Any\n\n@dataclass\nclass DataConfig:\n    \"\"\"\n    Cấu hình liên quan đến dữ liệu\n    - Đường dẫn tới các thư mục chứa data\n    - Kích thước ảnh và batch size\n    - Tỷ lệ chia train/validation\n    \"\"\"\n    # Đường dẫn dữ liệu\n    train_image_path: str = '/kaggle/input/bkai-igh-neopolyp/train/train'\n    train_mask_path: str = '/kaggle/input/bkai-igh-neopolyp/train_gt/train_gt'\n    test_image_path: str = '/kaggle/input/bkai-igh-neopolyp/test/test'\n    \n    # Kích thước và batch\n    image_size: Tuple[int, int] = (512, 512)\n    batch_size: int = 4  # Giảm batch size để train lâu hơn\n    num_workers: int = 4\n    \n    # Chia dữ liệu\n    train_split: float = 0.8\n    val_split: float = 0.1\n    test_split: float = 0.1\n    \n    # Augmentation\n    use_augmentation: bool = True\n    augmentation_prob: float = 0.5\n\n@dataclass\nclass ModelConfig:\n    \"\"\"\n    Cấu hình mô hình\n    - Loại mô hình và backbone\n    - Số lớp phân loại\n    - Pretrained weights\n    \"\"\"\n    # Mô hình chính\n    model_name: str = \"segformer\"\n    backbone: str = \"nvidia/mit-b5\"\n    num_classes: int = 3\n    \n    # Labels\n    id2label: Optional[Dict[int, str]] = None\n    label2id: Optional[Dict[str, int]] = None\n    \n    # Pretrained\n    use_pretrained: bool = True\n    \n    def __post_init__(self):\n        if self.id2label is None:\n            self.id2label = {\n                0: \"neoplastic\",\n                1: \"non-neoplastic\", \n                2: \"background\"\n            }\n        if self.label2id is None:\n            self.label2id = {v: k for k, v in self.id2label.items()}\n\n@dataclass\nclass TrainingConfig:\n    \"\"\"\n    Cấu hình training\n    - Learning rate, optimizer\n    - Số epoch, early stopping\n    - Loss function và metrics\n    \"\"\"\n    # Optimizer\n    optimizer: str = \"AdamW\"\n    learning_rate: float = 1e-5  # Giảm learning rate để train ổn định hơn\n    weight_decay: float = 0.01\n    \n    # Training\n    epochs: int = 150  # Tăng epochs để train lâu hơn (2 tiếng)\n    warmup_steps: int = 1000  # Tăng warmup steps\n    \n    # Early stopping\n    patience: int = 30  # Tăng patience để không dừng sớm\n    min_delta: float = 0.0002  # Giảm min_delta để nhạy cảm hơn\n    \n    # Loss\n    loss_function: str = \"cross_entropy\"\n    class_weights: Optional[List[float]] = None\n    \n    # Metrics\n    metrics: Optional[List[str]] = None\n    \n    def __post_init__(self):\n        if self.metrics is None:\n            self.metrics = [\"mean_iou\", \"dice\", \"accuracy\"]\n\n@dataclass\nclass ExperimentConfig:\n    \"\"\"\n    Cấu hình thí nghiệm\n    - Logging, checkpointing\n    - Wandb settings\n    - Output paths\n    \"\"\"\n    # Experiment name\n    experiment_name: str = \"polyp_segmentation\"\n    run_name: str = \"segformer_b5_baseline\"\n    \n    # Paths\n    output_dir: str = \"/kaggle/working\"\n    model_save_dir: str = \"/kaggle/working/models\"\n    log_dir: str = \"/kaggle/working/logs\"\n    \n    # Logging\n    use_wandb: bool = True\n    wandb_project: str = \"polyp-segmentation\"\n    log_interval: int = 100\n    \n    # Checkpointing\n    save_best_only: bool = True\n    save_last: bool = True\n    monitor_metric: str = \"val_mean_iou\"\n    \n    # Reproducibility\n    seed: int = 42\n    deterministic: bool = True\n\n@dataclass\nclass InferenceConfig:\n    \"\"\"\n    Cấu hình inference\n    - Test Time Augmentation\n    - Post-processing\n    - Output format\n    \"\"\"\n    # TTA\n    use_tta: bool = True\n    tta_transforms: int = 8\n    \n    # Post-processing\n    use_crf: bool = False\n    use_morphology: bool = True\n    \n    # Output\n    output_format: str = \"rle\"  # \"rle\" or \"mask\"\n    submission_file: str = \"submission.csv\"\n\n# Tạo config tổng hợp\n@dataclass\nclass Config:\n    \"\"\"\n    Config tổng hợp chứa tất cả các cấu hình con\n    Mục đích: Tập trung tất cả settings vào một object duy nhất\n    \"\"\"\n    data: DataConfig = field(default_factory=DataConfig)\n    model: ModelConfig = field(default_factory=ModelConfig)\n    training: TrainingConfig = field(default_factory=TrainingConfig)\n    experiment: ExperimentConfig = field(default_factory=ExperimentConfig)\n    inference: InferenceConfig = field(default_factory=InferenceConfig)\n    \n    def create_directories(self):\n        \"\"\"\n        Tạo các thư mục cần thiết cho experiment\n        \"\"\"\n        directories = [\n            self.experiment.output_dir,\n            self.experiment.model_save_dir,\n            self.experiment.log_dir\n        ]\n        \n        for directory in directories:\n            os.makedirs(directory, exist_ok=True)\n            \n    def get_device(self):\n        \"\"\"\n        Lấy device phù hợp (GPU hoặc CPU)\n        \"\"\"\n        import torch\n        return torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\")\n\n# Tạo config mặc định\ndef get_default_config() -> Config:\n    \"\"\"\n    Hàm trả về config mặc định\n    Mục đích: Dễ dàng import và sử dụng config trong các file khác\n    \"\"\"\n    return Config() \nprint('DONE')","metadata":{"execution":{"iopub.status.busy":"2025-07-21T07:58:27.913119Z","iopub.execute_input":"2025-07-21T07:58:27.913425Z","iopub.status.idle":"2025-07-21T07:58:27.938639Z","shell.execute_reply.started":"2025-07-21T07:58:27.913395Z","shell.execute_reply":"2025-07-21T07:58:27.937883Z"},"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"\"\"\"\nFile chứa các hàm tiện ích cho bài toán phân đoạn polyp\nMục đích: Cung cấp các hàm hỗ trợ cho visualization, RLE encoding, post-processing\n\"\"\"\n\nimport os\nimport numpy as np\nimport torch\nimport cv2\nimport matplotlib.pyplot as plt\nfrom PIL import Image\nimport pandas as pd\nfrom typing import List, Tuple, Dict, Any, Optional\nimport logging\nimport random\nfrom pathlib import Path\n\n# Setup logging\nlogging.basicConfig(level=logging.INFO)\nlogger = logging.getLogger(__name__)\n\ndef set_seed(seed: int = 42):\n    \"\"\"\n    Thiết lập random seed cho reproducibility\n    \n    Args:\n        seed: Giá trị seed\n    \"\"\"\n    random.seed(seed)\n    np.random.seed(seed)\n    torch.manual_seed(seed)\n    torch.cuda.manual_seed(seed)\n    torch.cuda.manual_seed_all(seed)\n    torch.backends.cudnn.deterministic = True\n    torch.backends.cudnn.benchmark = False\n    \n    logger.info(f\"Đã thiết lập seed: {seed}\")\n\ndef create_directories(paths: List[str]):\n    \"\"\"\n    Tạo các thư mục cần thiết\n    \n    Args:\n        paths: Danh sách đường dẫn thư mục cần tạo\n    \"\"\"\n    for path in paths:\n        Path(path).mkdir(parents=True, exist_ok=True)\n        logger.info(f\"Đã tạo thư mục: {path}\")\n\ndef rle_encode(mask: np.ndarray) -> str:\n    \"\"\"\n    Mã hóa mask thành Run Length Encoding (RLE)\n    \n    Args:\n        mask: Binary mask array\n        \n    Returns:\n        RLE encoded string\n    \"\"\"\n    pixels = mask.flatten()\n    pixels = np.concatenate([[0], pixels, [0]])\n    runs = np.where(pixels[1:] != pixels[:-1])[0] + 1\n    runs[1::2] -= runs[::2]\n    return ' '.join(str(x) for x in runs)\n\ndef rle_decode(rle_string: str, shape: Tuple[int, int]) -> np.ndarray:\n    \"\"\"\n    Giải mã RLE string thành mask\n    \n    Args:\n        rle_string: RLE encoded string\n        shape: Kích thước mask (height, width)\n        \n    Returns:\n        Decoded binary mask\n    \"\"\"\n    if pd.isna(rle_string) or rle_string == '':\n        return np.zeros(shape, dtype=np.uint8)\n    \n    s = rle_string.split()\n    starts, lengths = [np.asarray(x, dtype=int) for x in (s[0:][::2], s[1:][::2])]\n    starts -= 1\n    ends = starts + lengths\n    \n    mask = np.zeros(shape[0] * shape[1], dtype=np.uint8)\n    for start, end in zip(starts, ends):\n        mask[start:end] = 1\n    \n    return mask.reshape(shape)\n\ndef mask_to_rle(mask: np.ndarray) -> str:\n    \"\"\"\n    Chuyển đổi mask thành RLE format cho submission\n    \n    Args:\n        mask: Segmentation mask [H, W] hoặc [H, W, C]\n        \n    Returns:\n        RLE encoded string\n    \"\"\"\n    if len(mask.shape) == 3:\n        # Multi-class mask\n        rle_strings = []\n        for class_idx in range(mask.shape[2]):\n            class_mask = mask[:, :, class_idx]\n            rle = rle_encode(class_mask)\n            rle_strings.append(rle)\n        return rle_strings\n    else:\n        # Single class mask\n        return rle_encode(mask)\n\ndef visualize_sample(\n    image: np.ndarray, \n    mask: np.ndarray, \n    prediction: Optional[np.ndarray] = None,\n    class_names: List[str] = None,\n    save_path: Optional[str] = None\n):\n    \"\"\"\n    Visualize ảnh, mask và prediction\n    \n    Args:\n        image: Input image [H, W, C]\n        mask: Ground truth mask [H, W]\n        prediction: Predicted mask [H, W] (optional)\n        class_names: Tên các classes\n        save_path: Đường dẫn lưu ảnh\n    \"\"\"\n    if class_names is None:\n        class_names = ['neoplastic', 'non-neoplastic', 'background']\n    \n    # Tạo colormap cho visualization\n    colors = np.array([\n        [255, 0, 0],      # Red cho neoplastic\n        [0, 255, 0],      # Green cho non-neoplastic\n        [0, 0, 255]       # Blue cho background\n    ])\n    \n    # Số subplot\n    n_plots = 3 if prediction is not None else 2\n    \n    fig, axes = plt.subplots(1, n_plots, figsize=(5 * n_plots, 5))\n    if n_plots == 1:\n        axes = [axes]\n    \n    # Plot original image\n    axes[0].imshow(image)\n    axes[0].set_title('Ảnh gốc')\n    axes[0].axis('off')\n    \n    # Plot ground truth mask\n    mask_colored = colors[mask]\n    axes[1].imshow(mask_colored.astype(np.uint8))\n    axes[1].set_title('Ground Truth')\n    axes[1].axis('off')\n    \n    # Plot prediction if available\n    if prediction is not None:\n        pred_colored = colors[prediction]\n        axes[2].imshow(pred_colored.astype(np.uint8))\n        axes[2].set_title('Prediction')\n        axes[2].axis('off')\n    \n    # Thêm legend\n    legend_elements = [plt.Rectangle((0,0),1,1, facecolor=colors[i]/255, label=class_names[i]) \n                      for i in range(len(class_names))]\n    fig.legend(handles=legend_elements, loc='upper right')\n    \n    plt.tight_layout()\n    \n    if save_path:\n        plt.savefig(save_path, dpi=300, bbox_inches='tight')\n        logger.info(f\"Đã lưu visualization: {save_path}\")\n    \n    plt.show()\n\ndef visualize_batch(\n    images: torch.Tensor,\n    masks: torch.Tensor,\n    predictions: Optional[torch.Tensor] = None,\n    num_samples: int = 4,\n    class_names: List[str] = None,\n    save_path: Optional[str] = None\n):\n    \"\"\"\n    Visualize một batch samples\n    \n    Args:\n        images: Batch images [B, C, H, W]\n        masks: Batch masks [B, H, W]\n        predictions: Batch predictions [B, H, W] (optional)\n        num_samples: Số samples để visualize\n        class_names: Tên các classes\n        save_path: Đường dẫn lưu ảnh\n    \"\"\"\n    if class_names is None:\n        class_names = ['neoplastic', 'non-neoplastic', 'background']\n    \n    # Convert to numpy\n    images = images.cpu().numpy()\n    masks = masks.cpu().numpy()\n    if predictions is not None:\n        predictions = predictions.cpu().numpy()\n    \n    # Normalize images to [0, 1]\n    images = (images - images.min()) / (images.max() - images.min())\n    \n    # Số samples thực tế\n    actual_samples = min(num_samples, images.shape[0])\n    \n    # Số columns\n    n_cols = 3 if predictions is not None else 2\n    \n    fig, axes = plt.subplots(actual_samples, n_cols, figsize=(5 * n_cols, 5 * actual_samples))\n    if actual_samples == 1:\n        axes = axes.reshape(1, -1)\n    \n    colors = np.array([\n        [255, 0, 0],      # Red\n        [0, 255, 0],      # Green  \n        [0, 0, 255]       # Blue\n    ])\n    \n    for i in range(actual_samples):\n        # Original image\n        img = np.transpose(images[i], (1, 2, 0))\n        axes[i, 0].imshow(img)\n        axes[i, 0].set_title(f'Sample {i+1} - Ảnh gốc')\n        axes[i, 0].axis('off')\n        \n        # Ground truth\n        mask_colored = colors[masks[i]]\n        axes[i, 1].imshow(mask_colored.astype(np.uint8))\n        axes[i, 1].set_title(f'Sample {i+1} - Ground Truth')\n        axes[i, 1].axis('off')\n        \n        # Prediction\n        if predictions is not None:\n            pred_colored = colors[predictions[i]]\n            axes[i, 2].imshow(pred_colored.astype(np.uint8))\n            axes[i, 2].set_title(f'Sample {i+1} - Prediction')\n            axes[i, 2].axis('off')\n    \n    # Legend\n    legend_elements = [plt.Rectangle((0,0),1,1, facecolor=colors[i]/255, label=class_names[i]) \n                      for i in range(len(class_names))]\n    fig.legend(handles=legend_elements, loc='upper right')\n    \n    plt.tight_layout()\n    \n    if save_path:\n        plt.savefig(save_path, dpi=300, bbox_inches='tight')\n        logger.info(f\"Đã lưu batch visualization: {save_path}\")\n    \n    plt.show()\n\ndef plot_training_history(\n    train_losses: List[float],\n    val_losses: List[float],\n    train_metrics: Dict[str, List[float]],\n    val_metrics: Dict[str, List[float]],\n    save_path: Optional[str] = None\n):\n    \"\"\"\n    Vẽ biểu đồ training history\n    \n    Args:\n        train_losses: Training losses\n        val_losses: Validation losses\n        train_metrics: Training metrics\n        val_metrics: Validation metrics\n        save_path: Đường dẫn lưu ảnh\n    \"\"\"\n    n_metrics = len(train_metrics) + 1  # +1 cho loss\n    \n    fig, axes = plt.subplots(1, n_metrics, figsize=(6 * n_metrics, 5))\n    if n_metrics == 1:\n        axes = [axes]\n    \n    # Plot loss\n    epochs = range(1, len(train_losses) + 1)\n    axes[0].plot(epochs, train_losses, 'b-', label='Training Loss')\n    axes[0].plot(epochs, val_losses, 'r-', label='Validation Loss')\n    axes[0].set_title('Training và Validation Loss')\n    axes[0].set_xlabel('Epoch')\n    axes[0].set_ylabel('Loss')\n    axes[0].legend()\n    axes[0].grid(True)\n    \n    # Plot metrics\n    for idx, (metric_name, train_values) in enumerate(train_metrics.items()):\n        val_values = val_metrics.get(metric_name, [])\n        \n        axes[idx + 1].plot(epochs, train_values, 'b-', label=f'Training {metric_name}')\n        if val_values:\n            axes[idx + 1].plot(epochs, val_values, 'r-', label=f'Validation {metric_name}')\n        \n        axes[idx + 1].set_title(f'Training và Validation {metric_name}')\n        axes[idx + 1].set_xlabel('Epoch')\n        axes[idx + 1].set_ylabel(metric_name)\n        axes[idx + 1].legend()\n        axes[idx + 1].grid(True)\n    \n    plt.tight_layout()\n    \n    if save_path:\n        plt.savefig(save_path, dpi=300, bbox_inches='tight')\n        logger.info(f\"Đã lưu training history: {save_path}\")\n    \n    plt.show()\n\ndef apply_morphological_operations(mask: np.ndarray, kernel_size: int = 3) -> np.ndarray:\n    \"\"\"\n    Áp dụng morphological operations để làm sạch mask\n    \n    Args:\n        mask: Binary mask\n        kernel_size: Kích thước kernel\n        \n    Returns:\n        Cleaned mask\n    \"\"\"\n    kernel = cv2.getStructuringElement(cv2.MORPH_ELLIPSE, (kernel_size, kernel_size))\n    \n    # Opening: loại bỏ noise\n    mask = cv2.morphologyEx(mask, cv2.MORPH_OPEN, kernel)\n    \n    # Closing: fill holes\n    mask = cv2.morphologyEx(mask, cv2.MORPH_CLOSE, kernel)\n    \n    return mask\n\ndef remove_small_objects(mask: np.ndarray, min_size: int = 100) -> np.ndarray:\n    \"\"\"\n    Loại bỏ các object nhỏ trong mask\n    \n    Args:\n        mask: Binary mask\n        min_size: Kích thước tối thiểu (pixels)\n        \n    Returns:\n        Cleaned mask\n    \"\"\"\n    # Find connected components\n    num_labels, labels = cv2.connectedComponents(mask.astype(np.uint8))\n    \n    # Create output mask\n    output_mask = np.zeros_like(mask)\n    \n    # Keep only large components\n    for label in range(1, num_labels):\n        component_mask = (labels == label)\n        if np.sum(component_mask) >= min_size:\n            output_mask[component_mask] = 1\n    \n    return output_mask\n\ndef post_process_mask(mask: np.ndarray, min_size: int = 100, kernel_size: int = 3) -> np.ndarray:\n    \"\"\"\n    Post-processing cho segmentation mask\n    \n    Args:\n        mask: Raw segmentation mask\n        min_size: Kích thước tối thiểu object\n        kernel_size: Kích thước kernel cho morphological ops\n        \n    Returns:\n        Post-processed mask\n    \"\"\"\n    # Apply morphological operations\n    mask = apply_morphological_operations(mask, kernel_size)\n    \n    # Remove small objects\n    mask = remove_small_objects(mask, min_size)\n    \n    return mask\n\ndef create_submission_file(\n    predictions: List[np.ndarray],\n    image_ids: List[str],\n    output_path: str = \"submission.csv\"\n):\n    \"\"\"\n    Tạo file submission cho Kaggle\n    \n    Args:\n        predictions: List các predicted masks\n        image_ids: List các image IDs\n        output_path: Đường dẫn file output\n    \"\"\"\n    submission_data = []\n    \n    for pred_mask, image_id in zip(predictions, image_ids):\n        # Giả sử mask có 3 classes (neoplastic, non-neoplastic)\n        # Chỉ submit 2 classes đầu (background không cần)\n        \n        for class_idx in range(2):  # 0: neoplastic, 1: non-neoplastic\n            class_mask = (pred_mask == class_idx).astype(np.uint8)\n            \n            # Apply post-processing\n            class_mask = post_process_mask(class_mask)\n            \n            # Convert to RLE\n            rle = rle_encode(class_mask)\n            \n            submission_data.append({\n                'Id': f\"{image_id}_{class_idx}\",\n                'Expected': rle\n            })\n    \n    # Tạo DataFrame và lưu\n    df = pd.DataFrame(submission_data)\n    df.to_csv(output_path, index=False)\n    \n    logger.info(f\"Đã tạo submission file: {output_path}\")\n    logger.info(f\"Số lượng predictions: {len(submission_data)}\")\n\ndef calculate_class_weights(masks: List[np.ndarray], num_classes: int = 3) -> np.ndarray:\n    \"\"\"\n    Tính class weights để xử lý class imbalance\n    \n    Args:\n        masks: List các ground truth masks\n        num_classes: Số lượng classes\n        \n    Returns:\n        Class weights array\n    \"\"\"\n    class_counts = np.zeros(num_classes)\n    \n    for mask in masks:\n        for class_idx in range(num_classes):\n            class_counts[class_idx] += np.sum(mask == class_idx)\n    \n    # Tính weights (inverse frequency)\n    total_pixels = np.sum(class_counts)\n    class_weights = total_pixels / (num_classes * class_counts)\n    \n    # Normalize weights\n    class_weights = class_weights / np.sum(class_weights) * num_classes\n    \n    logger.info(f\"Class counts: {class_counts}\")\n    logger.info(f\"Class weights: {class_weights}\")\n    \n    return class_weights.astype(np.float32)\n\ndef save_model_checkpoint(\n    model: torch.nn.Module,\n    optimizer: torch.optim.Optimizer,\n    epoch: int,\n    loss: float,\n    metrics: Dict[str, float],\n    save_path: str\n):\n    \"\"\"\n    Lưu model checkpoint\n    \n    Args:\n        model: PyTorch model\n        optimizer: Optimizer\n        epoch: Epoch hiện tại\n        loss: Loss value\n        metrics: Dict các metrics\n        save_path: Đường dẫn lưu checkpoint\n    \"\"\"\n    checkpoint = {\n        'epoch': epoch,\n        'model_state_dict': model.state_dict(),\n        'optimizer_state_dict': optimizer.state_dict(),\n        'loss': loss,\n        'metrics': metrics\n    }\n    \n    torch.save(checkpoint, save_path)\n    logger.info(f\"Đã lưu checkpoint: {save_path}\")\n\ndef load_model_checkpoint(\n    model: torch.nn.Module,\n    optimizer: torch.optim.Optimizer,\n    checkpoint_path: str\n) -> Tuple[int, float, Dict[str, float]]:\n    \"\"\"\n    Load model checkpoint\n    \n    Args:\n        model: PyTorch model\n        optimizer: Optimizer\n        checkpoint_path: Đường dẫn checkpoint\n        \n    Returns:\n        Tuple (epoch, loss, metrics)\n    \"\"\"\n    checkpoint = torch.load(checkpoint_path)\n    \n    model.load_state_dict(checkpoint['model_state_dict'])\n    optimizer.load_state_dict(checkpoint['optimizer_state_dict'])\n    \n    epoch = checkpoint['epoch']\n    loss = checkpoint['loss']\n    metrics = checkpoint['metrics']\n    \n    logger.info(f\"Đã load checkpoint từ epoch {epoch}\")\n    \n    return epoch, loss, metrics\n\ndef get_device() -> torch.device:\n    \"\"\"\n    Lấy device phù hợp (GPU hoặc CPU)\n    \n    Returns:\n        PyTorch device\n    \"\"\"\n    if torch.cuda.is_available():\n        device = torch.device('cuda')\n        logger.info(f\"Sử dụng GPU: {torch.cuda.get_device_name()}\")\n    else:\n        device = torch.device('cpu')\n        logger.info(\"Sử dụng CPU\")\n    \n    return device\n\ndef print_model_summary(model: torch.nn.Module, input_size: Tuple[int, int, int, int]):\n    \"\"\"\n    In thông tin tóm tắt về model\n    \n    Args:\n        model: PyTorch model\n        input_size: Kích thước input (B, C, H, W)\n    \"\"\"\n    total_params = sum(p.numel() for p in model.parameters())\n    trainable_params = sum(p.numel() for p in model.parameters() if p.requires_grad)\n    \n    print(\"=\" * 50)\n    print(\"MODEL SUMMARY\")\n    print(\"=\" * 50)\n    print(f\"Model: {model.__class__.__name__}\")\n    print(f\"Input size: {input_size}\")\n    print(f\"Total parameters: {total_params:,}\")\n    print(f\"Trainable parameters: {trainable_params:,}\")\n    print(f\"Non-trainable parameters: {total_params - trainable_params:,}\")\n    \n    # Tính model size\n    param_size = 0\n    for param in model.parameters():\n        param_size += param.nelement() * param.element_size()\n    \n    buffer_size = 0\n    for buffer in model.buffers():\n        buffer_size += buffer.nelement() * buffer.element_size()\n    \n    size_mb = (param_size + buffer_size) / 1024 / 1024\n    print(f\"Model size: {size_mb:.2f} MB\")\n    print(\"=\" * 50) \nprint('DONE')","metadata":{"execution":{"iopub.status.busy":"2025-07-21T07:58:32.832228Z","iopub.execute_input":"2025-07-21T07:58:32.832826Z","iopub.status.idle":"2025-07-21T07:58:42.324062Z","shell.execute_reply.started":"2025-07-21T07:58:32.832793Z","shell.execute_reply":"2025-07-21T07:58:42.323246Z"},"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"\"\"\"\nFile xử lý dữ liệu cho bài toán phân đoạn polyp\nMục đích: Tạo Dataset class, augmentation và data loading\n\"\"\"\n\nimport os\nimport numpy as np\nimport torch\nfrom torch.utils.data import Dataset, DataLoader, random_split\nfrom torchvision.transforms import PILToTensor, ToPILImage\nfrom PIL import Image\nimport albumentations as A\nfrom albumentations.pytorch import ToTensorV2\nfrom transformers import SegformerImageProcessor\nfrom typing import List, Tuple, Optional, Dict, Any\nimport cv2\nimport logging\n\n# Setup logging\nlogging.basicConfig(level=logging.INFO)\nlogger = logging.getLogger(__name__)\n\nclass PolypDataset(Dataset):\n    \"\"\"\n    Dataset class cho bài toán phân đoạn polyp\n    \n    Chức năng:\n    - Load ảnh và mask từ đường dẫn\n    - Áp dụng augmentation\n    - Xử lý mask thành format phù hợp cho model\n    - Trả về data đã được preprocessing\n    \"\"\"\n    \n    def __init__(\n        self, \n        image_paths: List[str], \n        mask_paths: List[str], \n        processor: SegformerImageProcessor,\n        augmentation: Optional[A.Compose] = None,\n        is_training: bool = True\n    ):\n        \"\"\"\n        Khởi tạo Dataset\n        \n        Args:\n            image_paths: Danh sách đường dẫn tới ảnh\n            mask_paths: Danh sách đường dẫn tới mask\n            processor: SegformerImageProcessor để xử lý ảnh\n            augmentation: Augmentation pipeline\n            is_training: Có phải training mode không\n        \"\"\"\n        self.image_paths = image_paths\n        self.mask_paths = mask_paths\n        self.processor = processor\n        self.augmentation = augmentation\n        self.is_training = is_training\n        \n        # Kiểm tra số lượng ảnh và mask phải bằng nhau\n        assert len(image_paths) == len(mask_paths), \\\n            f\"Số lượng ảnh ({len(image_paths)}) và mask ({len(mask_paths)}) không khớp\"\n        \n        logger.info(f\"Đã tạo dataset với {len(self.image_paths)} samples\")\n    \n    def __len__(self) -> int:\n        \"\"\"\n        Trả về số lượng samples trong dataset\n        \"\"\"\n        return len(self.image_paths)\n    \n    def __getitem__(self, idx: int) -> Dict[str, torch.Tensor]:\n        \"\"\"\n        Lấy một sample từ dataset\n        \n        Args:\n            idx: Index của sample\n            \n        Returns:\n            Dict chứa pixel_values và labels đã được xử lý\n        \"\"\"\n        try:\n            # Lấy đường dẫn ảnh và mask\n            image_path = self.image_paths[idx]\n            mask_path = self.mask_paths[idx]\n            \n            # Kiểm tra file tồn tại\n            if not os.path.exists(image_path):\n                raise FileNotFoundError(f\"Không tìm thấy ảnh: {image_path}\")\n            if not os.path.exists(mask_path):\n                raise FileNotFoundError(f\"Không tìm thấy mask: {mask_path}\")\n            \n            # Load ảnh và mask\n            image = Image.open(image_path).convert('RGB')\n            mask = Image.open(mask_path).convert('RGB')\n            \n            # Convert sang numpy để áp dụng augmentation\n            image_np = np.array(image)\n            mask_np = np.array(mask)\n            \n            # Áp dụng augmentation nếu có\n            if self.augmentation is not None and self.is_training:\n                augmented = self.augmentation(image=image_np, mask=mask_np)\n                image_np = augmented['image']\n                mask_np = augmented['mask']\n            \n            # Convert lại thành PIL Image\n            image = Image.fromarray(image_np)\n            mask = Image.fromarray(mask_np)\n            \n            # Xử lý mask\n            mask_processed = self._process_mask(mask)\n            \n            # Sử dụng processor để chuẩn bị input cho model\n            encoded_inputs = self.processor(\n                image, \n                mask_processed, \n                return_tensors=\"pt\"\n            )\n            \n            # Loại bỏ batch dimension\n            for key, value in encoded_inputs.items():\n                encoded_inputs[key] = value.squeeze(0)\n            \n            return encoded_inputs\n            \n        except Exception as e:\n            logger.error(f\"Lỗi khi load sample {idx}: {str(e)}\")\n            raise e\n    \n    def _process_mask(self, mask: Image.Image) -> Image.Image:\n        \"\"\"\n        Xử lý mask thành format phù hợp\n        \n        Args:\n            mask: Mask PIL Image\n            \n        Returns:\n            Mask đã được xử lý\n        \"\"\"\n        # Convert sang tensor\n        mask_tensor = PILToTensor()(mask).float() / 255.0\n        \n        # Áp dụng threshold\n        mask_tensor = torch.where(mask_tensor > 0.65, 1.0, 0.0)\n        \n        # Xử lý background channel\n        if mask_tensor.shape[0] >= 3:\n            mask_tensor[2, :, :] = 0.0001\n        \n        # Lấy class có confidence cao nhất\n        mask_class = torch.argmax(mask_tensor, dim=0).numpy().astype(np.uint8)\n        \n        # Convert lại thành PIL Image\n        return Image.fromarray(mask_class)\n\ndef get_augmentation_pipeline(is_training: bool = True) -> Optional[A.Compose]:\n    \"\"\"\n    Tạo augmentation pipeline cho training/validation\n    \n    Args:\n        is_training: Có phải training mode không\n        \n    Returns:\n        Augmentation pipeline hoặc None nếu không cần augmentation\n    \"\"\"\n    if not is_training:\n        return None\n    \n    # Augmentation pipeline cho training\n    train_transform = A.Compose([\n        # Geometric transformations\n        A.RandomRotate90(p=0.5),\n        A.HorizontalFlip(p=0.5),\n        A.VerticalFlip(p=0.5),\n        A.ShiftScaleRotate(\n            shift_limit=0.1,\n            scale_limit=0.1,\n            rotate_limit=15,\n            p=0.5\n        ),\n        \n        # Color transformations\n        A.RandomBrightnessContrast(\n            brightness_limit=0.2,\n            contrast_limit=0.2,\n            p=0.3\n        ),\n        A.RandomGamma(gamma_limit=(70, 130), p=0.2),\n        A.RGBShift(\n            r_shift_limit=10,\n            g_shift_limit=10,\n            b_shift_limit=10,\n            p=0.3\n        ),\n        \n        # Noise and blur\n        A.GaussNoise(var_limit=(10.0, 50.0), p=0.2),\n        A.GaussianBlur(blur_limit=3, p=0.1),\n        \n        # Elastic transformation\n        A.ElasticTransform(\n            alpha=1,\n            sigma=50,\n            alpha_affine=50,\n            p=0.2\n        ),\n    ])\n    \n    return train_transform\n\ndef create_data_splits(\n    image_paths: List[str],\n    mask_paths: List[str],\n    train_split: float = 0.8,\n    val_split: float = 0.1,\n    seed: int = 42\n) -> Tuple[List[str], List[str], List[str], List[str], List[str], List[str]]:\n    \"\"\"\n    Chia dữ liệu thành train/validation/test\n    \n    Args:\n        image_paths: Danh sách đường dẫn ảnh\n        mask_paths: Danh sách đường dẫn mask\n        train_split: Tỷ lệ train\n        val_split: Tỷ lệ validation\n        seed: Random seed\n        \n    Returns:\n        Tuple chứa (train_images, train_masks, val_images, val_masks, test_images, test_masks)\n    \"\"\"\n    # Set random seed\n    np.random.seed(seed)\n    torch.manual_seed(seed)\n    \n    # Tạo indices và shuffle\n    indices = np.arange(len(image_paths))\n    np.random.shuffle(indices)\n    \n    # Tính toán số lượng samples cho mỗi split\n    n_total = len(image_paths)\n    n_train = int(train_split * n_total)\n    n_val = int(val_split * n_total)\n    \n    # Chia indices\n    train_indices = indices[:n_train]\n    val_indices = indices[n_train:n_train + n_val]\n    test_indices = indices[n_train + n_val:]\n    \n    # Tạo lists cho từng split\n    train_images = [image_paths[i] for i in train_indices]\n    train_masks = [mask_paths[i] for i in train_indices]\n    \n    val_images = [image_paths[i] for i in val_indices]\n    val_masks = [mask_paths[i] for i in val_indices]\n    \n    test_images = [image_paths[i] for i in test_indices]\n    test_masks = [mask_paths[i] for i in test_indices]\n    \n    logger.info(f\"Chia dữ liệu: Train={len(train_images)}, Val={len(val_images)}, Test={len(test_images)}\")\n    \n    return train_images, train_masks, val_images, val_masks, test_images, test_masks\n\ndef create_dataloaders(\n    train_images: List[str],\n    train_masks: List[str],\n    val_images: List[str],\n    val_masks: List[str],\n    processor: SegformerImageProcessor,\n    batch_size: int = 8,\n    num_workers: int = 4,\n    use_augmentation: bool = True\n) -> Tuple[DataLoader, DataLoader]:\n    \"\"\"\n    Tạo DataLoader cho training và validation\n    \n    Args:\n        train_images: Danh sách ảnh training\n        train_masks: Danh sách mask training\n        val_images: Danh sách ảnh validation\n        val_masks: Danh sách mask validation\n        processor: SegformerImageProcessor\n        batch_size: Kích thước batch\n        num_workers: Số worker cho DataLoader\n        use_augmentation: Có sử dụng augmentation không\n        \n    Returns:\n        Tuple chứa (train_dataloader, val_dataloader)\n    \"\"\"\n    # Tạo augmentation\n    train_augmentation = get_augmentation_pipeline(is_training=True) if use_augmentation else None\n    val_augmentation = get_augmentation_pipeline(is_training=False)\n    \n    # Tạo datasets\n    train_dataset = PolypDataset(\n        image_paths=train_images,\n        mask_paths=train_masks,\n        processor=processor,\n        augmentation=train_augmentation,\n        is_training=True\n    )\n    \n    val_dataset = PolypDataset(\n        image_paths=val_images,\n        mask_paths=val_masks,\n        processor=processor,\n        augmentation=val_augmentation,\n        is_training=False\n    )\n    \n    # Tạo dataloaders\n    train_dataloader = DataLoader(\n        train_dataset,\n        batch_size=batch_size,\n        shuffle=True,\n        num_workers=num_workers,\n        pin_memory=True,\n        drop_last=True\n    )\n    \n    val_dataloader = DataLoader(\n        val_dataset,\n        batch_size=batch_size,\n        shuffle=False,\n        num_workers=num_workers,\n        pin_memory=True,\n        drop_last=False\n    )\n    \n    logger.info(f\"Tạo DataLoader: Train batches={len(train_dataloader)}, Val batches={len(val_dataloader)}\")\n    \n    return train_dataloader, val_dataloader\n\ndef get_file_paths(image_dir: str, mask_dir: str) -> Tuple[List[str], List[str]]:\n    \"\"\"\n    Lấy danh sách đường dẫn file ảnh và mask\n    \n    Args:\n        image_dir: Thư mục chứa ảnh\n        mask_dir: Thư mục chứa mask\n        \n    Returns:\n        Tuple chứa (image_paths, mask_paths)\n    \"\"\"\n    # Lấy danh sách file\n    image_files = sorted([f for f in os.listdir(image_dir) if f.lower().endswith(('.png', '.jpg', '.jpeg'))])\n    mask_files = sorted([f for f in os.listdir(mask_dir) if f.lower().endswith(('.png', '.jpg', '.jpeg'))])\n    \n    # Tạo đường dẫn đầy đủ\n    image_paths = [os.path.join(image_dir, f) for f in image_files]\n    mask_paths = [os.path.join(mask_dir, f) for f in mask_files]\n    \n    logger.info(f\"Tìm thấy {len(image_paths)} ảnh và {len(mask_paths)} mask\")\n    \n    return image_paths, mask_paths\n\ndef analyze_dataset(image_paths: List[str], mask_paths: List[str]) -> Dict[str, Any]:\n    \"\"\"\n    Phân tích dataset để hiểu rõ hơn về dữ liệu\n    \n    Args:\n        image_paths: Danh sách đường dẫn ảnh\n        mask_paths: Danh sách đường dẫn mask\n        \n    Returns:\n        Dict chứa thông tin phân tích\n    \"\"\"\n    analysis = {\n        'total_samples': len(image_paths),\n        'image_sizes': [],\n        'class_distribution': {'neoplastic': 0, 'non_neoplastic': 0, 'background': 0}\n    }\n    \n    logger.info(\"Đang phân tích dataset...\")\n    \n    for i, (img_path, mask_path) in enumerate(zip(image_paths[:100], mask_paths[:100])):  # Chỉ phân tích 100 sample đầu\n        try:\n            # Phân tích kích thước ảnh\n            img = Image.open(img_path)\n            analysis['image_sizes'].append(img.size)\n            \n            # Phân tích phân bố class trong mask\n            mask = Image.open(mask_path)\n            mask_array = np.array(mask)\n            \n            # Đếm pixel theo class (giả sử có 3 channel cho 3 class)\n            if len(mask_array.shape) == 3:\n                for class_idx in range(min(3, mask_array.shape[2])):\n                    class_pixels = np.sum(mask_array[:, :, class_idx] > 127)\n                    if class_idx == 0:\n                        analysis['class_distribution']['neoplastic'] += class_pixels\n                    elif class_idx == 1:\n                        analysis['class_distribution']['non_neoplastic'] += class_pixels\n                    else:\n                        analysis['class_distribution']['background'] += class_pixels\n                        \n        except Exception as e:\n            logger.warning(f\"Lỗi khi phân tích sample {i}: {str(e)}\")\n            continue\n    \n    # Tính toán thống kê\n    if analysis['image_sizes']:\n        widths, heights = zip(*analysis['image_sizes'])\n        analysis['avg_width'] = np.mean(widths)\n        analysis['avg_height'] = np.mean(heights)\n        analysis['min_size'] = (min(widths), min(heights))\n        analysis['max_size'] = (max(widths), max(heights))\n    \n    logger.info(f\"Phân tích hoàn thành: {analysis}\")\n    \n    return analysis \nprint('DONE')","metadata":{"execution":{"iopub.status.busy":"2025-07-21T07:58:47.683329Z","iopub.execute_input":"2025-07-21T07:58:47.683732Z","iopub.status.idle":"2025-07-21T07:59:17.189642Z","shell.execute_reply.started":"2025-07-21T07:58:47.683707Z","shell.execute_reply":"2025-07-21T07:59:17.188892Z"},"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"\"\"\"\nFile chứa các mô hình cho bài toán phân đoạn polyp\nMục đích: Định nghĩa các kiến trúc mô hình khác nhau (Segformer, U-Net, DeepLab)\n\"\"\"\n\nimport torch\nimport torch.nn as nn\nimport torch.nn.functional as F\nfrom typing import Dict, Any, Optional, List\nimport logging\n\n# Setup logging\nlogging.basicConfig(level=logging.INFO)\nlogger = logging.getLogger(__name__)\n\ntry:\n    from transformers import SegformerForSemanticSegmentation, SegformerImageProcessor\n    TRANSFORMERS_AVAILABLE = True\nexcept ImportError:\n    TRANSFORMERS_AVAILABLE = False\n    logger.warning(\"Transformers không có sẵn, chỉ có thể sử dụng các mô hình custom\")\nclass DoubleConv(nn.Module):\n    \"\"\"Double Convolution block\"\"\"\n    def __init__(self, in_channels: int, out_channels: int):\n        super().__init__()\n        self.double_conv = nn.Sequential(\n            nn.Conv2d(in_channels, out_channels, kernel_size=3, padding=1),\n            nn.BatchNorm2d(out_channels),\n            nn.ReLU(inplace=True),\n            nn.Conv2d(out_channels, out_channels, kernel_size=3, padding=1),\n            nn.BatchNorm2d(out_channels),\n            nn.ReLU(inplace=True)\n        )\n    \n    def forward(self, x: torch.Tensor) -> torch.Tensor:\n        return self.double_conv(x)\n\nclass SegformerModel(nn.Module):\n    \"\"\"\n    Wrapper cho Segformer model từ Hugging Face\n    \n    Chức năng:\n    - Load pretrained Segformer model\n    - Tùy chỉnh số lượng classes\n    - Hỗ trợ fine-tuning\n    \"\"\"\n    \n    def __init__(\n        self,\n        model_name: str = \"nvidia/mit-b5\",\n        num_classes: int = 3,\n        id2label: Optional[Dict[int, str]] = None,\n        label2id: Optional[Dict[str, int]] = None,\n        pretrained: bool = True\n    ):\n        \"\"\"\n        Khởi tạo Segformer model\n        \n        Args:\n            model_name: Tên model từ Hugging Face\n            num_classes: Số lượng classes\n            id2label: Mapping từ id sang label\n            label2id: Mapping từ label sang id\n            pretrained: Có sử dụng pretrained weights không\n        \"\"\"\n        super().__init__()\n        \n        if not TRANSFORMERS_AVAILABLE:\n            raise ImportError(\"Transformers library không có sẵn\")\n        \n        self.num_classes = num_classes\n        self.model_name = model_name\n        \n        # Tạo default labels nếu không có\n        if id2label is None:\n            id2label = {i: f\"class_{i}\" for i in range(num_classes)}\n        if label2id is None:\n            label2id = {v: k for k, v in id2label.items()}\n        \n        # Load model\n        if pretrained:\n            self.model = SegformerForSemanticSegmentation.from_pretrained(\n                model_name,\n                num_labels=num_classes,\n                id2label=id2label,\n                label2id=label2id,\n                ignore_mismatched_sizes=True\n            )\n        else:\n            from transformers import SegformerConfig\n            config = SegformerConfig.from_pretrained(model_name)\n            config.num_labels = num_classes\n            config.id2label = id2label\n            config.label2id = label2id\n            self.model = SegformerForSemanticSegmentation(config)\n        \n        logger.info(f\"Đã tạo Segformer model: {model_name} với {num_classes} classes\")\n    \n    def forward(self, pixel_values: torch.Tensor, labels: Optional[torch.Tensor] = None):\n        \"\"\"\n        Forward pass của model\n        \n        Args:\n            pixel_values: Input images tensor\n            labels: Ground truth labels (optional)\n            \n        Returns:\n            Output từ model\n        \"\"\"\n        return self.model(pixel_values=pixel_values, labels=labels)\n    \n    def predict(self, pixel_values: torch.Tensor) -> torch.Tensor:\n        \"\"\"\n        Predict segmentation map\n        \n        Args:\n            pixel_values: Input images tensor\n            \n        Returns:\n            Predicted segmentation maps\n        \"\"\"\n        with torch.no_grad():\n            outputs = self.model(pixel_values=pixel_values)\n            logits = outputs.logits\n            \n            # Upsample logits to original image size\n            upsampled_logits = F.interpolate(\n                logits,\n                size=pixel_values.shape[-2:],\n                mode=\"bilinear\",\n                align_corners=False\n            )\n            \n            # Get predicted classes\n            predicted = upsampled_logits.argmax(dim=1)\n            \n        return predicted\n\n\nclass ASPPModule(nn.Module):\n    \"\"\"\n    Atrous Spatial Pyramid Pooling module cho DeepLabV3\n    \n    Chức năng:\n    - Multi-scale feature extraction\n    - Dilated convolutions với các dilation rates khác nhau\n    \"\"\"\n    \n    def __init__(self, in_channels: int, out_channels: int, atrous_rates: List[int]):\n        \"\"\"\n        Khởi tạo ASPP module\n        \n        Args:\n            in_channels: Số input channels\n            out_channels: Số output channels\n            atrous_rates: List các dilation rates\n        \"\"\"\n        super().__init__()\n        \n        modules = []\n        \n        # 1x1 convolution\n        modules.append(nn.Sequential(\n            nn.Conv2d(in_channels, out_channels, 1, bias=False),\n            nn.BatchNorm2d(out_channels),\n            nn.ReLU(inplace=True)\n        ))\n        \n        # Atrous convolutions\n        for rate in atrous_rates:\n            modules.append(nn.Sequential(\n                nn.Conv2d(in_channels, out_channels, 3, padding=rate, dilation=rate, bias=False),\n                nn.BatchNorm2d(out_channels),\n                nn.ReLU(inplace=True)\n            ))\n        \n        # Global average pooling\n        modules.append(nn.Sequential(\n            nn.AdaptiveAvgPool2d(1),\n            nn.Conv2d(in_channels, out_channels, 1, bias=False),\n            nn.BatchNorm2d(out_channels),\n            nn.ReLU(inplace=True)\n        ))\n        \n        self.convs = nn.ModuleList(modules)\n        \n        # Final projection\n        self.project = nn.Sequential(\n            nn.Conv2d(len(self.convs) * out_channels, out_channels, 1, bias=False),\n            nn.BatchNorm2d(out_channels),\n            nn.ReLU(inplace=True),\n            nn.Dropout(0.5)\n        )\n    \n    def forward(self, x: torch.Tensor) -> torch.Tensor:\n        \"\"\"\n        Forward pass của ASPP\n        \n        Args:\n            x: Input tensor\n            \n        Returns:\n            Output tensor với multi-scale features\n        \"\"\"\n        res = []\n        \n        for conv in self.convs:\n            res.append(conv(x))\n        \n        # Upsample global pooling result\n        res[-1] = F.interpolate(res[-1], size=x.shape[2:], mode='bilinear', align_corners=False)\n        \n        # Concatenate all features\n        res = torch.cat(res, dim=1)\n        \n        return self.project(res)\n\nclass DeepLabV3(nn.Module):\n    \"\"\"\n    DeepLabV3 model cho semantic segmentation\n    \n    Chức năng:\n    - Sử dụng dilated convolutions\n    - ASPP module cho multi-scale features\n    - Backbone có thể là ResNet hoặc MobileNet\n    \"\"\"\n    \n    def __init__(self, num_classes: int = 3, backbone: str = \"resnet50\"):\n        \"\"\"\n        Khởi tạo DeepLabV3 model\n        \n        Args:\n            num_classes: Số lượng classes\n            backbone: Backbone network (resnet50, mobilenet)\n        \"\"\"\n        super().__init__()\n        \n        self.num_classes = num_classes\n        self.backbone = backbone\n        \n        # Simplified backbone (có thể thay thế bằng pretrained ResNet)\n        self.backbone_layers = nn.Sequential(\n            nn.Conv2d(3, 64, kernel_size=7, stride=2, padding=3, bias=False),\n            nn.BatchNorm2d(64),\n            nn.ReLU(inplace=True),\n            nn.MaxPool2d(kernel_size=3, stride=2, padding=1),\n            \n            # Residual blocks (simplified)\n            DoubleConv(64, 128),\n            nn.MaxPool2d(2),\n            DoubleConv(128, 256),\n            nn.MaxPool2d(2),\n            DoubleConv(256, 512),\n        )\n        \n        # ASPP module\n        self.aspp = ASPPModule(512, 256, [6, 12, 18])\n        \n        # Final classifier\n        self.classifier = nn.Sequential(\n            nn.Conv2d(256, 256, 3, padding=1, bias=False),\n            nn.BatchNorm2d(256),\n            nn.ReLU(inplace=True),\n            nn.Dropout(0.5),\n            nn.Conv2d(256, num_classes, 1)\n        )\n        \n        logger.info(f\"Đã tạo DeepLabV3 model với {num_classes} classes\")\n    \n    def forward(self, x: torch.Tensor) -> torch.Tensor:\n        \"\"\"\n        Forward pass của DeepLabV3\n        \n        Args:\n            x: Input tensor [B, C, H, W]\n            \n        Returns:\n            Output segmentation logits [B, num_classes, H, W]\n        \"\"\"\n        input_shape = x.shape[-2:]\n        \n        # Backbone\n        x = self.backbone_layers(x)\n        \n        # ASPP\n        x = self.aspp(x)\n        \n        # Classifier\n        x = self.classifier(x)\n        \n        # Upsample to original size\n        x = F.interpolate(x, size=input_shape, mode='bilinear', align_corners=False)\n        \n        return x\n\nclass FocalLoss(nn.Module):\n    \"\"\"\n    Focal Loss cho class imbalance\n    \n    Chức năng:\n    - Giảm weight của easy examples\n    - Tăng focus vào hard examples\n    - Hữu ích cho medical image segmentation\n    \"\"\"\n    \n    def __init__(self, alpha: float = 1.0, gamma: float = 2.0, reduction: str = 'mean'):\n        \"\"\"\n        Khởi tạo Focal Loss\n        \n        Args:\n            alpha: Weighting factor\n            gamma: Focusing parameter\n            reduction: Reduction method\n        \"\"\"\n        super().__init__()\n        self.alpha = alpha\n        self.gamma = gamma\n        self.reduction = reduction\n    \n    def forward(self, inputs: torch.Tensor, targets: torch.Tensor) -> torch.Tensor:\n        \"\"\"\n        Tính Focal Loss\n        \n        Args:\n            inputs: Predicted logits [B, C, H, W]\n            targets: Ground truth labels [B, H, W]\n            \n        Returns:\n            Focal loss value\n        \"\"\"\n        # Compute cross entropy\n        ce_loss = F.cross_entropy(inputs, targets, reduction='none')\n        \n        # Compute probabilities\n        pt = torch.exp(-ce_loss)\n        \n        # Compute focal loss\n        focal_loss = self.alpha * (1 - pt) ** self.gamma * ce_loss\n        \n        if self.reduction == 'mean':\n            return focal_loss.mean()\n        elif self.reduction == 'sum':\n            return focal_loss.sum()\n        else:\n            return focal_loss\n\ndef get_model(model_name: str, num_classes: int = 3, **kwargs) -> nn.Module:\n    \"\"\"\n    Factory function để tạo model theo tên\n    \n    Args:\n        model_name: Tên model (segformer, unet, deeplabv3)\n        num_classes: Số lượng classes\n        **kwargs: Các tham số khác\n        \n    Returns:\n        Model instance\n    \"\"\"\n    model_name = model_name.lower()\n    \n    if model_name == \"segformer\":\n        return SegformerModel(num_classes=num_classes, **kwargs)\n    elif model_name == \"deeplabv3\":\n        return DeepLabV3(num_classes=num_classes, **kwargs)\n    else:\n        raise ValueError(f\"Không hỗ trợ model: {model_name}\")\n\ndef count_parameters(model: nn.Module) -> int:\n    \"\"\"\n    Đếm số lượng parameters trong model\n    \n    Args:\n        model: PyTorch model\n        \n    Returns:\n        Số lượng trainable parameters\n    \"\"\"\n    return sum(p.numel() for p in model.parameters() if p.requires_grad)\n\ndef get_model_info(model: nn.Module) -> Dict[str, Any]:\n    \"\"\"\n    Lấy thông tin về model\n    \n    Args:\n        model: PyTorch model\n        \n    Returns:\n        Dict chứa thông tin model\n    \"\"\"\n    total_params = count_parameters(model)\n    \n    # Tính model size (MB)\n    param_size = 0\n    for param in model.parameters():\n        param_size += param.nelement() * param.element_size()\n    \n    buffer_size = 0\n    for buffer in model.buffers():\n        buffer_size += buffer.nelement() * buffer.element_size()\n    \n    size_mb = (param_size + buffer_size) / 1024 / 1024\n    \n    return {\n        'total_parameters': total_params,\n        'model_size_mb': size_mb,\n        'model_type': type(model).__name__\n    } \nprint('DONE')","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-07-21T07:59:20.427976Z","iopub.execute_input":"2025-07-21T07:59:20.428590Z","iopub.status.idle":"2025-07-21T07:59:23.295417Z","shell.execute_reply.started":"2025-07-21T07:59:20.428563Z","shell.execute_reply":"2025-07-21T07:59:23.294666Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"\"\"\"\nFile chứa các metrics đánh giá cho bài toán phân đoạn polyp\nMục đích: Tính toán các metrics như IoU, Dice, Accuracy, Precision, Recall\n\"\"\"\n\nimport numpy as np\nimport torch\nimport torch.nn.functional as F\nfrom typing import Dict, List, Tuple, Optional, Any\nimport logging\nfrom sklearn.metrics import confusion_matrix, classification_report\nimport matplotlib.pyplot as plt\nimport seaborn as sns\n\n# Setup logging\nlogging.basicConfig(level=logging.INFO)\nlogger = logging.getLogger(__name__)\n\nclass SegmentationMetrics:\n    \"\"\"\n    Class tính toán các metrics cho segmentation\n    \n    Chức năng:\n    - Tính IoU (Intersection over Union)\n    - Tính Dice coefficient\n    - Tính Accuracy, Precision, Recall\n    - Tính Hausdorff distance\n    \"\"\"\n    \n    def __init__(self, num_classes: int = 3, ignore_index: int = -1):\n        \"\"\"\n        Khởi tạo metrics calculator\n        \n        Args:\n            num_classes: Số lượng classes\n            ignore_index: Index cần ignore khi tính metrics\n        \"\"\"\n        self.num_classes = num_classes\n        self.ignore_index = ignore_index\n        self.reset()\n    \n    def reset(self):\n        \"\"\"\n        Reset tất cả metrics về 0\n        \"\"\"\n        self.total_area_intersect = np.zeros(self.num_classes)\n        self.total_area_union = np.zeros(self.num_classes)\n        self.total_area_pred_label = np.zeros(self.num_classes)\n        self.total_area_label = np.zeros(self.num_classes)\n        self.total_correct = 0\n        self.total_label = 0\n    \n    def update(self, pred: np.ndarray, target: np.ndarray):\n        \"\"\"\n        Cập nhật metrics với prediction và target mới\n        \n        Args:\n            pred: Predicted segmentation [H, W] hoặc [B, H, W]\n            target: Ground truth segmentation [H, W] hoặc [B, H, W]\n        \"\"\"\n        if pred.ndim == 3:  # Batch input\n            for i in range(pred.shape[0]):\n                self._update_single(pred[i], target[i])\n        else:  # Single input\n            self._update_single(pred, target)\n    \n    def _update_single(self, pred: np.ndarray, target: np.ndarray):\n        \"\"\"\n        Cập nhật metrics cho một sample\n        \n        Args:\n            pred: Predicted segmentation [H, W]\n            target: Ground truth segmentation [H, W]\n        \"\"\"\n        # Loại bỏ ignore index\n        mask = (target != self.ignore_index)\n        pred = pred[mask]\n        target = target[mask]\n        \n        # Tính intersection và union cho từng class\n        for class_id in range(self.num_classes):\n            pred_class = (pred == class_id)\n            target_class = (target == class_id)\n            \n            intersect = np.sum(pred_class & target_class)\n            union = np.sum(pred_class | target_class)\n            \n            self.total_area_intersect[class_id] += intersect\n            self.total_area_union[class_id] += union\n            self.total_area_pred_label[class_id] += np.sum(pred_class)\n            self.total_area_label[class_id] += np.sum(target_class)\n        \n        # Tính accuracy tổng thể\n        self.total_correct += np.sum(pred == target)\n        self.total_label += len(pred)\n    \n    def get_iou(self) -> Dict[str, float]:\n        \"\"\"\n        Tính IoU (Intersection over Union) cho từng class\n        \n        Returns:\n            Dict chứa IoU cho từng class và mean IoU\n        \"\"\"\n        iou_per_class = self.total_area_intersect / (self.total_area_union + 1e-8)\n        \n        # Chỉ tính mean IoU cho các class có ground truth\n        valid_classes = self.total_area_label > 0\n        mean_iou = np.mean(iou_per_class[valid_classes])\n        \n        results = {\n            'mean_iou': mean_iou,\n            'iou_per_class': iou_per_class.tolist()\n        }\n        \n        return results\n    \n    def get_dice(self) -> Dict[str, float]:\n        \"\"\"\n        Tính Dice coefficient cho từng class\n        \n        Returns:\n            Dict chứa Dice cho từng class và mean Dice\n        \"\"\"\n        dice_per_class = (2 * self.total_area_intersect) / (\n            self.total_area_pred_label + self.total_area_label + 1e-8\n        )\n        \n        # Chỉ tính mean Dice cho các class có ground truth\n        valid_classes = self.total_area_label > 0\n        mean_dice = np.mean(dice_per_class[valid_classes])\n        \n        results = {\n            'mean_dice': mean_dice,\n            'dice_per_class': dice_per_class.tolist()\n        }\n        \n        return results\n    \n    def get_accuracy(self) -> float:\n        \"\"\"\n        Tính pixel accuracy tổng thể\n        \n        Returns:\n            Pixel accuracy\n        \"\"\"\n        return self.total_correct / (self.total_label + 1e-8)\n    \n    def get_precision_recall(self) -> Dict[str, Any]:\n        \"\"\"\n        Tính Precision và Recall cho từng class\n        \n        Returns:\n            Dict chứa precision và recall cho từng class\n        \"\"\"\n        precision_per_class = self.total_area_intersect / (self.total_area_pred_label + 1e-8)\n        recall_per_class = self.total_area_intersect / (self.total_area_label + 1e-8)\n        \n        # F1 score\n        f1_per_class = 2 * (precision_per_class * recall_per_class) / (\n            precision_per_class + recall_per_class + 1e-8\n        )\n        \n        # Mean values (chỉ tính cho các class có ground truth)\n        valid_classes = self.total_area_label > 0\n        mean_precision = np.mean(precision_per_class[valid_classes])\n        mean_recall = np.mean(recall_per_class[valid_classes])\n        mean_f1 = np.mean(f1_per_class[valid_classes])\n        \n        results = {\n            'mean_precision': mean_precision,\n            'mean_recall': mean_recall,\n            'mean_f1': mean_f1,\n            'precision_per_class': precision_per_class.tolist(),\n            'recall_per_class': recall_per_class.tolist(),\n            'f1_per_class': f1_per_class.tolist()\n        }\n        \n        return results\n    \n    def get_all_metrics(self) -> Dict[str, Any]:\n        \"\"\"\n        Lấy tất cả metrics\n        \n        Returns:\n            Dict chứa tất cả metrics\n        \"\"\"\n        results = {}\n        \n        # IoU metrics\n        iou_results = self.get_iou()\n        results.update(iou_results)\n        \n        # Dice metrics\n        dice_results = self.get_dice()\n        results.update(dice_results)\n        \n        # Accuracy\n        results['accuracy'] = self.get_accuracy()\n        \n        # Precision/Recall\n        pr_results = self.get_precision_recall()\n        results.update(pr_results)\n        \n        return results\n\ndef calculate_hausdorff_distance(pred: np.ndarray, target: np.ndarray) -> float:\n    \"\"\"\n    Tính Hausdorff distance giữa prediction và ground truth\n    \n    Args:\n        pred: Predicted binary mask\n        target: Ground truth binary mask\n        \n    Returns:\n        Hausdorff distance\n    \"\"\"\n    try:\n        import cv2\n        from scipy.spatial.distance import directed_hausdorff\n        \n        # Tìm contours\n        pred_contours, _ = cv2.findContours(pred.astype(np.uint8), cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)\n        target_contours, _ = cv2.findContours(target.astype(np.uint8), cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)\n        \n        if len(pred_contours) == 0 or len(target_contours) == 0:\n            return float('inf')\n        \n        # Lấy tất cả điểm contour\n        pred_points = np.concatenate([contour.reshape(-1, 2) for contour in pred_contours])\n        target_points = np.concatenate([contour.reshape(-1, 2) for contour in target_contours])\n        \n        # Tính Hausdorff distance\n        hausdorff_dist = max(\n            directed_hausdorff(pred_points, target_points)[0],\n            directed_hausdorff(target_points, pred_points)[0]\n        )\n        \n        return hausdorff_dist\n        \n    except ImportError:\n        logger.warning(\"Không thể tính Hausdorff distance: thiếu scipy\")\n        return 0.0\n\ndef evaluate_model(\n    model: torch.nn.Module,\n    dataloader: torch.utils.data.DataLoader,\n    device: torch.device,\n    num_classes: int = 3,\n    class_names: Optional[List[str]] = None\n) -> Dict[str, Any]:\n    \"\"\"\n    Đánh giá model trên validation/test set\n    \n    Args:\n        model: PyTorch model\n        dataloader: DataLoader cho evaluation\n        device: Device để chạy model\n        num_classes: Số lượng classes\n        class_names: Tên các classes\n        \n    Returns:\n        Dict chứa các metrics\n    \"\"\"\n    if class_names is None:\n        class_names = [f'class_{i}' for i in range(num_classes)]\n    \n    model.eval()\n    metrics_calculator = SegmentationMetrics(num_classes)\n    \n    all_predictions = []\n    all_targets = []\n    \n    with torch.no_grad():\n        for batch in dataloader:\n            pixel_values = batch['pixel_values'].to(device)\n            labels = batch['labels'].to(device)\n            \n            # Forward pass\n            if hasattr(model, 'model'):  # SegformerModel wrapper\n                outputs = model(pixel_values=pixel_values)\n                logits = outputs.logits\n            else:  # Direct model\n                logits = model(pixel_values)\n            \n            # Upsample logits to match label size\n            logits = F.interpolate(\n                logits,\n                size=labels.shape[-2:],\n                mode='bilinear',\n                align_corners=False\n            )\n            \n            # Get predictions\n            predictions = torch.argmax(logits, dim=1)\n            \n            # Convert to numpy\n            pred_np = predictions.cpu().numpy()\n            target_np = labels.cpu().numpy()\n            \n            # Update metrics\n            metrics_calculator.update(pred_np, target_np)\n            \n            # Store for confusion matrix\n            all_predictions.extend(pred_np.flatten())\n            all_targets.extend(target_np.flatten())\n    \n    # Tính tất cả metrics\n    results = metrics_calculator.get_all_metrics()\n    \n    # Thêm confusion matrix\n    cm = confusion_matrix(all_targets, all_predictions, labels=range(num_classes))\n    results['confusion_matrix'] = cm.tolist()\n    \n    # Classification report\n    try:\n        class_report = classification_report(\n            all_targets, \n            all_predictions, \n            target_names=class_names,\n            output_dict=True,\n            zero_division=0\n        )\n        results['classification_report'] = class_report\n    except Exception as e:\n        logger.warning(f\"Không thể tạo classification report: {e}\")\n    \n    return results\n\ndef plot_confusion_matrix(\n    cm: np.ndarray,\n    class_names: List[str],\n    title: str = \"Confusion Matrix\",\n    save_path: Optional[str] = None\n):\n    \"\"\"\n    Vẽ confusion matrix\n    \n    Args:\n        cm: Confusion matrix\n        class_names: Tên các classes\n        title: Tiêu đề biểu đồ\n        save_path: Đường dẫn lưu ảnh\n    \"\"\"\n    plt.figure(figsize=(8, 6))\n    \n    # Normalize confusion matrix\n    cm_normalized = cm.astype('float') / cm.sum(axis=1)[:, np.newaxis]\n    \n    # Plot heatmap\n    sns.heatmap(\n        cm_normalized,\n        annot=True,\n        fmt='.2f',\n        cmap='Blues',\n        xticklabels=class_names,\n        yticklabels=class_names\n    )\n    \n    plt.title(title)\n    plt.ylabel('True Label')\n    plt.xlabel('Predicted Label')\n    plt.tight_layout()\n    \n    if save_path:\n        plt.savefig(save_path, dpi=300, bbox_inches='tight')\n        logger.info(f\"Đã lưu confusion matrix: {save_path}\")\n    \n    plt.show()\n\ndef plot_metrics_comparison(\n    metrics_dict: Dict[str, Dict[str, float]],\n    metric_names: List[str] = ['mean_iou', 'mean_dice', 'accuracy'],\n    save_path: Optional[str] = None\n):\n    \"\"\"\n    Vẽ biểu đồ so sánh metrics giữa các models\n    \n    Args:\n        metrics_dict: Dict chứa metrics của các models\n        metric_names: Danh sách metrics cần so sánh\n        save_path: Đường dẫn lưu ảnh\n    \"\"\"\n    fig, axes = plt.subplots(1, len(metric_names), figsize=(6 * len(metric_names), 5))\n    if len(metric_names) == 1:\n        axes = [axes]\n    \n    for idx, metric_name in enumerate(metric_names):\n        model_names = list(metrics_dict.keys())\n        metric_values = [metrics_dict[model][metric_name] for model in model_names]\n        \n        bars = axes[idx].bar(model_names, metric_values)\n        axes[idx].set_title(f'{metric_name.replace(\"_\", \" \").title()}')\n        axes[idx].set_ylabel('Score')\n        axes[idx].set_ylim(0, 1)\n        \n        # Thêm giá trị lên bars\n        for bar, value in zip(bars, metric_values):\n            axes[idx].text(\n                bar.get_x() + bar.get_width()/2,\n                bar.get_height() + 0.01,\n                f'{value:.3f}',\n                ha='center',\n                va='bottom'\n            )\n    \n    plt.tight_layout()\n    \n    if save_path:\n        plt.savefig(save_path, dpi=300, bbox_inches='tight')\n        logger.info(f\"Đã lưu metrics comparison: {save_path}\")\n    \n    plt.show()\n\ndef calculate_class_wise_metrics(\n    predictions: List[np.ndarray],\n    targets: List[np.ndarray],\n    num_classes: int = 3,\n    class_names: Optional[List[str]] = None\n) -> Dict[str, Any]:\n    \"\"\"\n    Tính metrics chi tiết cho từng class\n    \n    Args:\n        predictions: List predicted masks\n        targets: List ground truth masks\n        num_classes: Số lượng classes\n        class_names: Tên các classes\n        \n    Returns:\n        Dict chứa metrics chi tiết cho từng class\n    \"\"\"\n    if class_names is None:\n        class_names = [f'class_{i}' for i in range(num_classes)]\n    \n    metrics_calculator = SegmentationMetrics(num_classes)\n    \n    for pred, target in zip(predictions, targets):\n        metrics_calculator.update(pred, target)\n    \n    results = metrics_calculator.get_all_metrics()\n    \n    # Tạo detailed report cho từng class\n    detailed_results = {}\n    \n    for class_idx, class_name in enumerate(class_names):\n        detailed_results[class_name] = {\n            'iou': results['iou_per_class'][class_idx],\n            'dice': results['dice_per_class'][class_idx],\n            'precision': results['precision_per_class'][class_idx],\n            'recall': results['recall_per_class'][class_idx],\n            'f1': results['f1_per_class'][class_idx]\n        }\n    \n    # Thêm overall metrics\n    detailed_results['overall'] = {\n        'mean_iou': results['mean_iou'],\n        'mean_dice': results['mean_dice'],\n        'accuracy': results['accuracy'],\n        'mean_precision': results['mean_precision'],\n        'mean_recall': results['mean_recall'],\n        'mean_f1': results['mean_f1']\n    }\n    \n    return detailed_results\n\ndef print_evaluation_report(results: Dict[str, Any], class_names: List[str]):\n    \"\"\"\n    In báo cáo đánh giá chi tiết\n    \n    Args:\n        results: Results từ evaluate_model\n        class_names: Tên các classes\n    \"\"\"\n    print(\"=\" * 60)\n    print(\"EVALUATION REPORT\")\n    print(\"=\" * 60)\n    \n    # Overall metrics\n    print(\"\\n📊 OVERALL METRICS:\")\n    print(f\"  • Mean IoU: {results['mean_iou']:.4f}\")\n    print(f\"  • Mean Dice: {results['mean_dice']:.4f}\")\n    print(f\"  • Accuracy: {results['accuracy']:.4f}\")\n    print(f\"  • Mean Precision: {results['mean_precision']:.4f}\")\n    print(f\"  • Mean Recall: {results['mean_recall']:.4f}\")\n    print(f\"  • Mean F1: {results['mean_f1']:.4f}\")\n    \n    # Per-class metrics\n    print(\"\\n📈 PER-CLASS METRICS:\")\n    for class_idx, class_name in enumerate(class_names):\n        print(f\"\\n  {class_name.upper()}:\")\n        print(f\"    IoU: {results['iou_per_class'][class_idx]:.4f}\")\n        print(f\"    Dice: {results['dice_per_class'][class_idx]:.4f}\")\n        print(f\"    Precision: {results['precision_per_class'][class_idx]:.4f}\")\n        print(f\"    Recall: {results['recall_per_class'][class_idx]:.4f}\")\n        print(f\"    F1: {results['f1_per_class'][class_idx]:.4f}\")\n    \n    print(\"\\n\" + \"=\" * 60)\n\ndef compare_models(\n    model_results: Dict[str, Dict[str, Any]],\n    class_names: List[str]\n) -> Dict[str, str]:\n    \"\"\"\n    So sánh performance của các models\n    \n    Args:\n        model_results: Dict chứa results của các models\n        class_names: Tên các classes\n        \n    Returns:\n        Dict chứa model tốt nhất cho từng metric\n    \"\"\"\n    metrics_to_compare = ['mean_iou', 'mean_dice', 'accuracy', 'mean_f1']\n    best_models = {}\n    \n    print(\"=\" * 60)\n    print(\"MODEL COMPARISON\")\n    print(\"=\" * 60)\n    \n    for metric in metrics_to_compare:\n        best_model = None\n        best_score = -1\n        \n        print(f\"\\n📊 {metric.replace('_', ' ').upper()}:\")\n        \n        for model_name, results in model_results.items():\n            score = results[metric]\n            print(f\"  • {model_name}: {score:.4f}\")\n            \n            if score > best_score:\n                best_score = score\n                best_model = model_name\n        \n        best_models[metric] = best_model\n        print(f\"  🏆 Best: {best_model} ({best_score:.4f})\")\n    \n    print(\"\\n\" + \"=\" * 60)\n    \n    return best_models\nprint('DONE')","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-07-21T07:59:26.600108Z","iopub.execute_input":"2025-07-21T07:59:26.600460Z","iopub.status.idle":"2025-07-21T07:59:26.900002Z","shell.execute_reply.started":"2025-07-21T07:59:26.600434Z","shell.execute_reply":"2025-07-21T07:59:26.899076Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"\"\"\"\nFile chứa inference pipeline cho bài toán phân đoạn polyp\nMục đích: Predict trên test set và tạo submission file\n\"\"\"\n\nimport os\nimport torch\nimport torch.nn as nn\nimport torch.nn.functional as F\nimport numpy as np\nfrom PIL import Image\nfrom typing import List, Dict, Tuple, Optional, Any\nimport logging\nfrom tqdm import tqdm\nimport cv2\n\n# Import local modules\n# Config đã được load từ cell trước, không cần import\n# from config import Config\n# from models import get_model\n# from utils import (...)\n# from data_processing import get_augmentation_pipeline\n\n# Sử dụng trực tiếp các class/function đã được load\n\n# Setup logging\nlogging.basicConfig(level=logging.INFO)\nlogger = logging.getLogger(__name__)\n\nclass PolypInference:\n    \"\"\"\n    Inference class cho bài toán phân đoạn polyp\n    \n    Chức năng:\n    - Load trained model\n    - Predict trên test images\n    - Apply Test Time Augmentation (TTA)\n    - Post-processing predictions\n    - Tạo submission file\n    \"\"\"\n    \n    def __init__(self, config: Config, model_path: str):\n        \"\"\"\n        Khởi tạo inference pipeline\n        \n        Args:\n            config: Configuration object\n            model_path: Đường dẫn tới trained model\n        \"\"\"\n        self.config = config\n        self.device = config.get_device()\n        self.model_path = model_path\n        \n        # Load model\n        self.model = self._load_model()\n        self.model.eval()\n        \n        # Load image processor\n        try:\n            from transformers import SegformerImageProcessor\n            self.processor = SegformerImageProcessor(reduce_labels=False)\n        except ImportError:\n            self.processor = None\n            logger.warning(\"SegformerImageProcessor không có sẵn\")\n        \n        logger.info(\"Đã khởi tạo inference pipeline\")\n    \n    def _load_model(self) -> nn.Module:\n        \"\"\"\n        Load trained model từ checkpoint\n        \n        Returns:\n            Loaded PyTorch model\n        \"\"\"\n        # Tạo model architecture\n        model = get_model(\n            model_name=self.config.model.model_name,\n            num_classes=self.config.model.num_classes,\n            backbone=self.config.model.backbone,\n            id2label=self.config.model.id2label,\n            label2id=self.config.model.label2id,\n            pretrained=False  # Không cần pretrained weights\n        )\n        \n        # Load checkpoint\n        if os.path.exists(self.model_path):\n            checkpoint = torch.load(self.model_path, map_location=self.device)\n            \n            # Load model state dict\n            if 'model_state_dict' in checkpoint:\n                model.load_state_dict(checkpoint['model_state_dict'])\n                logger.info(f\"Loaded model từ checkpoint: {self.model_path}\")\n            else:\n                model.load_state_dict(checkpoint)\n                logger.info(f\"Loaded model weights: {self.model_path}\")\n        else:\n            raise FileNotFoundError(f\"Không tìm thấy model: {self.model_path}\")\n        \n        model.to(self.device)\n        return model\n    \n    def predict_single_image(\n        self, \n        image_path: str, \n        use_tta: bool = False,\n        visualize: bool = False\n    ) -> np.ndarray:\n        \"\"\"\n        Predict segmentation cho một ảnh\n        \n        Args:\n            image_path: Đường dẫn tới ảnh\n            use_tta: Có sử dụng Test Time Augmentation không\n            visualize: Có visualize kết quả không\n            \n        Returns:\n            Predicted segmentation mask [H, W]\n        \"\"\"\n        # Load và preprocess ảnh\n        image = Image.open(image_path).convert('RGB')\n        original_size = image.size\n        \n        if use_tta and self.config.inference.use_tta:\n            prediction = self._predict_with_tta(image)\n        else:\n            prediction = self._predict_single(image)\n        \n        # Resize về kích thước gốc\n        prediction = cv2.resize(\n            prediction.astype(np.uint8), \n            original_size, \n            interpolation=cv2.INTER_NEAREST\n        )\n        \n        # Post-processing\n        if self.config.inference.use_morphology:\n            prediction = post_process_mask(prediction)\n        \n        # Visualization\n        if visualize:\n            image_np = np.array(image)\n            visualize_sample(image_np, prediction, class_names=list(self.config.model.id2label.values()))\n        \n        return prediction\n    \n    def _predict_single(self, image: Image.Image) -> np.ndarray:\n        \"\"\"\n        Predict cho một ảnh đơn lẻ\n        \n        Args:\n            image: PIL Image\n            \n        Returns:\n            Predicted mask [H, W]\n        \"\"\"\n        # Preprocess image\n        if self.processor is not None:\n            inputs = self.processor(image, return_tensors=\"pt\")\n            pixel_values = inputs.pixel_values.to(self.device)\n        else:\n            # Fallback preprocessing\n            from torchvision import transforms\n            transform = transforms.Compose([\n                transforms.Resize((512, 512)),\n                transforms.ToTensor(),\n                transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225])\n            ])\n            pixel_values = transform(image).unsqueeze(0).to(self.device)\n        \n        # Inference\n        with torch.no_grad():\n            if hasattr(self.model, 'model'):  # SegformerModel wrapper\n                outputs = self.model(pixel_values=pixel_values)\n                logits = outputs.logits\n            else:  # Direct model\n                logits = self.model(pixel_values)\n            \n            # Upsample logits to original image size\n            logits = F.interpolate(\n                logits,\n                size=image.size[::-1],  # (height, width)\n                mode='bilinear',\n                align_corners=False\n            )\n            \n            # Get predictions\n            predictions = torch.argmax(logits, dim=1)\n            prediction = predictions.cpu().numpy()[0]\n        \n        return prediction\n    \n    def _predict_with_tta(self, image: Image.Image) -> np.ndarray:\n        \"\"\"\n        Predict với Test Time Augmentation\n        \n        Args:\n            image: PIL Image\n            \n        Returns:\n            Averaged prediction mask [H, W]\n        \"\"\"\n        # Tạo augmentation pipeline\n        tta_transforms = get_augmentation_pipeline(is_training=True)\n        \n        predictions = []\n        image_np = np.array(image)\n        \n        # Original prediction\n        predictions.append(self._predict_single(image))\n        \n        # Augmented predictions\n        for _ in range(self.config.inference.tta_transforms - 1):\n            if tta_transforms is not None:\n                # Apply augmentation\n                augmented = tta_transforms(image=image_np)\n                aug_image = Image.fromarray(augmented['image'])\n                \n                # Predict\n                pred = self._predict_single(aug_image)\n                predictions.append(pred)\n            else:\n                # Fallback: simple flips\n                if np.random.random() > 0.5:\n                    flipped_image = image.transpose(Image.Transpose.FLIP_LEFT_RIGHT)\n                    pred = self._predict_single(flipped_image)\n                    pred = np.fliplr(pred)  # Flip prediction back\n                    predictions.append(pred)\n        \n        # Average predictions\n        averaged_prediction = np.mean(predictions, axis=0)\n        final_prediction = np.argmax(averaged_prediction, axis=0) if len(averaged_prediction.shape) > 2 else averaged_prediction\n        \n        return final_prediction.astype(np.uint8)\n    \n    def predict_test_set(\n        self, \n        test_image_paths: List[str],\n        output_dir: str,\n        use_tta: bool = False,\n        save_masks: bool = True\n    ) -> List[np.ndarray]:\n        \"\"\"\n        Predict cho toàn bộ test set\n        \n        Args:\n            test_image_paths: Danh sách đường dẫn ảnh test\n            output_dir: Thư mục lưu kết quả\n            use_tta: Có sử dụng TTA không\n            save_masks: Có lưu mask predictions không\n            \n        Returns:\n            List predicted masks\n        \"\"\"\n        os.makedirs(output_dir, exist_ok=True)\n        \n        predictions = []\n        \n        logger.info(f\"Bắt đầu predict {len(test_image_paths)} ảnh test\")\n        \n        for image_path in tqdm(test_image_paths, desc=\"Predicting\"):\n            try:\n                # Predict\n                prediction = self.predict_single_image(\n                    image_path, \n                    use_tta=use_tta,\n                    visualize=False\n                )\n                \n                predictions.append(prediction)\n                \n                # Lưu mask nếu cần\n                if save_masks:\n                    image_name = os.path.basename(image_path).split('.')[0]\n                    mask_path = os.path.join(output_dir, f\"{image_name}_mask.png\")\n                    \n                    # Convert prediction to RGB for visualization\n                    mask_rgb = self._prediction_to_rgb(prediction)\n                    Image.fromarray(mask_rgb).save(mask_path)\n                \n            except Exception as e:\n                logger.error(f\"Lỗi khi predict {image_path}: {str(e)}\")\n                # Tạo empty prediction\n                predictions.append(np.zeros((512, 512), dtype=np.uint8))\n        \n        logger.info(f\"Hoàn thành prediction cho {len(predictions)} ảnh\")\n        \n        return predictions\n    \n    def _prediction_to_rgb(self, prediction: np.ndarray) -> np.ndarray:\n        \"\"\"\n        Chuyển prediction mask thành RGB cho visualization\n        \n        Args:\n            prediction: Prediction mask [H, W]\n            \n        Returns:\n            RGB mask [H, W, 3]\n        \"\"\"\n        # Color mapping\n        colors = np.array([\n            [255, 0, 0],      # Red cho neoplastic\n            [0, 255, 0],      # Green cho non-neoplastic\n            [0, 0, 255]       # Blue cho background\n        ])\n        \n        rgb_mask = colors[prediction]\n        return rgb_mask.astype(np.uint8)\n    \n    def create_submission(\n        self,\n        test_image_paths: List[str],\n        predictions: List[np.ndarray],\n        output_path: str = \"submission.csv\"\n    ):\n        \"\"\"\n        Tạo submission file cho Kaggle\n        \n        Args:\n            test_image_paths: Danh sách đường dẫn ảnh test\n            predictions: Danh sách predicted masks\n            output_path: Đường dẫn file submission\n        \"\"\"\n        # Lấy image IDs\n        image_ids = []\n        for image_path in test_image_paths:\n            image_name = os.path.basename(image_path)\n            image_id = image_name.split('.')[0]\n            image_ids.append(image_id)\n        \n        # Tạo submission file\n        create_submission_file(predictions, image_ids, output_path)\n        \n        logger.info(f\"Đã tạo submission file: {output_path}\")\n    \n    def batch_inference(\n        self,\n        test_image_paths: List[str],\n        output_dir: str,\n        submission_path: str = \"submission.csv\",\n        use_tta: bool = False,\n        save_masks: bool = True\n    ):\n        \"\"\"\n        Thực hiện inference cho toàn bộ test set và tạo submission\n        \n        Args:\n            test_image_paths: Danh sách đường dẫn ảnh test\n            output_dir: Thư mục lưu kết quả\n            submission_path: Đường dẫn file submission\n            use_tta: Có sử dụng TTA không\n            save_masks: Có lưu mask predictions không\n        \"\"\"\n        # Predict test set\n        predictions = self.predict_test_set(\n            test_image_paths,\n            output_dir,\n            use_tta=use_tta,\n            save_masks=save_masks\n        )\n        \n        # Tạo submission file\n        self.create_submission(\n            test_image_paths,\n            predictions,\n            os.path.join(output_dir, submission_path)\n        )\n        \n        logger.info(\"Hoàn thành batch inference\")\n\ndef run_inference(\n    config: Config,\n    model_path: str,\n    test_image_paths: List[str],\n    output_dir: str,\n    use_tta: bool = False\n) -> List[np.ndarray]:\n    \"\"\"\n    Function wrapper để chạy inference\n    \n    Args:\n        config: Configuration object\n        model_path: Đường dẫn tới trained model\n        test_image_paths: Danh sách đường dẫn ảnh test\n        output_dir: Thư mục lưu kết quả\n        use_tta: Có sử dụng TTA không\n        \n    Returns:\n        List predicted masks\n    \"\"\"\n    # Tạo inference pipeline\n    inference = PolypInference(config, model_path)\n    \n    # Chạy batch inference\n    inference.batch_inference(\n        test_image_paths,\n        output_dir,\n        use_tta=use_tta,\n        save_masks=True\n    )\n    \n    # Predict và return\n    predictions = inference.predict_test_set(\n        test_image_paths,\n        output_dir,\n        use_tta=use_tta,\n        save_masks=False\n    )\n    \n    return predictions\n\ndef ensemble_predictions(\n    predictions_list: List[List[np.ndarray]],\n    method: str = 'voting'\n) -> List[np.ndarray]:\n    \"\"\"\n    Ensemble predictions từ nhiều models\n    \n    Args:\n        predictions_list: List các predictions từ các models khác nhau\n        method: Phương pháp ensemble ('voting', 'averaging')\n        \n    Returns:\n        Ensembled predictions\n    \"\"\"\n    if len(predictions_list) == 1:\n        return predictions_list[0]\n    \n    n_samples = len(predictions_list[0])\n    ensembled_predictions = []\n    \n    for i in range(n_samples):\n        # Lấy predictions của sample i từ tất cả models\n        sample_predictions = [preds[i] for preds in predictions_list]\n        \n        if method == 'voting':\n            # Majority voting\n            stacked_preds = np.stack(sample_predictions, axis=0)\n            ensembled_pred = np.apply_along_axis(\n                lambda x: np.bincount(x).argmax(), \n                axis=0, \n                arr=stacked_preds\n            )\n        elif method == 'averaging':\n            # Average probabilities (if available)\n            averaged_pred = np.mean(sample_predictions, axis=0)\n            ensembled_pred = np.argmax(averaged_pred, axis=-1) if len(averaged_pred.shape) > 2 else averaged_pred\n        else:\n            raise ValueError(f\"Ensemble method không hỗ trợ: {method}\")\n        \n        ensembled_predictions.append(ensembled_pred.astype(np.uint8))\n    \n    logger.info(f\"Ensembled {len(predictions_list)} models với method: {method}\")\n    \n    return ensembled_predictions\n\ndef compare_predictions(\n    predictions_list: List[List[np.ndarray]],\n    model_names: List[str],\n    test_image_paths: List[str],\n    output_dir: str,\n    num_samples: int = 5\n):\n    \"\"\"\n    So sánh predictions từ các models khác nhau\n    \n    Args:\n        predictions_list: List predictions từ các models\n        model_names: Tên các models\n        test_image_paths: Đường dẫn ảnh test\n        output_dir: Thư mục lưu kết quả\n        num_samples: Số samples để so sánh\n    \"\"\"\n    os.makedirs(output_dir, exist_ok=True)\n    \n    # Chọn random samples để so sánh\n    import random\n    sample_indices = random.sample(range(len(test_image_paths)), min(num_samples, len(test_image_paths)))\n    \n    for idx in sample_indices:\n        image_path = test_image_paths[idx]\n        image_name = os.path.basename(image_path).split('.')[0]\n        \n        # Load original image\n        image = Image.open(image_path).convert('RGB')\n        image_np = np.array(image)\n        \n        # Tạo comparison plot\n        n_models = len(predictions_list)\n        fig, axes = plt.subplots(1, n_models + 1, figsize=(5 * (n_models + 1), 5))\n        \n        # Original image\n        axes[0].imshow(image_np)\n        axes[0].set_title('Original Image')\n        axes[0].axis('off')\n        \n        # Predictions từ các models\n        for model_idx, (predictions, model_name) in enumerate(zip(predictions_list, model_names)):\n            prediction = predictions[idx]\n            \n            # Convert to RGB\n            colors = np.array([\n                [255, 0, 0],      # Red\n                [0, 255, 0],      # Green\n                [0, 0, 255]       # Blue\n            ])\n            pred_rgb = colors[prediction]\n            \n            axes[model_idx + 1].imshow(pred_rgb.astype(np.uint8))\n            axes[model_idx + 1].set_title(f'{model_name}')\n            axes[model_idx + 1].axis('off')\n        \n        plt.tight_layout()\n        \n        # Lưu comparison\n        comparison_path = os.path.join(output_dir, f\"comparison_{image_name}.png\")\n        plt.savefig(comparison_path, dpi=300, bbox_inches='tight')\n        plt.close()\n    \n    logger.info(f\"Đã tạo comparison cho {len(sample_indices)} samples\")\n\n# Import matplotlib để tránh lỗi\nimport matplotlib.pyplot as plt \nprint('DONE')","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-07-21T07:59:38.479462Z","iopub.execute_input":"2025-07-21T07:59:38.480535Z","iopub.status.idle":"2025-07-21T07:59:38.520803Z","shell.execute_reply.started":"2025-07-21T07:59:38.480499Z","shell.execute_reply":"2025-07-21T07:59:38.520002Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"\"\"\"\nFile training riêng cho DeepLabV3 model\nMục đích: Training DeepLabV3 model với ASPP module cho multi-scale features\n\"\"\"\n\nimport os\nimport torch\nimport torch.nn as nn\nimport torch.optim as optim\nfrom torch.utils.data import DataLoader\nimport numpy as np\nfrom typing import Dict, List, Optional, Tuple\nimport logging\nfrom tqdm import tqdm\nimport time\n\n\n# Import WandB for experiment tracking\ntry:\n    import wandb\n    WANDB_AVAILABLE = True\n    print(\"✅ WandB imported successfully\")\n    \n    # Login to WandB with provided key\n    wandb.login(key=\"d6fd0da4e97ed749ef3eeacb295c97e7ab2e6aa7\")\n    print(\"🔐 WandB logged in successfully with provided key\")\nexcept ImportError:\n    WANDB_AVAILABLE = False\n    print(\"❌ WandB not available. Install with: pip install wandb\")\nexcept Exception as e:\n    print(f\"⚠️ WandB login failed: {e}\")\n    WANDB_AVAILABLE = False\n\n# Import transformers for processor\ntry:\n    from transformers import SegformerImageProcessor\n    TRANSFORMERS_AVAILABLE = True\nexcept ImportError:\n    TRANSFORMERS_AVAILABLE = False\n    # Fallback: tạo dummy processor\n    class DummyProcessor:\n        def __init__(self, reduce_labels=False):\n            pass\n    SegformerImageProcessor = DummyProcessor\n\n# Setup logging\nlogging.basicConfig(level=logging.INFO)\nlogger = logging.getLogger(__name__)\n\nclass DeepLabV3Trainer:\n    \"\"\"\n    Trainer class riêng cho DeepLabV3 model\n    \n    Chức năng:\n    - Training DeepLabV3 với ASPP module\n    - Multi-scale training strategy\n    - Advanced learning rate scheduling\n    - Comprehensive metrics tracking\n    \"\"\"\n    \n    def __init__(self, config):\n        \"\"\"\n        Khởi tạo DeepLabV3 trainer\n        \n        Args:\n            config: Configuration object\n        \"\"\"\n        self.config = config\n        self.device = config.get_device()\n        \n        # Set random seed\n        set_seed(42)\n        \n        # Tạo thư mục cần thiết\n        config.create_directories()\n        \n        # Initialize processor\n        if TRANSFORMERS_AVAILABLE:\n            self.processor = SegformerImageProcessor(reduce_labels=False)\n        else:\n            self.processor = None\n            logger.warning(\"SegformerImageProcessor không có sẵn, sử dụng basic preprocessing\")\n        \n        # Initialize model\n        self.model = self._build_model()\n        self.model.to(self.device)\n        \n        # Initialize optimizer và scheduler\n        self.optimizer = self._build_optimizer()\n        self.scheduler = self._build_scheduler()\n        \n        # Initialize loss function\n        self.criterion = self._build_criterion()\n        \n        # Initialize metrics\n        self.metrics = SegmentationMetrics(config.model.num_classes)\n        \n        # Training history\n        self.history = {\n            'train_loss': [],\n            'val_loss': [],\n            'train_iou': [],\n            'val_iou': [],\n            'train_dice': [],\n            'val_dice': [],\n            'learning_rates': []\n        }\n        \n        # Best model tracking\n        self.best_val_iou = 0.0\n        self.patience_counter = 0\n        \n        # Initialize WandB project - Project sẽ được tạo khi init\n        self.use_wandb = WANDB_AVAILABLE\n        if self.use_wandb:\n            # Create run config for wandb\n            wandb_config = {\n                \"model_name\": \"DeepLabV3\",\n                \"backbone\": \"ResNet50\",\n                \"num_classes\": config.model.num_classes,\n                \"learning_rate\": 1e-5,\n                \"batch_size\": 4,\n                \"epochs\": 150,\n                \"optimizer\": \"SGD\",\n                \"scheduler\": \"PolynomialLR\",\n                \"loss_function\": \"CrossEntropyLoss\",\n                \"aspp_features\": [256, 256, 256, 256],\n                \"patience\": 30\n            }\n            \n            # Initialize wandb - Project sẽ được tạo tự động\n            wandb.init(\n                project=\"polyp-segmentation-deeplabv3\",  # Tên project cụ thể\n                name=f\"deeplabv3_{config.experiment.run_name}\",\n                config=wandb_config,\n                tags=[\"deeplabv3\", \"polyp-segmentation\", \"medical-imaging\", \"aspp\"],\n                notes=\"DeepLabV3 training for polyp segmentation with ASPP module\"\n            )\n            \n            # Watch model for gradients and parameters\n            wandb.watch(self.model, log=\"all\", log_freq=100)\n            logger.info(\"✅ WandB project 'polyp-segmentation-deeplabv3' đã được tạo và tracking initialized\")\n        else:\n            logger.info(\"❌ WandB tracking disabled\")\n        \n        logger.info(\"Đã khởi tạo DeepLabV3Trainer\")\n    \n    def _build_model(self) -> DeepLabV3:\n        \"\"\"\n        Tạo DeepLabV3 model\n        \n        Returns:\n            DeepLabV3 instance\n        \"\"\"\n        model = DeepLabV3(\n            num_classes=self.config.model.num_classes,\n            backbone=\"resnet50\"\n        )\n        \n        logger.info(f\"Đã tạo DeepLabV3 model với backbone: resnet50\")\n        \n        # In thông tin model\n        total_params = sum(p.numel() for p in model.parameters())\n        trainable_params = sum(p.numel() for p in model.parameters() if p.requires_grad)\n        \n        logger.info(f\"Total parameters: {total_params:,}\")\n        logger.info(f\"Trainable parameters: {trainable_params:,}\")\n        \n        return model\n    \n    def _build_optimizer(self) -> optim.Optimizer:\n        \"\"\"\n        Tạo optimizer cho DeepLabV3\n        \n        Returns:\n            PyTorch optimizer\n        \"\"\"\n        # Sử dụng SGD với momentum cho DeepLabV3 (theo paper gốc)\n        optimizer = optim.SGD(\n            self.model.parameters(),\n            lr=5e-3,  # Giảm learning rate để train ổn định hơn\n            momentum=0.9,\n            weight_decay=1e-4\n        )\n        \n        logger.info(f\"Đã tạo SGD optimizer với lr=5e-3, momentum=0.9\")\n        \n        return optimizer\n    \n    def _build_scheduler(self) -> optim.lr_scheduler.LRScheduler:\n        \"\"\"\n        Tạo learning rate scheduler\n        \n        Returns:\n            PyTorch scheduler\n        \"\"\"\n        # Sử dụng Polynomial LR decay cho DeepLabV3\n        scheduler = optim.lr_scheduler.PolynomialLR(\n            self.optimizer,\n            total_iters=150, \n            power=0.9\n        )\n        \n        logger.info(\"Đã tạo PolynomialLR scheduler\")\n        \n        return scheduler\n    \n    def _build_criterion(self) -> nn.Module:\n        \"\"\"\n        Tạo loss function cho DeepLabV3\n        \n        Returns:\n            Loss function\n        \"\"\"\n        # Sử dụng CrossEntropyLoss với class weights\n        if hasattr(self.config.training, 'class_weights') and self.config.training.class_weights:\n            weights = torch.tensor(self.config.training.class_weights).to(self.device)\n            criterion = nn.CrossEntropyLoss(weight=weights)\n        else:\n            criterion = nn.CrossEntropyLoss()\n        \n        logger.info(\"Đã tạo CrossEntropyLoss\")\n        \n        return criterion\n    \n    def train_epoch(self, train_loader: DataLoader, epoch: int) -> Dict[str, float]:\n        \"\"\"\n        Training một epoch\n        \n        Args:\n            train_loader: DataLoader cho training\n            epoch: Epoch hiện tại\n            \n        Returns:\n            Dict chứa training metrics\n        \"\"\"\n        self.model.train()\n        self.metrics.reset()\n        \n        total_loss = 0\n        num_batches = len(train_loader)\n        \n        pbar = tqdm(train_loader, desc=f\"Epoch {epoch+1} - Training\", bar_format=' {l_bar}{bar}| {n_fmt}/{total_fmt} [{elapsed}<{remaining}, {rate_fmt}] {postfix}')\n        \n        for batch_idx, batch in enumerate(pbar):\n            if self.processor:\n                pixel_values = batch['pixel_values'].to(self.device)\n                labels = batch['labels'].to(self.device)\n            else:\n                pixel_values = batch['image'].to(self.device)\n                labels = batch['mask'].to(self.device)\n            \n            # Zero gradients\n            self.optimizer.zero_grad()\n            \n            # Forward pass\n            logits = self.model(pixel_values)\n            \n            # Calculate loss\n            loss = self.criterion(logits, labels)\n            \n            # Backward pass\n            loss.backward()\n            \n            # Gradient clipping\n            torch.nn.utils.clip_grad_norm_(self.model.parameters(), max_norm=1.0)\n            \n            # Update weights\n            self.optimizer.step()\n            \n            # Update metrics\n            total_loss += loss.item()\n            \n            # Calculate predictions for metrics\n            with torch.no_grad():\n                predictions = torch.argmax(logits, dim=1)\n                \n                self.metrics.update(\n                    predictions.cpu().numpy(),\n                    labels.cpu().numpy()\n                )\n            \n            # Update progress bar\n            current_lr = self.optimizer.param_groups[0]['lr']\n            pbar.set_postfix({\n                'loss': f'{loss.item():.4f}',\n                'lr': f'{current_lr:.6f}'\n            })\n        \n        # Calculate epoch metrics\n        avg_loss = total_loss / num_batches\n        metrics = self.metrics.get_all_metrics()\n        \n        results = {\n            'loss': avg_loss,\n            'iou': metrics['mean_iou'],\n            'dice': metrics['mean_dice'],\n            'accuracy': metrics['accuracy']\n        }\n        \n        return results\n    \n    def validate_epoch(self, val_loader: DataLoader, epoch: int) -> Dict[str, float]:\n        \"\"\"\n        Validation một epoch\n        \n        Args:\n            val_loader: DataLoader cho validation\n            epoch: Epoch hiện tại\n            \n        Returns:\n            Dict chứa validation metrics\n        \"\"\"\n        self.model.eval()\n        self.metrics.reset()\n        \n        total_loss = 0\n        num_batches = len(val_loader)\n        \n        with torch.no_grad():\n            pbar = tqdm(val_loader, desc=f\"Epoch {epoch+1} - Validation\", bar_format=' {l_bar}{bar}| {n_fmt}/{total_fmt} [{elapsed}<{remaining}, {rate_fmt}] {postfix}')\n            \n            for batch in pbar:\n                if self.processor:\n                    pixel_values = batch['pixel_values'].to(self.device)\n                    labels = batch['labels'].to(self.device)\n                else:\n                    pixel_values = batch['image'].to(self.device)\n                    labels = batch['mask'].to(self.device)\n                \n                # Forward pass\n                logits = self.model(pixel_values)\n                \n                # Calculate loss\n                loss = self.criterion(logits, labels)\n                \n                total_loss += loss.item()\n                \n                # Calculate predictions for metrics\n                predictions = torch.argmax(logits, dim=1)\n                \n                self.metrics.update(\n                    predictions.cpu().numpy(),\n                    labels.cpu().numpy()\n                )\n                \n                pbar.set_postfix({'loss': f'{loss.item():.4f}'})\n        \n        # Calculate epoch metrics\n        avg_loss = total_loss / num_batches\n        metrics = self.metrics.get_all_metrics()\n        \n        results = {\n            'loss': avg_loss,\n            'iou': metrics['mean_iou'],\n            'dice': metrics['mean_dice'],\n            'accuracy': metrics['accuracy']\n        }\n        \n        return results\n    \n    def train(\n        self,\n        train_loader: DataLoader,\n        val_loader: DataLoader\n    ) -> Dict[str, List[float]]:\n        \"\"\"\n        Main training loop cho DeepLabV3\n        \n        Args:\n            train_loader: Training DataLoader\n            val_loader: Validation DataLoader\n            \n        Returns:\n            Training history\n        \"\"\"\n        epochs = 150  # Tăng epochs để train 2 tiếng\n        logger.info(f\"Bắt đầu training DeepLabV3 với {epochs} epochs\")\n        \n        start_time = time.time()\n        \n        for epoch in range(epochs):\n            epoch_start_time = time.time()\n            \n            # Training phase\n            train_results = self.train_epoch(train_loader, epoch)\n            \n            # Validation phase\n            val_results = self.validate_epoch(val_loader, epoch)\n            \n            # Update learning rate\n            self.scheduler.step()\n            current_lr = self.optimizer.param_groups[0]['lr']\n            \n            # Update history\n            self.history['train_loss'].append(train_results['loss'])\n            self.history['val_loss'].append(val_results['loss'])\n            self.history['train_iou'].append(train_results['iou'])\n            self.history['val_iou'].append(val_results['iou'])\n            self.history['train_dice'].append(train_results['dice'])\n            self.history['val_dice'].append(val_results['dice'])\n            self.history['learning_rates'].append(current_lr)\n            \n            # Log epoch results\n            epoch_time = time.time() - epoch_start_time\n            logger.info(f\"Epoch {epoch+1}/{epochs} - {epoch_time:.2f}s\")\n            logger.info(f\"Train - Loss: {train_results['loss']:.4f}, IoU: {train_results['iou']:.4f}, Dice: {train_results['dice']:.4f}\")\n            logger.info(f\"Val - Loss: {val_results['loss']:.4f}, IoU: {val_results['iou']:.4f}, Dice: {val_results['dice']:.4f}\")\n            logger.info(f\"Learning Rate: {current_lr:.6f}\")\n            \n            # Log to WandB\n            if self.use_wandb:\n                wandb.log({\n                    # Loss metrics (train, val, test trên cùng bảng)\n                    \"loss/train\": train_results['loss'],\n                    \"loss/val\": val_results['loss'],\n                    \n                    # IoU metrics (train, val, test trên cùng bảng) \n                    \"iou/train\": train_results['iou'],\n                    \"iou/val\": val_results['iou'],\n                    \n                    # Dice metrics (train, val, test trên cùng bảng)\n                    \"dice/train\": train_results['dice'],\n                    \"dice/val\": val_results['dice'],\n                    \n                    # Accuracy metrics\n                    \"accuracy/train\": train_results['accuracy'],\n                    \"accuracy/val\": val_results['accuracy'],\n                    \n                    # Training info\n                    \"learning_rate\": current_lr,\n                    \"epoch\": epoch + 1,\n                    \"epoch_time\": epoch_time,\n                    \"best_val_iou\": self.best_val_iou,\n                    \"patience_counter\": self.patience_counter\n                })\n            \n            # Save best model\n            if val_results['iou'] > self.best_val_iou:\n                self.best_val_iou = val_results['iou']\n                self.patience_counter = 0\n                \n                # Save checkpoint\n                best_model_path = os.path.join(\n                    self.config.experiment.model_save_dir,\n                    f\"best_deeplabv3_{self.config.experiment.run_name}.pth\"\n                )\n                \n                save_model_checkpoint(\n                    self.model,\n                    self.optimizer,\n                    epoch + 1,\n                    val_results['loss'],\n                    val_results,\n                    best_model_path\n                )\n                \n                logger.info(f\"Lưu best model với IoU: {val_results['iou']:.4f}\")\n                \n                # Log best model to WandB\n                if self.use_wandb:\n                    wandb.log({\n                        \"best_model/val_iou\": val_results['iou'],\n                        \"best_model/val_dice\": val_results['dice'],\n                        \"best_model/val_accuracy\": val_results['accuracy'],\n                        \"best_model/val_loss\": val_results['loss'],\n                        \"best_model/epoch\": epoch + 1\n                    })\n                    \n                    # Save model artifact to WandB\n                    model_artifact = wandb.Artifact(\n                        name=f\"deeplabv3_best_model_{wandb.run.id}\",\n                        type=\"model\",\n                        description=f\"Best DeepLabV3 model at epoch {epoch+1} with IoU: {val_results['iou']:.4f}\"\n                    )\n                    model_artifact.add_file(best_model_path)\n                    wandb.log_artifact(model_artifact)\n            else:\n                self.patience_counter += 1\n            \n            # Early stopping\n            if self.patience_counter >= 30:\n                logger.info(f\"Early stopping tại epoch {epoch+1}\")\n                break\n        \n        # Training completed\n        total_time = time.time() - start_time\n        logger.info(f\"Training hoàn thành trong {total_time:.2f}s\")\n        logger.info(f\"Best validation IoU: {self.best_val_iou:.4f}\")\n        \n        # Save final results\n        self._save_final_results(total_time)\n        \n        # Finish WandB run\n        if self.use_wandb:\n            wandb.finish()\n        \n        return self.history\n    \n    def _save_final_results(self, total_time: float):\n        \"\"\"\n        Lưu kết quả cuối cùng vào file JSON\n        \n        Args:\n            total_time: Tổng thời gian training\n        \"\"\"\n        import json\n        \n        results_data = {\n            'model_name': 'DeepLabV3',\n            'backbone': 'ResNet50',\n            'best_val_iou': float(self.best_val_iou),\n            'best_val_dice': float(max(self.history['val_dice'])),\n            'best_val_accuracy': float(max([0.0] + [acc for acc in self.history.get('val_accuracy', [0.0])])),\n            'final_train_loss': float(self.history['train_loss'][-1]) if self.history['train_loss'] else 0.0,\n            'final_val_loss': float(self.history['val_loss'][-1]) if self.history['val_loss'] else 0.0,\n            'total_params': sum(p.numel() for p in self.model.parameters()),\n            'trainable_params': sum(p.numel() for p in self.model.parameters() if p.requires_grad),\n            'training_time_seconds': float(total_time),\n            'training_time_hours': float(total_time / 3600),\n            'total_epochs': len(self.history['train_loss']),\n            'learning_rate': 1e-5,\n            'batch_size': 4,\n            'optimizer': 'SGD',\n            'scheduler': 'PolynomialLR',\n            'loss_function': 'CrossEntropyLoss'\n        }\n        \n        results_file = os.path.join(\n            self.config.experiment.log_dir,\n            f\"deeplabv3_final_results_{self.config.experiment.run_name}.json\"\n        )\n        \n        os.makedirs(os.path.dirname(results_file), exist_ok=True)\n        with open(results_file, 'w') as f:\n            json.dump(results_data, f, indent=2)\n        \n        logger.info(f\"Đã lưu kết quả cuối cùng: {results_file}\")\n    \n    def save_training_plots(self, save_dir: str):\n        \"\"\"\n        Lưu biểu đồ training history\n        \n        Args:\n            save_dir: Thư mục lưu plots\n        \"\"\"\n        os.makedirs(save_dir, exist_ok=True)\n        \n        # Prepare data for plotting\n        train_metrics = {\n            'iou': self.history['train_iou'],\n            'dice': self.history['train_dice']\n        }\n        \n        val_metrics = {\n            'iou': self.history['val_iou'],\n            'dice': self.history['val_dice']\n        }\n        \n        # Plot training history\n        plot_path = os.path.join(save_dir, f\"deeplabv3_training_history_{self.config.experiment.run_name}.png\")\n        plot_training_history(\n            self.history['train_loss'],\n            self.history['val_loss'],\n            train_metrics,\n            val_metrics,\n            save_path=plot_path\n        )\n        \n        # Plot learning rate\n        import matplotlib.pyplot as plt\n        \n        plt.figure(figsize=(10, 6))\n        plt.plot(self.history['learning_rates'])\n        plt.title('Learning Rate Schedule (Polynomial Decay)')\n        plt.xlabel('Epoch')\n        plt.ylabel('Learning Rate')\n        plt.grid(True)\n        \n        lr_plot_path = os.path.join(save_dir, f\"deeplabv3_lr_schedule_{self.config.experiment.run_name}.png\")\n        plt.savefig(lr_plot_path, dpi=300, bbox_inches='tight')\n        plt.close()\n        \n        logger.info(f\"Đã lưu training plots: {save_dir}\")\n\ndef train_deeplabv3_model(\n    config,\n    train_image_dir: str,\n    train_mask_dir: str\n) -> Tuple[DeepLabV3, Dict[str, List[float]]]:\n    \"\"\"\n    Function chính để train DeepLabV3 model\n    \n    Args:\n        config: Configuration object\n        train_image_dir: Thư mục ảnh training\n        train_mask_dir: Thư mục mask training\n        \n    Returns:\n        Tuple (trained_model, training_history)\n    \"\"\"\n    # Lấy đường dẫn file\n    image_paths, mask_paths = get_file_paths(train_image_dir, train_mask_dir)\n    \n    # Chia dữ liệu\n    train_images, train_masks, val_images, val_masks, _, _ = create_data_splits(\n        image_paths,\n        mask_paths,\n        train_split=0.8,\n        val_split=0.2,\n        seed=42\n    )\n    \n    # Tạo processor\n    if TRANSFORMERS_AVAILABLE:\n        processor = SegformerImageProcessor(reduce_labels=False)\n    else:\n        processor = None\n    \n    # Tạo dataloaders\n    train_loader, val_loader = create_dataloaders(\n        train_images,\n        train_masks,\n        val_images,\n        val_masks,\n        processor,\n        batch_size=4,  # Giảm batch size để train lâu hơn\n        num_workers=4,\n        use_augmentation=True\n    )\n    \n    # Tạo trainer\n    trainer = DeepLabV3Trainer(config)\n    \n    # Training\n    history = trainer.train(train_loader, val_loader)\n    \n    # Lưu training plots\n    trainer.save_training_plots(config.experiment.log_dir)\n    \n    return trainer.model, history\n\n# Example usage\nif __name__ == \"__main__\":\n    # Tạo config cho DeepLabV3\n    config = Config()\n    config.model.model_name = \"deeplabv3\"\n    config.experiment.run_name = \"deeplabv3_experiment\"\n    \n    # Đường dẫn data\n    train_image_dir = \"/kaggle/input/bkai-igh-neopolyp/train/train\"\n    train_mask_dir = \"/kaggle/input/bkai-igh-neopolyp/train_gt/train_gt\"\n    \n    # Train model\n    model, history = train_deeplabv3_model(\n        config,\n        train_image_dir,\n        train_mask_dir\n    )\n    \n    logger.info(\"DeepLabV3 training hoàn thành!\") ","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-07-21T07:59:45.149727Z","iopub.execute_input":"2025-07-21T07:59:45.150233Z","iopub.status.idle":"2025-07-21T08:05:46.129705Z","shell.execute_reply.started":"2025-07-21T07:59:45.150206Z","shell.execute_reply":"2025-07-21T08:05:46.128536Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"### file test deeplabv3\nimport os\nimport numpy as np\nimport torch\nimport torch.nn as nn\nimport torch.nn.functional as F\nimport cv2\nimport matplotlib.pyplot as plt\nfrom PIL import Image\nfrom typing import List, Tuple, Optional, Dict, Any\nimport logging\nfrom tqdm import tqdm\n\n\nclass DoubleConv(nn.Module):\n    \"\"\"Double Convolution block\"\"\"\n    def __init__(self, in_channels: int, out_channels: int):\n        super().__init__()\n        self.double_conv = nn.Sequential(\n            nn.Conv2d(in_channels, out_channels, kernel_size=3, padding=1),\n            nn.BatchNorm2d(out_channels),\n            nn.ReLU(inplace=True),\n            nn.Conv2d(out_channels, out_channels, kernel_size=3, padding=1),\n            nn.BatchNorm2d(out_channels),\n            nn.ReLU(inplace=True)\n        )\n    \n    def forward(self, x: torch.Tensor) -> torch.Tensor:\n        return self.double_conv(x)\n\nclass ASPPModule(nn.Module):\n    \"\"\"ASPP Module cho DeepLabV3\"\"\"\n    def __init__(self, in_channels: int, out_channels: int, atrous_rates: List[int]):\n        super().__init__()\n        \n        modules = []\n        # 1x1 convolution\n        modules.append(nn.Sequential(\n            nn.Conv2d(in_channels, out_channels, 1, bias=False),\n            nn.BatchNorm2d(out_channels),\n            nn.ReLU(inplace=True)\n        ))\n        \n        # Atrous convolutions\n        for rate in atrous_rates:\n            modules.append(nn.Sequential(\n                nn.Conv2d(in_channels, out_channels, 3, padding=rate, dilation=rate, bias=False),\n                nn.BatchNorm2d(out_channels),\n                nn.ReLU(inplace=True)\n            ))\n        \n        # Global average pooling\n        modules.append(nn.Sequential(\n            nn.AdaptiveAvgPool2d(1),\n            nn.Conv2d(in_channels, out_channels, 1, bias=False),\n            nn.BatchNorm2d(out_channels),\n            nn.ReLU(inplace=True)\n        ))\n        \n        self.convs = nn.ModuleList(modules)\n        \n        # Project layer\n        self.project = nn.Sequential(\n            nn.Conv2d(len(self.convs) * out_channels, out_channels, 1, bias=False),\n            nn.BatchNorm2d(out_channels),\n            nn.ReLU(inplace=True),\n            nn.Dropout(0.5)\n        )\n    \n    def forward(self, x: torch.Tensor) -> torch.Tensor:\n        res = []\n        for conv in self.convs:\n            res.append(conv(x))\n        \n        # Upsample global pooling result\n        res[-1] = F.interpolate(res[-1], size=x.shape[-2:], mode='bilinear', align_corners=False)\n        \n        res = torch.cat(res, dim=1)\n        return self.project(res)\n\nclass DeepLabV3Fixed(nn.Module):\n    \"\"\"DeepLabV3 model được fix để match với checkpoint đã train\"\"\"\n    \n    def __init__(self, num_classes: int = 3, backbone: str = \"resnet50\"):\n        super().__init__()\n        \n        self.num_classes = num_classes\n        self.backbone = backbone\n        \n        # Backbone layers - match với checkpoint\n        self.backbone_layers = nn.Sequential(\n            nn.Conv2d(3, 64, kernel_size=7, stride=2, padding=3, bias=False),\n            nn.BatchNorm2d(64),\n            nn.ReLU(inplace=True),\n            nn.MaxPool2d(kernel_size=3, stride=2, padding=1),\n            \n            # Residual blocks (simplified) - match với checkpoint\n            DoubleConv(64, 128),\n            nn.MaxPool2d(2),\n            DoubleConv(128, 256),\n            nn.MaxPool2d(2),\n            DoubleConv(256, 512),\n        )\n        \n        # ASPP module - match với checkpoint structure\n        self.aspp = ASPPModule(512, 256, [6, 12, 18])\n        \n        # Classifier - match với checkpoint (có nhiều layers)\n        self.classifier = nn.Sequential(\n            nn.Conv2d(256, 256, 3, padding=1, bias=False),  # layer 0\n            nn.BatchNorm2d(256),                           # layer 1\n            nn.ReLU(inplace=True),                         # layer 2\n            nn.Dropout(0.5),                               # layer 3\n            nn.Conv2d(256, num_classes, 1)                 # layer 4\n        )\n        \n        print(f\"✅ Tạo DeepLabV3Fixed với {num_classes} classes\")\n    \n    def forward(self, x: torch.Tensor) -> torch.Tensor:\n        input_shape = x.shape[-2:]\n        \n        # Backbone\n        x = self.backbone_layers(x)\n        \n        # ASPP\n        x = self.aspp(x)\n        \n        # Classifier\n        x = self.classifier(x)\n        \n        # Upsample to original size\n        x = F.interpolate(x, size=input_shape, mode='bilinear', align_corners=False)\n        \n        return x\n\nclass DeepLabV3Tester:\n    \"\"\"Test DeepLabV3 model với visualization\"\"\"\n    \n    def __init__(self, model_path: str, config=None):\n        self.model_path = model_path\n        self.config = config or Config()\n        self.device = get_device()\n        \n        # Định nghĩa màu sắc\n        self.colors = {\n            0: [255, 0, 0],      # Đỏ cho neoplastic  \n            1: [255, 165, 0],    # Cam cho non-neoplastic\n            2: [0, 0, 0]         # Đen cho background\n        }\n        \n        self.highlight_colors = {\n            0: [255, 80, 80],    # Đỏ sáng cho neoplastic\n            1: [255, 180, 80],   # Cam sáng cho non-neoplastic\n        }\n        \n        self.aspp_rates = [6, 12, 18]\n        \n        # Load model - Sử dụng DeepLabV3Fixed\n        self.model = self._load_deeplabv3_model()\n        self.model.eval()\n        \n        print(\"✅ Đã khởi tạo DeepLabV3Tester\")\n    \n    def _load_deeplabv3_model(self):\n        \"\"\"Load model với architecture đúng\"\"\"\n        # Tạo model architecture - Sử dụng DeepLabV3Fixed\n        model = DeepLabV3Fixed(\n            num_classes=self.config.model.num_classes,\n            backbone=\"resnet50\"\n        )\n        \n        # Load checkpoint với weights_only=False\n        if os.path.exists(self.model_path):\n            checkpoint = torch.load(self.model_path, map_location=self.device, weights_only=False)\n            \n            if 'model_state_dict' in checkpoint:\n                model.load_state_dict(checkpoint['model_state_dict'])\n                print(f\"✅ Loaded DeepLabV3 từ checkpoint: {self.model_path}\")\n            else:\n                model.load_state_dict(checkpoint)\n                print(f\"✅ Loaded DeepLabV3 weights: {self.model_path}\")\n        else:\n            raise FileNotFoundError(f\"❌ Không tìm thấy model: {self.model_path}\")\n        \n        model.to(self.device)\n        return model\n    \n    def predict_single_image(self, image_path: str):\n        \"\"\"Predict segmentation cho một ảnh\"\"\"\n        # Load và preprocess ảnh\n        image = Image.open(image_path).convert('RGB')\n        original_size = image.size\n        original_image = np.array(image)\n        \n        # Resize về 512x512\n        image_resized = image.resize((512, 512))\n        \n        # Convert to tensor và normalize\n        from torchvision import transforms\n        transform = transforms.Compose([\n            transforms.ToTensor(),\n            transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225])\n        ])\n        \n        # Transform trả về tensor, có thể gọi .unsqueeze()\n        image_tensor = transform(image_resized).unsqueeze(0).to(self.device)\n        \n        # Inference\n        with torch.no_grad():\n            logits = self.model(image_tensor)\n            \n            # Upsample về kích thước gốc\n            logits = torch.nn.functional.interpolate(\n                logits,\n                size=original_size[::-1],  # (height, width)\n                mode='bilinear',\n                align_corners=False\n            )\n            \n            # Get predictions\n            predictions = torch.argmax(logits, dim=1)\n            prediction = predictions.cpu().numpy()[0]\n        \n        return prediction, original_image\n    \n    def create_highlighted_visualization(self, image, prediction, alpha=0.6):\n        \"\"\"Tạo visualization highlight khối u\"\"\"\n        overlay = np.zeros_like(image)\n        \n        # Highlight neoplastic (có polyp)\n        neoplastic_mask = (prediction == 0)\n        if np.any(neoplastic_mask):\n            overlay[neoplastic_mask] = self.highlight_colors[0]\n        \n        # Highlight non-neoplastic \n        non_neoplastic_mask = (prediction == 1)\n        if np.any(non_neoplastic_mask):\n            overlay[non_neoplastic_mask] = self.highlight_colors[1]\n        \n        # Tạo ảnh kết hợp\n        highlighted_image = cv2.addWeighted(image, 1-alpha, overlay, alpha, 0)\n        \n        return highlighted_image\n    \n    def test_single_image(self, image_path: str):\n        \"\"\"Test một ảnh và hiển thị kết quả\"\"\"\n        print(f\"🔍 Testing: {os.path.basename(image_path)}\")\n        \n        # Predict\n        prediction, original_image = self.predict_single_image(image_path)\n        \n        # Tạo visualization\n        highlighted = self.create_highlighted_visualization(original_image, prediction)\n        \n        # Hiển thị kết quả\n        fig, axes = plt.subplots(1, 3, figsize=(15, 5))\n        \n        axes[0].imshow(original_image)\n        axes[0].set_title('Ảnh gốc')\n        axes[0].axis('off')\n        \n        # Colored mask\n        colored_mask = np.zeros((*prediction.shape, 3), dtype=np.uint8)\n        for class_id, color in self.colors.items():\n            colored_mask[prediction == class_id] = color\n        \n        axes[1].imshow(colored_mask)\n        axes[1].set_title('Prediction Mask')\n        axes[1].axis('off')\n        \n        axes[2].imshow(highlighted)\n        axes[2].set_title('Highlighted Result')\n        axes[2].axis('off')\n        \n        plt.tight_layout()\n        plt.show()\n        \n        # In thống kê\n        polyp_pixels = np.sum((prediction == 0) | (prediction == 1))\n        total_pixels = prediction.size\n        \n        print(f\"🔴 Neoplastic (Có polyp): {np.sum(prediction == 0)} pixels\")\n        print(f\"🟠 Non-neoplastic: {np.sum(prediction == 1)} pixels\") \n        print(f\"⚫ Background: {np.sum(prediction == 2)} pixels\")\n        print(f\"📈 Tỷ lệ polyp: {(polyp_pixels/total_pixels)*100:.2f}%\")\n        print(\"-\" * 50)\n\n# Sử dụng trực tiếp - không cần import\ndef run_deeplabv3_test_simple(model_path: str, test_image_paths: List[str]):\n    \"\"\"Function đơn giản để test DeepLabV3\"\"\"\n    print(\"🔬 BẮT ĐẦU TEST DEEPLABV3 MODEL\")\n    print(\"=\" * 50)\n    \n    # Tạo tester - sử dụng classes đã định nghĩa trong notebook\n    tester = DeepLabV3Tester(model_path)\n    \n    # Test từng ảnh\n    for i, img_path in enumerate(test_image_paths[:5]):\n        print(f\"\\n📸 Testing ảnh {i+1}/5\")\n        try:\n            tester.test_single_image(img_path)\n        except Exception as e:\n            print(f\"❌ Lỗi: {e}\")\n    \n    print(\"\\n🎉 HOÀN THÀNH TEST!\")\n\n# =================== CHẠY TEST TRỰC TIẾP ===================\n\n# Đường dẫn model (best model đã train) - Thay đổi theo đường dẫn thực tế của bạn\nmodel_path_options = [\n    \"/kaggle/working/models/best_deeplabv3_deeplabv3_experiment.pth\",\n]\n\n# Tìm đường dẫn model tồn tại\nmodel_path = None\nfor path in model_path_options:\n    if os.path.exists(path):\n        model_path = path\n        print(f\"✅ Tìm thấy model: {model_path}\")\n        break\n\nif model_path is None:\n    print(\"❌ Không tìm thấy model file. Hãy kiểm tra đường dẫn:\")\n    for path in model_path_options:\n        print(f\"   - {path}\")\n    # Fallback: sử dụng đường dẫn đầu tiên (có thể cần sửa)\n    model_path = model_path_options[0]\n    print(f\"⚠️ Sử dụng đường dẫn mặc định: {model_path}\")\n\n# Đường dẫn ảnh test - Thay đổi theo đường dẫn thực tế của bạn  \ntest_image_dir_options = [\n    \"/kaggle/input/deeplabv3-test/datatest\",\n]\n\n# Tìm thư mục test images tồn tại\ntest_image_dir = None\nfor dir_path in test_image_dir_options:\n    if os.path.exists(dir_path) and os.path.isdir(dir_path):\n        test_image_dir = dir_path\n        print(f\"✅ Tìm thấy thư mục test: {test_image_dir}\")\n        break\n\nif test_image_dir is None:\n    print(\"❌ Không tìm thấy thư mục test images. Hãy kiểm tra đường dẫn:\")\n    for dir_path in test_image_dir_options:\n        print(f\"   - {dir_path}\")\n    # Fallback: sử dụng đường dẫn đầu tiên  \n    test_image_dir = test_image_dir_options[0]\n    print(f\"⚠️ Sử dụng đường dẫn mặc định: {test_image_dir}\")\n\n# Lấy danh sách ảnh test\ntry:\n    test_images = [\n        os.path.join(test_image_dir, f) \n        for f in os.listdir(test_image_dir) \n        if f.lower().endswith(('.png', '.jpg', '.jpeg', '.bmp', '.tiff'))\n    ]\n    print(f\"📁 Tìm thấy {len(test_images)} ảnh test\")\n    \n    if len(test_images) == 0:\n        print(\"❌ Không có ảnh nào trong thư mục test\")\n    else:\n        print(\"📋 Danh sách 5 ảnh đầu tiên:\")\n        for i, img in enumerate(test_images[:5]):\n            print(f\"   {i+1}. {os.path.basename(img)}\")\n            \nexcept Exception as e:\n    print(f\"❌ Lỗi khi đọc thư mục test: {e}\")\n    test_images = []\n\n# CHẠY TEST\nif len(test_images) > 0:\n    print(\"\\n\" + \"=\"*60)\n    print(\"🚀 BẮT ĐẦU TEST DEEPLABV3 MODEL\")\n    print(\"=\"*60)\n    \n    try:\n        run_deeplabv3_test_simple(model_path, test_images)\n    except Exception as e:\n        print(f\"❌ Lỗi khi chạy test: {e}\")\n        print(\"💡 Hãy kiểm tra:\")\n        print(\"   1. Model file có tồn tại không?\")\n        print(\"   2. Thư mục test images có tồn tại không?\") \n        print(\"   3. Config và DeepLabV3 đã được định nghĩa trong notebook chưa?\")\nelse:\n    print(\"❌ Không thể chạy test - không có ảnh test\")","metadata":{"trusted":true},"outputs":[],"execution_count":null}]}